scala - 带有 akka-cluster 的 akka-streams

Question

我的 akka-streams 学习马拉松继续。我想将我的 akka-streams 应用程序与 akka-cluster and DistributedPubSubMediator 集成。

添加对 Publish 的支持相当简单，但我在 Subscribe 部分遇到了麻烦。

作为引用，订阅者在 Typesafe sample 中给出如下:

class ChatClient(name: String) extends Actor {
  val mediator = DistributedPubSub(context.system).mediator
  mediator ! Subscribe("some topic", self)

  def receive = {
    case ChatClient.Message(from, text) =>
      ...process message...
  }
}

我的问题是，我应该如何将此参与者与我的流集成，以及我应该如何确保在没有流背压的情况下获得发布消息？

我正在尝试实现一个 pubsub 模型，其中一个流可以发布一条消息，另一个流将使用它(如果已订阅)。

Answer 1

您可能想让您的 Actor 扩展 ActorPublisher。然后您可以从中创建一个源并将其集成到您的流中。

请在此处查看 ActorPublisher 上的文档:http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-integrations.html