akka-streams with akka-cluster

Question

My akka-streams learn-o-thon continues. I'd like to integrate my akka-streams application with akka-cluster and DistributedPubSubMediator.

Adding support for Publish is fairly straight forward, but the Subscribe part I'm having trouble with.

For reference, a subscriber is given as follows in the 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...
  }
}

My question is, how should I integrate this actor with my flow, and how should I ensure I'm getting publish messages in the absence of stream backpressure?

I'm trying to accomplish a pubsub model where one stream may publish a message and another stream would consume it (if subscribed).

Answer 1

You probably want to make your Actor extend ActorPublisher. Then you can create a Source from it and integrate that into your stream.

See the docs on ActorPublisher here: http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-integrations.html

Answer 2

The other answers are outdated: they suggest using ActorPublisher, which has been deprecated since version 2.5.0.

For those interested in a current approach, Colin Breck wrote an excellent series in his blog about integrating Akka Streams and Akka actors. Over the course of the series, Breck fleshes out a system that begins with Akka Streams and plain actors, then incorporates Akka Cluster and Akka Persistence. The first post in the series is here (the distributed stream processing piece is in part 3).