Substreams

Dependency

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

sbt

libraryDependencies += "com.typesafe.akka" %% "akka-stream" % "2.5.32"

Maven

<dependencies>
  <dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-stream_2.12</artifactId>
    <version>2.5.32</version>
  </dependency>
</dependencies>

Gradle

dependencies {
  implementation "com.typesafe.akka:akka-stream_2.12:2.5.32"
}

Introduction

Substreams are represented as SubSource or SubFlow instances, on which you can multiplex a single Source or Flow into a stream of streams.

SubFlows cannot contribute to the super-flow’s materialized value since they are materialized later, during the runtime of the stream processing.

operators that create substreams are listed on Nesting and flattening operators

Nesting operators

groupBy

A typical operation that generates substreams is groupBy.

Scala

copysourceval source = Source(1 to 10).groupBy(3, _ % 3)

Java

copysourceSource.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).groupBy(3, elem -> elem % 3);

This operation splits the incoming stream into separate output streams, one for each element key. The key is computed for each element using the given function, which is f in the above diagram. When a new key is encountered for the first time a new substream is opened and subsequently fed with all elements belonging to that key. If allowClosedSubstreamRecreation is set to true a substream belonging to a specific key will be recreated if it was closed before, otherwise elements belonging to that key will be dropped.

If you add a Sink or Flow right after the groupBy operator, all transformations are applied to all encountered substreams in the same fashion. So, if you add the following Sink, that is added to each of the substreams as in the below diagram.

Scala

copysourceSource(1 to 10).groupBy(3, _ % 3).to(Sink.ignore).run()

Java

copysourceSource.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    .groupBy(3, elem -> elem % 3)
    .to(Sink.ignore())
    .run(mat);

Also substreams, more precisely, SubFlow and SubSource have methods that allow you to merge or concat substreams into the master stream again.

The mergeSubstreams method merges an unbounded number of substreams back to the master stream.

Scala

copysourceSource(1 to 10).groupBy(3, _ % 3).mergeSubstreams.runWith(Sink.ignore)

Java

copysourceSource.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    .groupBy(3, elem -> elem % 3)
    .mergeSubstreams()
    .runWith(Sink.ignore(), mat);

You can limit the number of active substreams running and being merged at a time, with either the mergeSubstreamsWithParallelism or concatSubstreams method.

Scala

copysourceSource(1 to 10).groupBy(3, _ % 3).mergeSubstreamsWithParallelism(2).runWith(Sink.ignore)

//concatSubstreams is equivalent to mergeSubstreamsWithParallelism(1)
Source(1 to 10).groupBy(3, _ % 3).concatSubstreams.runWith(Sink.ignore)

Java

copysourceSource.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    .groupBy(3, elem -> elem % 3)
    .mergeSubstreamsWithParallelism(2)
    .runWith(Sink.ignore(), mat);
// concatSubstreams is equivalent to mergeSubstreamsWithParallelism(1)
Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    .groupBy(3, elem -> elem % 3)
    .concatSubstreams()
    .runWith(Sink.ignore(), mat);

However, since the number of running (i.e. not yet completed) substreams is capped, be careful so that these methods do not cause deadlocks with back pressure like in the below diagram.

Element one and two leads to two created substreams, but since the number of substreams are capped to 2 when element 3 comes in it cannot lead to creation of a new substream until one of the previous two are completed and this leads to the stream being deadlocked.

splitWhen and splitAfter

splitWhen and splitAfter are two other operations which generate substreams.

The difference from groupBy is that, if the predicate for splitWhen and splitAfter returns true, a new substream is generated, and the succeeding elements after split will flow into the new substream.

splitWhen flows the element on which the predicate returned true to a new substream, whereas splitAfter flows the next element to the new substream after the element on which predicate returned true.

Scala

copysourceSource(1 to 10).splitWhen(SubstreamCancelStrategy.drain)(_ == 3)

Source(1 to 10).splitAfter(SubstreamCancelStrategy.drain)(_ == 3)

Java

copysourceSource.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).splitWhen(elem -> elem == 3);

Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).splitAfter(elem -> elem == 3);

These are useful when you scanned over something and you don’t need to care about anything behind it. A typical example is counting the number of characters for each line like below.

Scala

copysourceval text =
  "This is the first line.\n" +
  "The second line.\n" +
  "There is also the 3rd line\n"

val charCount = Source(text.toList)
  .splitAfter { _ == '\n' }
  .filter(_ != '\n')
  .map(_ => 1)
  .reduce(_ + _)
  .to(Sink.foreach(println))
  .run()

Java

copysourceString text =
    "This is the first line.\n" + "The second line.\n" + "There is also the 3rd line\n";

Source.from(Arrays.asList(text.split("")))
    .map(x -> x.charAt(0))
    .splitAfter(x -> x == '\n')
    .filter(x -> x != '\n')
    .map(x -> 1)
    .reduce((x, y) -> x + y)
    .to(Sink.foreach(x -> System.out.println(x)))
    .run(mat);

This prints out the following output.

23
16
26

Flattening operators

flatMapConcat

flatMapConcat and flatMapMerge are substream operations different from groupBy and splitWhen/After.

flatMapConcat takes a function, which is f in the following diagram. The function f of flatMapConcat transforms each input element into a Source that is then flattened into the output stream by concatenation.

Scala

copysourceSource(1 to 2).flatMapConcat(i => Source(List.fill(3)(i))).runWith(Sink.ignore)

Java

copysourceSource.from(Arrays.asList(1, 2))
    .flatMapConcat(i -> Source.from(Arrays.asList(i, i, i)))
    .runWith(Sink.ignore(), mat);

Like the concat operation on Flow, it fully consumes one Source after the other. So, there is only one substream actively running at a given time.

Then once the active substream is fully consumed, the next substream can start running. Elements from all the substreams are concatenated to the sink.

flatMapMerge

flatMapMerge is similar to flatMapConcat, but it doesn’t wait for one Source to be fully consumed. Instead, up to breadth number of streams emit elements at any given time.

Scala

copysourceSource(1 to 2).flatMapMerge(2, i => Source(List.fill(3)(i))).runWith(Sink.ignore)

Java

copysourceSource.from(Arrays.asList(1, 2))
    .flatMapMerge(2, i -> Source.from(Arrays.asList(i, i, i)))
    .runWith(Sink.ignore(), mat);