scala – 从EventStream创建源代码

我正在使用PlayFramework 2.5.3,并希望从akka.event.EventStream创建一个akka.stream. scaladsl.source(事件流是actor系统的一部分).事件流将产生来自某种类型的事件,因此我需要订阅该特定类型的事件并使用play.api.mvc.Results.chunked来推送它们.有没有简单的方法使用Akka Streams 2.4.5创建这样的Source？

解决方法

您可以将 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.actorPublisher与延伸ActorPublisher[YourEventType]特性的actor一起使用,它将为您提供更多控制.但是,由于EventStream是一个纯粹的基于推送的源,因此使用ActorPublisher比使用Source.actorRef所做的更多,所以你也可以使用更简单的方法.