scala – 如何通过方法调用创建可以接收元素的源代码？

1.使用SourceQueue进行后期实现

您可以使用将Flow实现到SourceQueue中的Source.queue：

case class Weather(zipCode : String,temperature : Double,raining : Boolean)

val bufferSize = 100

//if the buffer fills up this drops the oldest elements when a new element
//comes in
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，并发送消息给该参考：

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",true)

3.具有演员的物化

类似地，您可以显式创建一个包含消息缓冲区的Actor，使用该Actor创建一个源，然后按照答案here中所述发送该Actor消息：

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