Akka 스트림이 외부 세계에서 무한한 스트림의 현재 가치를 값을 집계 결코 완전한 정의에 의해 무한 스트림의 현재 값을 얻을 수있는 가장 좋은 방법은 무엇

Question

Source.repeat(1) 
    .scan(0)(_+_) 
    .to(Sink.ignore) 

전에서 조회하고자하는 Akka HTTP는 현재 카운터 값입니다. 동적 스트림을 사용해야합니까? broadcastHub와 Akka http에서 GET 요청에 대한 무한 스트림을 구독합니까?

Answer 1

한 가지 해결책은 필요한 상태를 유지하기 위해 액터를 사용하는 것입니다. Sink.actorRef은 싱크에 기존 액터 ref를 래핑합니다 (예 :

class Keeper extends Actor { 
    var i: Int = 0 

    override def receive: Receive = { 
    case n: Int ⇒ i = n 
    case Keeper.Get ⇒ sender ! i 
    } 
} 

object Keeper { 
    case object Get 
} 

val actorRef = system.actorOf(Props(classOf[Keeper])) 

val q = Source.repeat(1) 
    .scan(0)(_+_) 
    .runWith(Sink.actorRef(actorRef, PoisonPill)) 

val result = (actorRef ? Keeper.Get).mapTo[Int] 

Sink.actorRef을 사용하면 배압이 보존되지 않습니다. 이것은 Sink.actorRefWithAck을 사용하여 향상시킬 수 있습니다. 이에 대한 자세한 내용은 docs에서 확인할 수 있습니다.