我想创建一个 Source 并稍后在其上推送元素,例如:
I would like to create a Source and later push elements on it, like in:
val src = ... // create the Source here // and then, do something like this pushElement(x1, src) pushElement(x2, src)推荐的方法是什么?
谢谢!
推荐答案可以通过三种方式实现:
There are three ways this can be achieved:
1.使用 SourceQueue 后期实现
您可以使用 Source.queue 将 Flow 实体化为 SourceQueue:
You can use Source.queue that materializes the Flow into a SourceQueue:
case class Weather(zipCode : String, temperature : Double, raining : Boolean) val bufferSize = 100 //if the buffer fills up then this strategy drops the oldest elements //upon the arrival of a new element. val overflowStrategy = akka.stream.OverflowStrategy.dropHead val queue = Source.queue(bufferSize, overflowStrategy) .filter(!_.raining) .to(Sink foreach println) .run() // in order to "keep" the queue Materialized value instead of the Sink's queue offer Weather("02139", 32.0, true)2.使用 Actor 进行后期实体化
有一个类似的问答此处,要点是将流具体化为 ActorRef 并向该 ref 发送消息:
There is a similar question and answer here, the gist being that you materialize the stream as an ActorRef and send messages to that ref:
val ref = Source.actorRef[Weather](Int.MaxValue, fail) .filter(!_.raining) .to(Sink foreach println ) .run() // in order to "keep" the ref Materialized value instead of the Sink's ref ! Weather("02139", 32.0, true)3.使用 Actor 预实现
同样,您可以显式创建一个包含消息缓冲区的 Actor,使用该 Actor 来创建一个源,然后按照答案中的描述发送该 Actor 消息 此处:
Similarly, you could explicitly create an Actor that contains a message buffer, use that Actor to create a Source, and then send that Actor messages as described in the answer here:
object WeatherForwarder { def props : Props = Props[WeatherForwarder] } //see provided link for example definition class WeatherForwarder extends Actor {...} val actorRef = actorSystem actorOf WeatherForwarder.props //note the stream has not been instatiated yet actorRef ! Weather("02139", 32.0, true) //stream already has 1 Weather value to process which is sitting in the //ActorRef's internal buffer val stream = Source(ActorPublisher[Weather](actorRef)).runWith{...}更多推荐
如何创建一个可以稍后通过方法调用接收元素的源?
发布评论