StreamRefs - Reactive Streams over the network

Stream references, or “stream refs” for short, allow running Akka Streams across multiple nodes within an Akka Cluster.

Unlike heavier “streaming data processing” frameworks, Akka Streams are not “deployed” nor automatically distributed. Akka stream refs are, as the name implies, references to existing parts of a stream, and can be used to create a distributed processing framework or introduce such capabilities in specific parts of your application.

Stream refs are trivial to make use of in existing clustered Akka applications, and require no additional configuration or setup. They automatically maintain flow-control / back-pressure over the network, and employ Akka’s failure detection mechanisms to fail-fast (“let it crash!”) in the case of failures of remote nodes. They can be seen as an implementation of the Work Pulling Pattern, which one would otherwise implement manually.

Stream References

The prime use case for stream refs is to replace raw actor or HTTP messaging between systems where a long running stream of data is expected between two entities. Often times, they can be used to effectively achieve point to point streaming without the need of setting up additional message brokers or similar secondary clusters.

Stream refs are well suited for any system in which you need to send messages between nodes and need to do so in a flow-controlled fashion. Typical examples include sending work requests to worker nodes, as fast as possible, but not faster than the worker node can process them, or sending data elements which the downstream may be slow at processing. It is recommended to mix and introduce stream refs in Actor messaging based systems, where the actor messaging is used to orchestrate and prepare such message flows, and later the stream refs are used to do the flow-controlled message transfer.

Stream refs are not persistent, however it is simple to build a resume-able stream by introducing such protocol on the actor messaging layer. Stream refs are absolutely expected to be sent over Akka remoting to other nodes within a cluster, and as such, complement and do not compete with plain Actor messaging. Actors would usually be used to establish the stream, by means of some initial message saying “I want to offer you many log elements (the stream ref)”, or alternatively in the opposite way “If you need to send me much data, here is the stream ref you can use to do so”.

Since the two sides (“local” and “remote”) of each reference may be confusing to simply refer to as “remote” and “local” – since either side can be seen as “local” or “remote” depending how we look at it – we propose to use the terminology “origin” and “target”, which is defined by where the stream ref was created. For SourceRefs, the “origin” is the side which has the data that it is going to stream out. For SinkRefs the “origin” side is the actor system that is ready to receive the data and has allocated the ref. Those two may be seen as duals of each other, however to explain patterns about sharing references, we found this wording to be rather useful.

Source Refs - offering streaming data to a remote system

A SourceRefSourceRef can be offered to a remote actor system in order for it to consume some source of data that we have prepared locally.

In order to share a Source with a remote endpoint you need to materialize it by running it into the Sink.sourceRef. That sink materializes the SourceRef that you can then send to other nodes. Please note that it materializes into a Future so you will have to use the pipeTo

Scala

import akka.stream.SourceRef
import akka.pattern.pipe

case class RequestLogs(streamId: Int)
case class LogsOffer(streamId: Int, sourceRef: SourceRef[String])

class DataSource extends Actor {
  import context.dispatcher
  implicit val mat = ActorMaterializer()(context)

  def receive = {
    case RequestLogs(streamId) ⇒
      // obtain the source you want to offer:
      val source: Source[String, NotUsed] = streamLogs(streamId)

      // materialize the SourceRef:
      val ref: Future[SourceRef[String]] = source.runWith(StreamRefs.sourceRef())

      // wrap the SourceRef in some domain message, such that the sender knows what source it is
      val reply: Future[LogsOffer] = ref.map(LogsOffer(streamId, _))

      // reply to sender
      reply pipeTo sender()
  }

  def streamLogs(streamId: Long): Source[String, NotUsed] = ???
}

Java

static class RequestLogs {
  public final long streamId;

  public RequestLogs(long streamId) {
    this.streamId = streamId;
  }
}

static class LogsOffer {
  final SourceRef<String> sourceRef;

  public LogsOffer(SourceRef<String> sourceRef) {
    this.sourceRef = sourceRef;
  }
}

static class DataSource extends AbstractActor {
  @Override
  public Receive createReceive() {
    return receiveBuilder()
        .match(RequestLogs.class, this::handleRequestLogs)
        .build();
  }

  private void handleRequestLogs(RequestLogs requestLogs) {
    Source<String, NotUsed> logs = streamLogs(requestLogs.streamId);
    CompletionStage<SourceRef<String>> logsRef = logs.runWith(StreamRefs.sourceRef(), mat);

    PatternsCS.pipe(logsRef.thenApply(ref -> new LogsOffer(ref)), context().dispatcher())
        .to(sender());
  }

  private Source<String, NotUsed> streamLogs(long streamId) {
    return Source.repeat("[INFO] some interesting logs here (for id: " + streamId + ")");
  }
}

The origin actor which creates and owns the Source could also perform some validation or additional setup when preparing the source. Once it has handed out the SourceRef the remote side can run it like this:

Scala

val sourceActor = system.actorOf(Props[DataSource], "dataSource")

sourceActor ! RequestLogs(1337)
val offer = expectMsgType[LogsOffer]

// implicitly converted to a Source:
offer.sourceRef.runWith(Sink.foreach(println))
// alternatively explicitly obtain Source from SourceRef:
// offer.sourceRef.source.runWith(Sink.foreach(println))

Java

ActorRef sourceActor = system.actorOf(Props.create(DataSource.class), "dataSource");

sourceActor.tell(new RequestLogs(1337), getTestActor());
LogsOffer offer = expectMsgClass(LogsOffer.class);

offer.sourceRef.getSource()
    .runWith(Sink.foreach(log -> System.out.println(log)), mat);

The process of preparing and running a SourceRef powered distributed stream is shown by the animation below:

Sink Refs - offering to receive streaming data from a remote system

The dual of SourceRefSourceRefs.

They can be used to offer the other side the capability to send to the origin side data in a streaming, flow-controlled fashion. The origin here allocates a Sink, which could be as simple as a Sink.foreach or as advanced as a complex sink which streams the incoming data into various other systems (e.g. any of the Alpakka provided Sinks).

Scala

import akka.pattern.pipe
import akka.stream.SinkRef

case class PrepareUpload(id: String)
case class MeasurementsSinkReady(id: String, sinkRef: SinkRef[String])

class DataReceiver extends Actor {

  import context.dispatcher
  implicit val mat = ActorMaterializer()(context)

  def receive = {
    case PrepareUpload(nodeId) ⇒
      // obtain the source you want to offer:
      val sink: Sink[String, NotUsed] = logsSinkFor(nodeId)

      // materialize the SinkRef (the remote is like a source of data for us):
      val ref: Future[SinkRef[String]] = StreamRefs.sinkRef[String]().to(sink).run()

      // wrap the SinkRef in some domain message, such that the sender knows what source it is
      val reply: Future[MeasurementsSinkReady] = ref.map(MeasurementsSinkReady(nodeId, _))

      // reply to sender
      reply pipeTo sender()
  }

  def logsSinkFor(nodeId: String): Sink[String, NotUsed] = ???
}

Java

static class PrepareUpload {
  final String id;

  public PrepareUpload(String id) {
    this.id = id;
  }
}
static class MeasurementsSinkReady {
  final String id;
  final SinkRef<String> sinkRef;

  public MeasurementsSinkReady(String id, SinkRef<String> ref) {
    this.id = id;
    this.sinkRef = ref;
  }
}

static class DataReceiver extends AbstractActor {
  @Override
  public Receive createReceive() {
    return receiveBuilder()
        .match(PrepareUpload.class, prepare -> {
          Sink<String, NotUsed> sink = logsSinkFor(prepare.id);
          CompletionStage<SinkRef<String>> sinkRef = StreamRefs.<String>sinkRef().to(sink).run(mat);

          PatternsCS.pipe(sinkRef.thenApply(ref -> new MeasurementsSinkReady(prepare.id, ref)), context().dispatcher())
              .to(sender());
        })
        .build();
  }

  private Sink<String, NotUsed> logsSinkFor(String id) {
    return Sink.<String>ignore().mapMaterializedValue(done -> NotUsed.getInstance());
  }
}

Using the offered SinkRef to send data to the origin of the Sink is also simple, as we can treat the SinkRef just as any other Sink and directly runWith or run with it.

Scala

val receiver = system.actorOf(Props[DataReceiver], "receiver")

receiver ! PrepareUpload("system-42-tmp")
val ready = expectMsgType[MeasurementsSinkReady]

// stream local metrics to Sink's origin:
localMetrics().runWith(ready.sinkRef)

Java

ActorRef receiver = system.actorOf(Props.create(DataReceiver.class), "dataReceiver");

receiver.tell(new PrepareUpload("system-42-tmp"), getTestActor());
MeasurementsSinkReady ready = expectMsgClass(MeasurementsSinkReady.class);

Source.repeat("hello")
    .runWith(ready.sinkRef.getSink(), mat);

The process of preparing and running a SinkRef powered distributed stream is shown by the animation below:

Bulk Stream References

Bulk stream refs can be used to create simple to use side-channels to transfer humongous amounts of data such as huge log files, messages or even media, with as much ease as if it was a trivial local stream.

Configuration

Stream reference subscription timeouts

All stream references have a subscription timeout, which is intended to prevent resource leaks in situations in which a remote node would requests the allocation of many streams yet never actually run them. In order to prevent this, each stream reference has a default timeout (of 30 seconds), after which the origin will abort the stream offer if the target has not materialized the stream ref in time. After the timeout has triggered, materialization of the target side will fail pointing out that the origin is missing.

Since these timeouts are often very different based on the kind of stream offered, and there can be many different kinds of them in the same application, it is possible to not only configure this setting globally (akka.stream.materializer.stream-ref.subscription-timeout), but also via attributes:

Scala

// configure the timeout for source
import scala.concurrent.duration._
import akka.stream.StreamRefAttributes

// configuring Sink.sourceRef (notice that we apply the attributes to the Sink!):
Source.repeat("hello")
  .runWith(StreamRefs.sourceRef().addAttributes(StreamRefAttributes.subscriptionTimeout(5.seconds)))

// configuring SinkRef.source:
StreamRefs.sinkRef().addAttributes(StreamRefAttributes.subscriptionTimeout(5.seconds))
  .runWith(Sink.ignore) // not very interesting Sink, just an example

Java

FiniteDuration timeout = FiniteDuration.create(5, TimeUnit.SECONDS);
Attributes timeoutAttributes = StreamRefAttributes.subscriptionTimeout(timeout);

// configuring Sink.sourceRef (notice that we apply the attributes to the Sink!):
Source.repeat("hello")
    .runWith(StreamRefs.<String>sourceRef().addAttributes(timeoutAttributes), mat);

// configuring SinkRef.source:
StreamRefs.<String>sinkRef().addAttributes(timeoutAttributes)
    .runWith(Sink.<String>ignore(), mat); // not very interesting sink, just an example

General configuration

Other settings can be set globally, in your application.conf, by overriding any of the following values in the akka.stream.materializer.stream-ref.* keyspace:

# configure defaults for SourceRef and SinkRef
stream-ref {
  # Buffer of a SinkRef that is used to batch Request elements from the other side of the stream ref
  #
  # The buffer will be attempted to be filled eagerly even while the local stage did not request elements,
  # because the delay of requesting over network boundaries is much higher.
  buffer-capacity = 32

  # Demand is signalled by sending a cumulative demand message ("requesting messages until the n-th sequence number)
  # Using a cumulative demand model allows us to re-deliver the demand message in case of message loss (which should
  # be very rare in any case, yet possible -- mostly under connection break-down and re-establishment).
  #
  # The semantics of handling and updating the demand however are in-line with what Reactive Streams dictates.
  #
  # In normal operation, demand is signalled in response to arriving elements, however if no new elements arrive
  # within `demand-redelivery-interval` a re-delivery of the demand will be triggered, assuming that it may have gotten lost.
  demand-redelivery-interval = 1 second

  # Subscription timeout, during which the "remote side" MUST subscribe (materialize) the handed out stream ref.
  # This timeout does not have to be very low in normal situations, since the remote side may also need to
  # prepare things before it is ready to materialize the reference. However the timeout is needed to avoid leaking
  # in-active streams which are never subscribed to.
  subscription-timeout = 30 seconds
}