akka.streams。您可以发出值的源（类似于monix.BehaviorSubject）

Question

[我正在寻找akka.stream.scaladsl.Source构造方法，该方法将允许我简单地从不同代码的位置发出下一个值（例如，监视系统事件）。

• 我需要类似于Promise的文字。 Promise向Future发出单个值。我需要向Source发出多个值。
• 例如monix.reactive.subjects.BehaviorSubject.onNext(_)
• 我不太在意背压。

当前，我已经使用monix和akka-streams实现了此功能（下面的代码），但我希望只有akka-streams解决方案：

import akka.stream.scaladsl.{Flow, Sink, Source}
import monix.reactive.subjects.BehaviorSubject
import monix.execution.Scheduler.Implicits.global

val bs = BehaviorSubject("") //monix subject is sink and source at the same time

//this is how it is currently implemented
def createSource() = { 
    val s1 = Source.fromPublisher(bs.toReactivePublisher) //here we create source from subject
    Flow.fromSinkAndSourceCoupled[String, String](Sink.ignore, s1)
}

//somewhere else in code... some event happened
//this is how it works in monix.
val res:Future[Ack] = bs.onNext("Event #1471 just happened!") //here we emit value

Answer 1

也许您正在寻找Actor Source

文档中的示例：

import akka.actor.typed.ActorRef
import akka.stream.OverflowStrategy
import akka.stream.scaladsl.{ Sink, Source }
import akka.stream.typed.scaladsl.ActorSource

trait Protocol
case class Message(msg: String) extends Protocol
case object Complete extends Protocol
case class Fail(ex: Exception) extends Protocol

val source: Source[Protocol, ActorRef[Protocol]] = ActorSource.actorRef[Protocol](completionMatcher = {
  case Complete =>
}, failureMatcher = {
  case Fail(ex) => ex
}, bufferSize = 8, overflowStrategy = OverflowStrategy.fail)

val ref = source
  .collect {
    case Message(msg) => msg
  }
  .to(Sink.foreach(println))
  .run()

ref ! Message("msg1")

这样，您将可以通过actor系统将消息发送给actor，这些消息将从ActorSource的下游发出。

Answer 2

我认为您正在寻找的是sink.foreach。它为收到的每个元素调用一个给定的过程。我认为代码将如下所示：

s1.runWith(Sink.foreach(write))

本质上，正在做的是，对于流源，接收器尝试写入该流的每个元素。

编辑

我认为您正在寻找maybe。它创建了一个源，一旦物化的Promise值完成，该源便会发射。请检查此documentation

编辑

[futureSource也可能起作用。一旦成功完成，它将流式处理给定未来来源的元素。

让我知道是否有帮助！

Answer 3

Source抽象，顾名思义，提供用于数据源的API。相反，您需要查看消耗数据的抽象-Sink。您最需要的是Sink.foreach操作，https://doc.akka.io/docs/akka/current/stream/operators/Sink/foreach.html

在这种情况下，代码看起来像：

import akka.stream.scaladsl.{Sink, Source}

val s1 = Source.// your WS akka stream source
s1.runWith(Sink.foreach(write))

希望这会有所帮助！