scala - 从 EventStream 创建源

Question

我正在使用 PlayFramework 2.5.3 并想创建一个 akka.stream.scaladsl.Source来自 akka.event.EventStream (事件流是参与者系统的一部分)。事件流会产生某种类型的事件，所以我需要订阅某种类型的事件并使用 play.api.mvc.Results.chunked 推送它们。 .有什么简单的方法可以创建这样的 Source使用 Akka Streams 2.4.5？

Answer 1

您可以使用 Source.actorRef 连同订阅。 Source.actorRef是实现为 ActorRef 的源，所以你可以这样做:

// choose the buffer size of the actor source and how the actor
// will react to its overflow
val eventListenerSource = Source.actorRef[YourEventType](32, OverflowStrategy.dropHead)

// run the stream and obtain all materialized values
val (eventListener, ...) = eventListenerSource
    .viaMat(...)(Keep.left)
    <...>
    .run()

// subscribe the source actor to the stream
actorSystem.eventStream.subscribe(eventListener, classOf[YourEventType])

// now events emitted by the source will go to the actor
// and through it to the stream

请注意 actorRef source 有一定的局限性，比如它的内部缓冲区自然不支持背压溢出策略。您可能想使用 Source.actorPublisher 与一个 Actor 延伸 ActorPublisher[YourEventType] trait，它会给你更多的控制。但是，由于 EventStream是一个纯粹的基于推送的源，您将无法使用 ActorPublisher 做更多事情比 Source.actorRef ，所以你也可以使用更简单的方法。