akka 可以发出值的源代码(类似于monix.BehaviorSubject)

gtlvzcf8 于 2022-11-06 发布在其他

关注(0)|答案(4)|浏览(163)

我正在寻找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

akka

来源：https://stackoverflow.com/questions/60544689/akka-streams-source-that-you-can-emit-values-similar-to-monix-behaviorsubject

4条答案

按热度按时间

vsikbqxv1#

也许你正在寻找演员来源
文档中的示例：

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向下发送到流中。

赞(0）回复(0）举报 2022-11-06

4dc9hkyq2#

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))

希望这对你有帮助！

赞(0）回复(0）举报 2022-11-06

lmvvr0a83#

我想你要找的是sink.foreach。它为每个接收到的元素调用一个给定的过程。我想代码看起来像这样：

s1.runWith(Sink.foreach(write))

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

编辑

我认为你正在寻找maybe。它创建一个源，一旦实现的承诺完成一个值发射。看看这个documentation

编辑

futureSource也可以工作，它在成功完成后，将给定的未来源的元素流化。
如果有帮助的话，让我知道！！

赞(0）回复(0）举报 2022-11-06

bnl4lu3b4#

https://doc.akka.io/docs/akka/current/stream/operators/Source/fromIterator.html或https://doc.akka.io/docs/akka/current/stream/operators/Source/fromPublisher.html是您需要的，这取决于Source从何处使用数据。

赞(0）回复(0）举报 2022-11-06

我来回答

akka 可以发出值的源代码(类似于monix.BehaviorSubject)

4条答案

相关问题

热门标签

最新问答