如何創建可以通過方法調用稍後接收元素的源？

Question

我想在其上創建Source，後來推元素，如在：

val src = ... // create the Source here 
// and then, do something like this 
pushElement(x1, src) 
pushElement(x2, src) 

什麼是推薦的方式做到這一點？

謝謝！

Answer 1

有三種方法可以做到這一點：

1.郵政物化與SourceQueue

您可以使用Source.queue是物化的流入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 。演員實現物化

有一個類似的問題和回答here，要點是，你兌現了流作爲ActorRef和郵件發送到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創建一個Source，然後發送該Actor消息，如回答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{...} 

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{...}