scala - 如何创建一个可以稍后通过方法调用接收元素的 Source？

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它将 Flow 物化为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：

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 创建一个 Source，然后按照此处答案中的描述发送该 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", 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{...}

Answer 2

由于 Akka 2.5Source有一个preMaterialize方法。

根据文档，这看起来像是按照您的要求做的指示：

在某些情况下，您需要一个具体Source化的值，然后才能Source连接到图表的其余部分。这在“物化价值驱动”来源的情况下特别有用，Source.queue例如Source.actorRef或Source.maybe。

下面是一个关于如何使用SourceQueue. 元素在具体化之前和之后被推送到队列中，以及从Flow:

import akka.actor.ActorSystem
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, OverflowStrategy}

implicit val system = ActorSystem("QuickStart")
implicit val materializer = ActorMaterializer()


val sourceDecl = Source.queue[String](bufferSize = 2, OverflowStrategy.backpressure)
val (sourceMat, source) = sourceDecl.preMaterialize()

// Adding element before actual materialization
sourceMat.offer("pre materialization element")

val flow = Flow[String].map { e =>
  if(!e.contains("new")) {
    // Adding elements from within the flow
    sourceMat.offer("new element generated inside the flow")
  }
  s"Processing $e"
}

// Actually materializing with `run`
source.via(flow).to(Sink.foreach(println)).run()

// Adding element after materialization
sourceMat.offer("post materialization element")

输出：

Processing pre materialization element
Processing post materialization element
Processing new element generated inside the flow
Processing new element generated inside the flow

Answer 3

在玩耍并寻找一个好的解决方案之后，我发现了这个解决方案，它干净、简单，并且在物化前和物化后都有效。 https://stackoverflow.com/a/32553913/6791842

  val (ref: ActorRef, publisher: Publisher[Int]) =
    Source.actorRef[Int](bufferSize = 1000, OverflowStrategy.fail)
      .toMat(Sink.asPublisher(true))(Keep.both).run()

  ref ! 1 //before

  val source = Source.fromPublisher(publisher)

  ref ! 2 //before
  Thread.sleep(1000)
  ref ! 3 //before

  source.runForeach(println)

  ref ! 4 //after
  Thread.sleep(1000)
  ref ! 5 //after

输出：

1
2
3
4
5