Reactive Streams Interop

Dependency

To use Akka Streams, add the module to your project:

sbt
val AkkaVersion = "2.6.15"
libraryDependencies += "com.typesafe.akka" %% "akka-stream" % AkkaVersion
Maven
<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.typesafe.akka</groupId>
      <artifactId>akka-bom_${scala.binary.version}</artifactId>
      <version>2.6.15</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
<dependencies>
  <dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-stream_${scala.binary.version}</artifactId>
  </dependency>
</dependencies>
Gradle
def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation platform("com.typesafe.akka:akka-bom_${versions.ScalaBinary}:2.6.15")

  implementation "com.typesafe.akka:akka-stream_${versions.ScalaBinary}"
}

Overview

Akka Streams implements the Reactive Streams standard for asynchronous stream processing with non-blocking back pressure.

Since Java 9 the APIs of Reactive Streams has been included in the Java Standard library, under the java.util.concurrent.Flow namespace. For Java 8 there is instead a separate Reactive Streams artifact with the same APIs in the package org.reactivestreams.

Akka streams provides interoperability for both these two API versions, the Reactive Streams interfaces directly through factories on the regular Source and Sink APIs. For the Java 9 and later built in interfaces there is a separate set of factories in akka.stream.scaladsl.JavaFlowSupportakka.stream.javadsl.JavaFlowSupport.

In the following samples the standalone Reactive Stream API factories has been used but each such call can be replaced with the corresponding method from JavaFlowSupport and the JDK java.util.concurrent.Flow._java.util.concurrent.Flow.* interfaces.

Note that it is not possible to use JavaFlowSupport on Java 8 since the needed interfaces simply is not available in the Java standard library.

The two most important interfaces in Reactive Streams are the Publisher and Subscriber.

Scala
sourceimport org.reactivestreams.Publisher
import org.reactivestreams.Subscriber
import org.reactivestreams.Processor
Java
sourceimport org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Processor;

Let us assume that a library provides a publisher of tweets:

Scala
sourcedef tweets: Publisher[Tweet]
Java
sourcePublisher<Tweet> tweets();

and another library knows how to store author handles in a database:

Scala
sourcedef storage: Subscriber[Author]
Java
sourceSubscriber<Author> storage();

Using an Akka Streams Flow we can transform the stream and connect those:

Scala
sourceval authors = Flow[Tweet].filter(_.hashtags.contains(akkaTag)).map(_.author)

Source.fromPublisher(tweets).via(authors).to(Sink.fromSubscriber(storage)).run()
Java
sourcefinal Flow<Tweet, Author, NotUsed> authors =
    Flow.of(Tweet.class).filter(t -> t.hashtags().contains(AKKA)).map(t -> t.author);

Source.fromPublisher(rs.tweets()).via(authors).to(Sink.fromSubscriber(rs.storage()));

The Publisher is used as an input Source to the flow and the Subscriber is used as an output Sink.

A Flow can also be also converted to a RunnableGraph[Processor[In, Out]] which materializes to a Processor when run() is called. run() itself can be called multiple times, resulting in a new Processor instance each time.

Scala
sourceval processor: Processor[Tweet, Author] = authors.toProcessor.run()

tweets.subscribe(processor)
processor.subscribe(storage)
Java
sourcefinal Processor<Tweet, Author> processor = authors.toProcessor().run(system);

rs.tweets().subscribe(processor);
processor.subscribe(rs.storage());

A publisher can be connected to a subscriber with the subscribe method.

It is also possible to expose a Source as a Publisher by using the Publisher-Sink:

Scala
sourceval authorPublisher: Publisher[Author] =
  Source.fromPublisher(tweets).via(authors).runWith(Sink.asPublisher(fanout = false))

authorPublisher.subscribe(storage)
Java
sourcefinal Publisher<Author> authorPublisher =
    Source.fromPublisher(rs.tweets())
        .via(authors)
        .runWith(Sink.asPublisher(AsPublisher.WITHOUT_FANOUT), system);

authorPublisher.subscribe(rs.storage());

A publisher that is created with Sink.asPublisher(fanout = false)Sink.asPublisher(AsPublisher.WITHOUT_FANOUT) supports only a single subscription. Additional subscription attempts will be rejected with an IllegalStateException.

A publisher that supports multiple subscribers using fan-out/broadcasting is created as follows:

Scala
sourcedef alert: Subscriber[Author]
def storage: Subscriber[Author]
Java
sourceSubscriber<Author> alert();
Subscriber<Author> storage();
Scala
sourceval authorPublisher: Publisher[Author] =
  Source.fromPublisher(tweets).via(authors).runWith(Sink.asPublisher(fanout = true))

authorPublisher.subscribe(storage)
authorPublisher.subscribe(alert)
Java
sourcefinal Publisher<Author> authorPublisher =
    Source.fromPublisher(rs.tweets())
        .via(authors)
        .runWith(Sink.asPublisher(AsPublisher.WITH_FANOUT), system);

authorPublisher.subscribe(rs.storage());
authorPublisher.subscribe(rs.alert());

The input buffer size of the operator controls how far apart the slowest subscriber can be from the fastest subscriber before slowing down the stream.

To make the picture complete, it is also possible to expose a Sink as a Subscriber by using the Subscriber-Source:

Scala
sourceval tweetSubscriber: Subscriber[Tweet] =
  authors.to(Sink.fromSubscriber(storage)).runWith(Source.asSubscriber[Tweet])

tweets.subscribe(tweetSubscriber)
Java
sourcefinal Subscriber<Author> storage = rs.storage();

final Subscriber<Tweet> tweetSubscriber =
    authors.to(Sink.fromSubscriber(storage)).runWith(Source.asSubscriber(), system);

rs.tweets().subscribe(tweetSubscriber);

It is also possible to use re-wrap Processor instances as a Flow by passing a factory function that will create the Processor instances:

Scala
source// An example Processor factory
def createProcessor: Processor[Int, Int] = Flow[Int].toProcessor.run()

val flow: Flow[Int, Int, NotUsed] = Flow.fromProcessor(() => createProcessor)
Java
source// An example Processor factory
final Creator<Processor<Integer, Integer>> factory =
    new Creator<Processor<Integer, Integer>>() {
      public Processor<Integer, Integer> create() {
        return Flow.of(Integer.class).toProcessor().run(system);
      }
    };

final Flow<Integer, Integer, NotUsed> flow = Flow.fromProcessor(factory);

Please note that a factory is necessary to achieve reusability of the resulting Flow.

Other implementations

Implementing Reactive Streams makes it possible to plug Akka Streams together with other stream libraries that adhere to the standard. An incomplete list of other implementations:

Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.