This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 559c5b9cee docs: add code examples for migrating off deprecated
Source.queue overloads (#3605)
559c5b9cee is described below
commit 559c5b9ceef376fd77cef77e650927c2cf62a802
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Oct 10 06:27:17 2026 +0100
docs: add code examples for migrating off deprecated Source.queue overloads
(#3605)
* docs: add code examples for migrating off deprecated Source.queue
overloads
Motivation:
2.0.0-M4 deprecated the Source.queue overloads that take an
OverflowStrategy, but the migration table only described replacements
without code, and the OverflowStrategy.backpressure alternatives were
hard to apply. The table also referenced OverflowStrategy.dropNew,
which was removed in 2.0, and described dropTail as equivalent to
BoundedSourceQueue, which it is not.
Modification:
Add compiled Scala and Java before/after examples to the Source.queue
operator page for the drop strategies, fail (BoundedSourceQueue.fail
with BufferOverflowException), backpressure with a single producer
(Source.actorRefWithBackpressure with ask) and backpressure with
multiple producers (MergeHub.source). Correct the dropNew, dropHead
and dropTail rows of the migration table. Link the section from the
1.x to 2.x migration guide.
Result:
Users of the deprecated overloads have tested code examples for each
OverflowStrategy replacement.
Tests:
- sbt "docs/testOnly docs.stream.operators.source.QueueMigrationDocSpec
jdocs.stream.operators.source.QueueMigrationDocTest" - 8 passed
References:
Fixes #3458
* docs: use a plain-text heading for the Source.queue migration anchor
Paradox only generates a heading anchor when the heading is a single
text node, so the heading containing inline code had no anchor and the
migration guide link failed validation.
---
.../paradox/migration/migration-guide-1.x-2.x.md | 9 +
.../main/paradox/stream/operators/Source/queue.md | 68 ++++++-
.../operators/source/QueueMigrationDocTest.java | 213 +++++++++++++++++++++
.../operators/source/QueueMigrationDocSpec.scala | 158 +++++++++++++++
4 files changed, 442 insertions(+), 6 deletions(-)
diff --git a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
index 7bb7a1ee40..0b4d1f7a78 100644
--- a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
+++ b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
@@ -28,6 +28,15 @@ Not every deprecated API has been removed in Pekko 2.x but
many have been.
Java API users may find that they have more deprecations to deal with because
the Scala API is more stable and there
were a few mistakes in the Java API where Scala classes leaked into some of
the Java API methods.
+### Deprecations in Pekko 2.x
+
+* The `Source.queue` overloads that take an `OverflowStrategy` and materialize
a `SourceQueueWithComplete` are
+deprecated. Their asynchronous `offer` can hang indefinitely under
`OverflowStrategy.backpressure` when downstream
+stalls. Use `Source.queue(bufferSize)` (which materializes a
`BoundedSourceQueue` with synchronous feedback),
+`Source.actorRefWithBackpressure` or `MergeHub.source` instead. See
+@ref:[Migrating from the deprecated Source.queue
overloads](../stream/operators/Source/queue.md#migrating-from-the-deprecated-overflowstrategy-overloads)
+for a replacement per `OverflowStrategy`, with code examples.
+
## Additional Breaking Changes in Pekko 2.x
* In the Scala DSL for Flow and Source, the `watchTermination` function call
no longer needs an empty param
list before a second param list. Instead of `watchTermination(){ ... }`, you
now must use `watchTermination{ ... }`.
diff --git a/docs/src/main/paradox/stream/operators/Source/queue.md
b/docs/src/main/paradox/stream/operators/Source/queue.md
index 02d706ba80..9227001975 100644
--- a/docs/src/main/paradox/stream/operators/Source/queue.md
+++ b/docs/src/main/paradox/stream/operators/Source/queue.md
@@ -8,7 +8,7 @@ Materialize a `BoundedSourceQueue` or `SourceQueue` onto which
elements can be p
The `Source.queue` overloads that accept an @apidoc[OverflowStrategy] and
materialize a `SourceQueueWithComplete` are **deprecated**. Their asynchronous
`offer` @scala[`Future`]@java[`CompletionStage`] can hang indefinitely under
`OverflowStrategy.backpressure` when downstream stalls, which has caused
real-world deadlocks.
-Prefer `Source.queue[T](bufferSize)` (this page), which materializes a
@apidoc[BoundedSourceQueue] with synchronous feedback and drop-newest overflow.
For backpressure towards the producer, use
@ref:[`Source.actorRefWithBackpressure`](actorRefWithBackpressure.md) (single
imperative producer) or `MergeHub.source` (multiple producers). See the
[migration
table](#migrating-from-the-deprecated-sourcequeueint-overflowstrategy-overloads)
below for a per-strategy replacement.
+Prefer `Source.queue[T](bufferSize)` (this page), which materializes a
@apidoc[BoundedSourceQueue] with synchronous feedback and drop-newest overflow.
For backpressure towards the producer, use
@ref:[`Source.actorRefWithBackpressure`](actorRefWithBackpressure.md) (single
imperative producer) or `MergeHub.source` (multiple producers). See the
[migration table](#migrating-from-the-deprecated-overflowstrategy-overloads)
below for a per-strategy replacement.
@@@
@@ -45,7 +45,7 @@ Java
@@@ warning
-These two overloads are deprecated since 2.0.0. See the [migration
table](#migrating-from-the-deprecated-sourcequeueint-overflowstrategy-overloads)
below.
+These two overloads are deprecated since 2.0.0. See the [migration
table](#migrating-from-the-deprecated-overflowstrategy-overloads) below.
@@@
@@ -83,15 +83,71 @@ Java
@@@
-## Migrating from the deprecated `Source.queue(Int, OverflowStrategy)`
overloads
+## Migrating from the deprecated OverflowStrategy overloads
| Old call | Replacement |
|----------|-------------|
-| `Source.queue(n, OverflowStrategy.dropNew)` | `Source.queue[T](n)` —
`BoundedSourceQueue` already drops the newest element. |
-| `Source.queue(n, OverflowStrategy.dropHead)` | `Source.queue[T](n)` if
drop-new is acceptable. Otherwise build a custom @apidoc[GraphStage] with a
FIFO buffer that drops the head. |
-| `Source.queue(n, OverflowStrategy.dropTail)` | Same as above;
`BoundedSourceQueue` always drops the newest offer (i.e. the tail). |
+| `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 @apidoc[GraphStage] 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 @apidoc[GraphStage] 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)` |
@ref:[`Source.actorRefWithBackpressure`](actorRefWithBackpressure.md) (single
imperative producer) or `MergeHub.source` (multiple producers). |
`SourceQueueWithComplete.offer` returned a
@scala[`Future[QueueOfferResult]`]@java[`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
@scala[`Future`]@java[`CompletionStage`].
+
+Before:
+
+Scala
+: @@snip
[QueueMigrationDocSpec.scala](/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala)
{ #drop-before }
+
+Java
+: @@snip
[QueueMigrationDocTest.java](/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java)
{ #drop-before }
+
+After:
+
+Scala
+: @@snip
[QueueMigrationDocSpec.scala](/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala)
{ #drop-after }
+
+Java
+: @@snip
[QueueMigrationDocTest.java](/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java)
{ #drop-after }
+
+### 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
+: @@snip
[QueueMigrationDocSpec.scala](/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala)
{ #fail-after }
+
+Java
+: @@snip
[QueueMigrationDocTest.java](/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java)
{ #fail-after }
+
+### Backpressuring the producer (`backpressure`)
+
+A typical use of `OverflowStrategy.backpressure` offers the next element only
after the previous offer completed:
+
+Scala
+: @@snip
[QueueMigrationDocSpec.scala](/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala)
{ #backpressure-before }
+
+Java
+: @@snip
[QueueMigrationDocTest.java](/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java)
{ #backpressure-before }
+
+For a single imperative producer,
@ref:[`Source.actorRefWithBackpressure`](actorRefWithBackpressure.md) 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`
@scala[`Future`]@java[`CompletionStage`] 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 [...]
+
+Scala
+: @@snip
[QueueMigrationDocSpec.scala](/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala)
{ #backpressure-single-producer }
+
+Java
+: @@snip
[QueueMigrationDocTest.java](/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java)
{ #backpressure-single-producer }
+
+For multiple producers, or when `maxConcurrentOffers` was greater than 1, use
@ref:[`MergeHub.source`](../../stream-dynamic.md#using-the-mergehub). 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
+: @@snip
[QueueMigrationDocSpec.scala](/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala)
{ #backpressure-multiple-producers }
+
+Java
+: @@snip
[QueueMigrationDocTest.java](/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java)
{ #backpressure-multiple-producers }
diff --git
a/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java
b/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java
new file mode 100644
index 0000000000..52e4fccf09
--- /dev/null
+++
b/docs/src/test/java/jdocs/stream/operators/source/QueueMigrationDocTest.java
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package jdocs.stream.operators.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import jdocs.AbstractJavaTest;
+import org.apache.pekko.Done;
+import org.apache.pekko.NotUsed;
+import org.apache.pekko.actor.ActorRef;
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.japi.Pair;
+import org.apache.pekko.pattern.Patterns;
+import org.apache.pekko.stream.BoundedSourceQueue;
+import org.apache.pekko.stream.BufferOverflowException;
+import org.apache.pekko.stream.CompletionStrategy;
+import org.apache.pekko.stream.OverflowStrategy;
+import org.apache.pekko.stream.QueueOfferResult;
+import org.apache.pekko.stream.javadsl.Keep;
+import org.apache.pekko.stream.javadsl.MergeHub;
+import org.apache.pekko.stream.javadsl.Sink;
+import org.apache.pekko.stream.javadsl.Source;
+import org.apache.pekko.stream.javadsl.SourceQueueWithComplete;
+import org.apache.pekko.stream.testkit.TestSubscriber;
+import org.apache.pekko.stream.testkit.javadsl.TestSink;
+import org.apache.pekko.testkit.javadsl.TestKit;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+public class QueueMigrationDocTest extends AbstractJavaTest {
+
+ static ActorSystem system;
+
+ @BeforeAll
+ public static void setup() {
+ system = ActorSystem.create("QueueMigrationDocTest");
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ TestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ @SuppressWarnings("deprecation")
+ private CompletionStage<QueueOfferResult> dropTailBefore() {
+ // #drop-before
+ SourceQueueWithComplete<Integer> queue =
+ Source.<Integer>queue(100, OverflowStrategy.dropTail())
+ .to(Sink.foreach(System.out::println))
+ .run(system);
+
+ CompletionStage<QueueOfferResult> result = queue.offer(1);
+ // #drop-before
+ return result;
+ }
+
+ @Test
+ public void replaceDropTail() throws Exception {
+ assertEquals(
+ QueueOfferResult.enqueued(),
+ dropTailBefore().toCompletableFuture().get(3, TimeUnit.SECONDS));
+
+ // #drop-after
+ BoundedSourceQueue<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
+ }
+ // #drop-after
+ assertEquals(QueueOfferResult.enqueued(), result);
+ }
+
+ // #fail-after
+ static 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;
+ }
+
+ // #fail-after
+
+ @Test
+ public void replaceFail() {
+ // downstream never requests, so the buffer fills up
+ Pair<BoundedSourceQueue<Integer>, TestSubscriber.Probe<Integer>> pair =
+ Source.<Integer>queue(bufferSize).toMat(TestSink.create(system),
Keep.both()).run(system);
+ BoundedSourceQueue<Integer> queue = pair.first();
+ assertEquals(QueueOfferResult.enqueued(), offerOrFail(queue, 1));
+ assertEquals(QueueOfferResult.enqueued(), offerOrFail(queue, 2));
+ assertEquals(QueueOfferResult.dropped(), offerOrFail(queue, 3));
+ assertInstanceOf(BufferOverflowException.class,
pair.second().expectSubscriptionAndError());
+ }
+
+ @SuppressWarnings("deprecation")
+ private CompletionStage<Done> backpressureBefore() {
+ // #backpressure-before
+ SourceQueueWithComplete<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);
+ // #backpressure-before
+ return queue.watchCompletion();
+ }
+
+ @Test
+ public void replaceBackpressureSingleProducer() throws Exception {
+ assertEquals(
+ Done.getInstance(), backpressureBefore().toCompletableFuture().get(3,
TimeUnit.SECONDS));
+
+ // #backpressure-single-producer
+ Pair<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()));
+ // #backpressure-single-producer
+
+ assertEquals(
+ IntStream.rangeClosed(1, 10).boxed().collect(Collectors.toList()),
+ pair.second().toCompletableFuture().get(3, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void replaceBackpressureMultipleProducers() throws Exception {
+ // #backpressure-multiple-producers
+ Pair<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);
+ // #backpressure-multiple-producers
+
+ List<Integer> result = pair.second().toCompletableFuture().get(3,
TimeUnit.SECONDS);
+ assertEquals(
+ IntStream.rangeClosed(1, 20).boxed().collect(Collectors.toSet()),
+ result.stream().collect(Collectors.toSet()));
+ }
+}
diff --git
a/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala
b/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala
new file mode 100644
index 0000000000..80167be97e
--- /dev/null
+++
b/docs/src/test/scala/docs/stream/operators/source/QueueMigrationDocSpec.scala
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package docs.stream.operators.source
+
+import scala.annotation.nowarn
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.Done
+import pekko.NotUsed
+import pekko.actor.ActorRef
+import pekko.pattern.ask
+import pekko.stream.BoundedSourceQueue
+import pekko.stream.BufferOverflowException
+import pekko.stream.CompletionStrategy
+import pekko.stream.OverflowStrategy
+import pekko.stream.QueueOfferResult
+import pekko.stream.scaladsl.Keep
+import pekko.stream.scaladsl.MergeHub
+import pekko.stream.scaladsl.Sink
+import pekko.stream.scaladsl.Source
+import pekko.stream.scaladsl.SourceQueueWithComplete
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.PekkoSpec
+import pekko.util.Timeout
+
+class QueueMigrationDocSpec extends PekkoSpec {
+ import system.dispatcher
+
+ "Source.queue migration" should {
+
+ "replace dropTail with Source.queue(bufferSize)" in {
+ @nowarn("msg=deprecated")
+ def before(): Future[QueueOfferResult] = {
+ // #drop-before
+ val queue: SourceQueueWithComplete[Int] =
+ Source.queue[Int](100,
OverflowStrategy.dropTail).to(Sink.foreach(println)).run()
+
+ val result: Future[QueueOfferResult] = queue.offer(1)
+ // #drop-before
+ result
+ }
+ before().futureValue shouldBe QueueOfferResult.Enqueued
+
+ // #drop-after
+ val 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
+ }
+ // #drop-after
+ }
+
+ "replace fail with Source.queue(bufferSize) and BoundedSourceQueue.fail"
in {
+ // #fail-after
+ val 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
+ }
+ // #fail-after
+
+ // downstream never requests, so the buffer fills up
+ val (queue, probe) =
Source.queue[Int](bufferSize).toMat(TestSink[Int]())(Keep.both).run()
+ offerOrFail(queue, 1) shouldBe QueueOfferResult.Enqueued
+ offerOrFail(queue, 2) shouldBe QueueOfferResult.Enqueued
+ offerOrFail(queue, 3) shouldBe QueueOfferResult.Dropped
+ probe.expectSubscriptionAndError() shouldBe a[BufferOverflowException]
+ }
+
+ "replace backpressure with Source.actorRefWithBackpressure for a single
producer" in {
+ @nowarn("msg=deprecated")
+ def before(): Future[Done] = {
+ // #backpressure-before
+ val 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())
+ // #backpressure-before
+ queue.watchCompletion()
+ }
+ before().futureValue shouldBe Done
+
+ // #backpressure-single-producer
+ case 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)
+ // #backpressure-single-producer
+
+ done.futureValue shouldBe (1 to 10)
+ }
+
+ "replace backpressure with MergeHub.source for multiple producers" in {
+ // #backpressure-multiple-producers
+ val (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)
+ // #backpressure-multiple-producers
+
+ done.futureValue should contain theSameElementsAs (1 to 20)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
