scala - 访问由 Source.actorRef 创建的 akka 流源的底层 ActorRef

Question

我正在尝试使用 Source.actorRef创建 akka.stream.scaladsl.Source 的方法目的。某种形式

import akka.stream.OverflowStrategy.fail
import akka.stream.scaladsl.Source

case class Weather(zip : String, temp : Double, raining : Boolean)

val weatherSource = Source.actorRef[Weather](Int.MaxValue, fail)

val sunnySource = weatherSource.filter(!_.raining)
...

我的问题是:如何将数据发送到基于 ActorRef 的源对象 ?

我认为向源发送消息是某种形式

//does not compile
weatherSource ! Weather("90210", 72.0, false)
weatherSource ! Weather("02139", 32.0, true)

但是weatherSource没有 !运算符(operator)或 tell方法。

documentation关于如何使用 Source.actorRef 并没有太多描述，它只是说你可以......

预先感谢您的评论和回复。

Answer 1

您需要一个 Flow :

  import akka.stream.OverflowStrategy.fail
  import akka.stream.scaladsl.Source
  import akka.stream.scaladsl.{Sink, Flow}

  case class Weather(zip : String, temp : Double, raining : Boolean)

  val weatherSource = Source.actorRef[Weather](Int.MaxValue, fail)

  val sunnySource = weatherSource.filter(!_.raining)

  val ref = Flow[Weather]
    .to(Sink.ignore)
    .runWith(sunnySource)

  ref ! Weather("02139", 32.0, true)

请记住，这都是实验性的，可能会改变!