Source.queue

Materialize a BoundedSourceQueue or SourceQueue onto which elements can be pushed for emitting from the source.

Source operators

Signature (BoundedSourceQueue)

Source.queueSource.queue

Description (BoundedSourceQueue)

The BoundedSourceQueue is an optimized variant of the SourceQueue which will drop the newest elements when back pressuring and buffer is full. The BoundedSourceQueue will give immediate, synchronous feedback whether an element was accepted or not and is therefore recommended for situations where overload and dropping elements is expected and needs to be handled quickly.

In contrast, the SourceQueue offers more variety of OverflowStrategies but feedback is only asynchronously provided through a FutureCompletionStage value. In cases where elements need to be discarded quickly at times of overload to avoid out-of-memory situations, delivering feedback asynchronously can itself become a problem. This happens if elements come in faster than the feedback can be delivered in which case the feedback mechanism itself is part of the reason that an out-of-memory situation arises.

In summary, prefer BoundedSourceQueue over SourceQueue especially in high-load scenarios. Use SourceQueue if you need one of the other OverflowStrategies.

The BoundedSourceQueue contains a buffer that can be used by many producers on different threads. When the buffer is full, the BoundedSourceQueue will not accept more elements. The return value of BoundedSourceQueue.offer() immediately returns a QueueOfferResult (as opposed to an asynchronous value returned by SourceQueue). A synchronous result is important in order to avoid situations where offer acknowledgements are handled slower than the rate of which elements are offered, which will eventually lead to an Out Of Memory error.

Example (BoundedSourceQueue)

Scala

copysourceval bufferSize = 1000

val queue = Source
  .queue[Int](bufferSize)
  .map(x => x * x)
  .toMat(Sink.foreach(x => println(s"completed $x")))(Keep.left)
  .run()

val fastElements = 1 to 10

fastElements.foreach { x =>
  queue.offer(x) match {
    case QueueOfferResult.Enqueued    => println(s"enqueued $x")
    case QueueOfferResult.Dropped     => println(s"dropped $x")
    case QueueOfferResult.Failure(ex) => println(s"Offer failed ${ex.getMessage}")
    case QueueOfferResult.QueueClosed => println("Source Queue closed")
  }
}

Java

copysourceint bufferSize = 10;
int elementsToProcess = 5;

BoundedSourceQueue<Integer> sourceQueue =
    Source.<Integer>queue(bufferSize)
        .throttle(elementsToProcess, Duration.ofSeconds(3))
        .map(x -> x * x)
        .to(Sink.foreach(x -> System.out.println("got: " + x)))
        .run(system);

List<Integer> fastElements = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

fastElements.stream()
    .forEach(
        x -> {
          QueueOfferResult result = sourceQueue.offer(x);
          if (result == QueueOfferResult.enqueued()) {
            System.out.println("enqueued " + x);
          } else if (result == QueueOfferResult.dropped()) {
            System.out.println("dropped " + x);
          } else if (result instanceof QueueOfferResult.Failure failure) {
            System.out.println("Offer failed " + failure.cause().getMessage());
          } else if (result instanceof QueueOfferResult.QueueClosed$) {
            System.out.println("Bounded Source Queue closed");
          }
        });

Signature (SourceQueue)

Source.queueSource.queue Source.queueSource.queue

Description (SourceQueue)

Materialize a SourceQueue onto which elements can be pushed for emitting from the source. The queue contains a buffer, if elements are pushed onto the queue faster than the source is consumed the overflow will be handled with a strategy specified by the user. Functionality for tracking when an element has been emitted is available through SourceQueue.offer.

Using Source.queue you can push elements to the queue and they will be emitted to the stream if there is demand from downstream, otherwise they will be buffered until request for demand is received. Elements in the buffer will be discarded if downstream is terminated.

In combination with the queue, the throttle operator can be used to control the processing to a given limit, e.g. 5 elements per 3 seconds.

Example (SourceQueue)

Scala

copysourceval bufferSize = 10
val elementsToProcess = 5

val queue: BoundedSourceQueue[Int] = Source
  .queue[Int](bufferSize)
  .throttle(elementsToProcess, 3.second)
  .map(x => x * x)
  .toMat(Sink.foreach(x => println(s"completed $x")))(Keep.left)
  .run()

val source = Source(1 to 10)

source
  .map { x =>
    queue.offer(x) match {
      case QueueOfferResult.Enqueued    => println(s"enqueued $x")
      case QueueOfferResult.Dropped     => println(s"dropped $x")
      case QueueOfferResult.Failure(ex) => println(s"Offer failed ${ex.getMessage}")
      case QueueOfferResult.QueueClosed => println("Source Queue closed")
    }
  }
  .runWith(Sink.ignore)

Java

copysourceint bufferSize = 10;
int elementsToProcess = 5;

BoundedSourceQueue<Integer> sourceQueue =
    Source.<Integer>queue(bufferSize)
        .throttle(elementsToProcess, Duration.ofSeconds(3))
        .map(x -> x * x)
        .to(Sink.foreach(x -> System.out.println("got: " + x)))
        .run(system);

Source<Integer, NotUsed> source = Source.from(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));

source.map(x -> sourceQueue.offer(x)).runWith(Sink.ignore(), system);

Reactive Streams semantics

Migrating from the deprecated OverflowStrategy overloads

Old call	Replacement
Source.queue(n, OverflowStrategy.dropNew) (dropNew was removed in 2.0.0)	Source.queue[T](n), which has the same semantics: the offered element is dropped when the buffer is full.
Source.queue(n, OverflowStrategy.dropHead)	Source.queue[T](n) if dropping the offered element instead of the oldest buffered one is acceptable. Otherwise build a custom GraphStageGraphStage with a FIFO buffer that drops the head.
Source.queue(n, OverflowStrategy.dropTail)	Source.queue[T](n) if dropping the offered element instead of the youngest buffered one is acceptable. BoundedSourceQueue leaves the buffer unchanged and reports QueueOfferResult.Dropped for the offered element.
Source.queue(n, OverflowStrategy.dropBuffer)	Source.queue[T](n) combined with a GraphStageGraphStage that clears the buffer on overflow, or rework the producer to tolerate drops.
Source.queue(n, OverflowStrategy.fail)	Source.queue[T](n) and, on QueueOfferResult.Dropped, call BoundedSourceQueue.fail with a BufferOverflowException.
Source.queue(n, OverflowStrategy.backpressure)	Source.actorRefWithBackpressure (single imperative producer) or MergeHub.source (multiple producers).

SourceQueueWithComplete.offer returned a Future[QueueOfferResult]CompletionStage<QueueOfferResult>; BoundedSourceQueue.offer returns QueueOfferResult synchronously. Call sites that previously chained .map/.flatMap on the offer future can usually be rewritten as a direct match/switch on the result.

Dropping elements (dropHead, dropTail, dropBuffer)

BoundedSourceQueue drops the element being offered when the buffer is full and tells the caller synchronously, instead of completing a FutureCompletionStage.

Before:

Scala

copysourceval queue: SourceQueueWithComplete[Int] =
  Source.queue[Int](100, OverflowStrategy.dropTail).to(Sink.foreach(println)).run()

val result: Future[QueueOfferResult] = queue.offer(1)

Java

copysourceSourceQueueWithComplete<Integer> queue =
    Source.<Integer>queue(100, OverflowStrategy.dropTail())
        .to(Sink.foreach(System.out::println))
        .run(system);

CompletionStage<QueueOfferResult> result = queue.offer(1);

After:

Scala

copysourceval queue: BoundedSourceQueue[Int] =
  Source.queue[Int](100).to(Sink.foreach(println)).run()

// offer returns the result synchronously, there is no Future to wait for
queue.offer(1) match {
  case QueueOfferResult.Enqueued    => // element accepted
  case QueueOfferResult.Dropped     => // buffer full, this element was dropped
  case QueueOfferResult.Failure(ex) => // the stream failed
  case QueueOfferResult.QueueClosed => // the stream was completed
}

Java

copysourceBoundedSourceQueue<Integer> queue =
    Source.<Integer>queue(100).to(Sink.foreach(System.out::println)).run(system);

// offer returns the result synchronously, there is no CompletionStage to wait for
QueueOfferResult result = queue.offer(1);
if (result == QueueOfferResult.enqueued()) {
  // element accepted
} else if (result == QueueOfferResult.dropped()) {
  // buffer full, this element was dropped
} else if (result instanceof QueueOfferResult.Failure) {
  // the stream failed
} else if (result == QueueOfferResult.closed()) {
  // the stream was completed
}

Failing the stream on overflow (fail)

BoundedSourceQueue reports QueueOfferResult.Dropped instead of failing the stream. To keep the old behavior, fail the queue yourself:

Scala

copysourceval bufferSize = 2

// fail the stream like OverflowStrategy.fail did when the buffer is full
def offerOrFail(queue: BoundedSourceQueue[Int], elem: Int): QueueOfferResult = {
  val result = queue.offer(elem)
  if (result == QueueOfferResult.Dropped)
    queue.fail(BufferOverflowException(s"Buffer overflow (max capacity was: $bufferSize)!"))
  result
}

Java

copysourcestatic final int bufferSize = 2;

// fail the stream like OverflowStrategy.fail did when the buffer is full
static QueueOfferResult offerOrFail(BoundedSourceQueue<Integer> queue, int elem) {
  QueueOfferResult result = queue.offer(elem);
  if (result == QueueOfferResult.dropped()) {
    queue.fail(
        new BufferOverflowException("Buffer overflow (max capacity was: " + bufferSize + ")!"));
  }
  return result;
}

Backpressuring the producer (backpressure)

A typical use of OverflowStrategy.backpressure offers the next element only after the previous offer completed:

Scala

copysourceval queue: SourceQueueWithComplete[Int] =
  Source.queue[Int](100, OverflowStrategy.backpressure).to(Sink.foreach(println)).run()

// the next offer is only made once the previous one has completed
val allOffered: Future[QueueOfferResult] = (1 to 10).foldLeft(
  Future.successful[QueueOfferResult](QueueOfferResult.Enqueued)) { (previous, elem) =>
  previous.flatMap(_ => queue.offer(elem))
}
allOffered.map(_ => queue.complete())

Java

copysourceSourceQueueWithComplete<Integer> queue =
    Source.<Integer>queue(100, OverflowStrategy.backpressure())
        .to(Sink.foreach(System.out::println))
        .run(system);

// the next offer is only made once the previous one has completed
CompletionStage<QueueOfferResult> allOffered =
    CompletableFuture.completedFuture(QueueOfferResult.enqueued());
for (int i = 1; i <= 10; i++) {
  final int elem = i;
  allOffered = allOffered.thenCompose(previous -> queue.offer(elem));
}
allOffered.thenRun(queue::complete);

For a single imperative producer, Source.actorRefWithBackpressure combined with ask gives the same “wait for the offer to be accepted” shape. The ack is sent to the sender of each element, so the ask FutureCompletionStage completes once the element was emitted. Sending a new element before the previous one was acknowledged fails the stream, just like exceeding maxConcurrentOffers failed the offer before. With the typed actor API, use ActorSource.actorRefWithBackpressure the same way.

Scala

copysourcecase object Ack
case object Complete

val (ref: ActorRef, done: Future[Seq[Int]]) =
  Source
    .actorRefWithBackpressure[Int](
      ackMessage = Ack,
      completionMatcher = { case Complete => CompletionStrategy.draining },
      failureMatcher = PartialFunction.empty)
    .toMat(Sink.seq)(Keep.both)
    .run()

// Replaces queue.offer(elem): the Future completes with Ack once the element was emitted
// downstream. Unlike the deprecated queue it fails with an AskTimeoutException instead of
// hanging forever if downstream stalls.
implicit val timeout: Timeout = 3.seconds
def offer(elem: Int): Future[Any] = ref ? elem

// Like before, the next element must only be offered once the previous one was acknowledged,
// otherwise the stream fails.
val allOffered: Future[Any] = (1 to 10).foldLeft(Future.successful[Any](Ack)) { (previous, elem) =>
  previous.flatMap(_ => offer(elem))
}
allOffered.foreach(_ => ref ! Complete)

Java

copysourcePair<ActorRef, CompletionStage<List<Integer>>> pair =
    Source.<Integer>actorRefWithBackpressure(
            "ack",
            // complete when we send "complete"
            msg ->
                "complete".equals(msg)
                    ? Optional.of(CompletionStrategy.draining())
                    : Optional.empty(),
            // do not fail on any message
            msg -> Optional.empty())
        .toMat(Sink.seq(), Keep.both())
        .run(system);
ActorRef ref = pair.first();

// Replaces queue.offer(elem): the CompletionStage completes with "ack" once the element was
// emitted downstream. Unlike the deprecated queue it fails with an AskTimeoutException instead
// of hanging forever if downstream stalls.
Duration timeout = Duration.ofSeconds(3);

// Like before, the next element must only be offered once the previous one was acknowledged,
// otherwise the stream fails.
CompletionStage<Object> allOffered = CompletableFuture.completedFuture("ack");
for (int i = 1; i <= 10; i++) {
  final int elem = i;
  allOffered = allOffered.thenCompose(previous -> Patterns.ask(ref, elem, timeout));
}
allOffered.thenRun(() -> ref.tell("complete", ActorRef.noSender()));

For multiple producers, or when maxConcurrentOffers was greater than 1, use MergeHub.source. Each producer becomes a stream connected to the materialized Sink and is back-pressured on its own. If the elements already come from a stream (an iterator, a Future, another Source), connecting that stream directly is usually simpler than offering elements one at a time.

Scala

copysourceval (sink: Sink[Int, NotUsed], done: Future[Seq[Int]]) =
  MergeHub.source[Int](perProducerBufferSize = 16).take(20).toMat(Sink.seq)(Keep.both).run()

// Each producer is a stream of its own and is back-pressured independently,
// there is no offer Future to track per element.
Source(1 to 10).runWith(sink)
Source(11 to 20).runWith(sink)

Java

copysourcePair<Sink<Integer, NotUsed>, CompletionStage<List<Integer>>> pair =
    MergeHub.of(Integer.class, 16).take(20).toMat(Sink.seq(), Keep.both()).run(system);
Sink<Integer, NotUsed> sink = pair.first();

// Each producer is a stream of its own and is back-pressured independently,
// there is no offer CompletionStage to track per element.
Source.range(1, 10).runWith(sink, system);
Source.range(11, 20).runWith(sink, system);