我正在使用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所做的更多,所以你也可以使用更简单的方法.