Page History

package com.psmon.cachedb.actortest;

import org.springframework.stereotype.Component;
import akka.*;
import akka.stream.*;
import akka.stream.javadsl.*;
import java.util.Arrays;
import java.util.concurrent.CompletionStage;

/*
  ActorStream class
 */
@Component
public class ActorStream extends ActorBase{
  public void runAll() {
    runBasic1();          
  }
  
  protected void runBasic1() {
    // Stream을 실행시킬수 있는곳 ( 액터시스템이 가지고 있음 )
    final Materializer materializer = ActorMaterializer.create(system);

    final Source<Integer, NotUsed> source =
    Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
    // note that the Future is scala.concurrent.Future

    final Sink<Integer, CompletionStage<Integer>> sink =
        Sink.<Integer, Integer> fold(0, (aggr, next) -> aggr + next);

    // connect the Source to the Sink, obtaining a RunnableFlow
    final RunnableGraph<CompletionStage<Integer>> runnable =
        source.toMat(sink, Keep.right());  //Left:읽은쪽을 유지 , Right:기록한쪽을 유지 . Both:둘다 유지 , None : 유지없음

    // materialize the flow
    final CompletionStage<Integer> sum = runnable.run(materializer);

    try {      
      log.info("==> {}",sum.toCompletableFuture().get());    
    } catch (Exception e) {
      //TODO: handle exception
    }
  }
}