Partition

Fan-out the stream to several streams.

Fan-out operators

Signature

PartitionPartition

Description

Fan-out the stream to several streams. Each upstream element is emitted to one downstream consumer according to the partitioner function applied to the element.

Example

Here is an example of using Partition to split a Source of integers to one Sink for the even numbers and another Sink for the odd numbers.

Scala

copysourceimport akka.NotUsed
import akka.stream.Attributes
import akka.stream.Attributes.LogLevels
import akka.stream.ClosedShape
import akka.stream.scaladsl.Flow
import akka.stream.scaladsl.GraphDSL
import akka.stream.scaladsl.Partition
import akka.stream.scaladsl.RunnableGraph
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Source

val source: Source[Int, NotUsed] = Source(1 to 10)

val even: Sink[Int, NotUsed] =
  Flow[Int].log("even").withAttributes(Attributes.logLevels(onElement = LogLevels.Info)).to(Sink.ignore)
val odd: Sink[Int, NotUsed] =
  Flow[Int].log("odd").withAttributes(Attributes.logLevels(onElement = LogLevels.Info)).to(Sink.ignore)

RunnableGraph
  .fromGraph(GraphDSL.create() { implicit builder =>
    import GraphDSL.Implicits._
    val partition = builder.add(Partition[Int](2, element => if (element % 2 == 0) 0 else 1))
    source ~> partition.in
    partition.out(0) ~> even
    partition.out(1) ~> odd
    ClosedShape
  })
  .run()

Java

copysourceimport akka.NotUsed;
import akka.stream.Attributes;
import akka.stream.ClosedShape;
import akka.stream.UniformFanOutShape;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.GraphDSL;
import akka.stream.javadsl.Partition;
import akka.stream.javadsl.RunnableGraph;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

Source<Integer, NotUsed> source = Source.range(1, 10);

Sink<Integer, NotUsed> even =
    Flow.of(Integer.class)
        .log("even")
        .withAttributes(Attributes.createLogLevels(Attributes.logLevelInfo()))
        .to(Sink.ignore());
Sink<Integer, NotUsed> odd =
    Flow.of(Integer.class)
        .log("odd")
        .withAttributes(Attributes.createLogLevels(Attributes.logLevelInfo()))
        .to(Sink.ignore());

RunnableGraph.fromGraph(
        GraphDSL.create(
            builder -> {
              UniformFanOutShape<Integer, Integer> partition =
                  builder.add(
                      Partition.create(
                          Integer.class, 2, element -> (element % 2 == 0) ? 0 : 1));
              builder.from(builder.add(source)).viaFanOut(partition);
              builder.from(partition.out(0)).to(builder.add(even));
              builder.from(partition.out(1)).to(builder.add(odd));
              return ClosedShape.getInstance();
            }))
    .run(system);

Reactive Streams semantics