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 26d8b60f19 feat(stream): add alsoTo overload with configurable
cancellation propagation (#3127)
26d8b60f19 is described below
commit 26d8b60f19ff31afdbf0ee920a400038d2ef1cd8
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Mon Jun 29 17:27:31 2026 +0800
feat(stream): add alsoTo overload with configurable cancellation
propagation (#3127)
* feat(stream): add alsoTo overload with configurable cancellation
propagation
Motivation:
alsoTo uses Broadcast[Out](2, eagerCancel = true) internally. When the
side sink fails or cancels, the entire stream is terminated. Users cannot
isolate the side sink — for example, a fire-and-forget logging sink
should not kill the main business stream when the logging destination is
temporarily unavailable.
Modification:
Add a new alsoTo(sink, propagateCancellation: Boolean) overload. When
propagateCancellation is false, a new ResilientAlsoTo GraphStage is used
instead of Broadcast. ResilientAlsoTo backpressures when either output
backpressures (same contract as alsoTo), but when the side sink cancels
or fails, elements continue flowing to the main downstream only. A
warning is logged on side sink failure. Also add alsoToMat overloads for
both Scala and Java DSLs, and update FlowWithContext/SourceWithContext.
Result:
Users can now use alsoTo(sink, propagateCancellation = false) to
fire-and-forget to a side sink without risking main stream termination.
Default behavior (propagateCancellation = true) is unchanged.
Tests:
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.FlowAlsoToSpec" — 11/11 passed
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.DslConsistencySpec" — 12/12 passed
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.FlowAlsoToAllSpec" — 2/2 passed
References:
Fixes #3104
* fix(stream): correct @since version, add Java tests, clean up
DefaultAttributes
Motivation:
Review feedback: @since should be 2.0.0, Java tests were missing, and
DefaultAttributes.resilientAlsoTo was unnecessary.
Modification:
- Fix @since 1.2.0 to @since 2.0.0 across all new alsoTo/alsoToMat
overloads
- Add Java DSL compilation tests for alsoTo(propagateCancellation)
and alsoToMat(propagateCancellation) in FlowTest and SourceTest
- Remove DefaultAttributes.resilientAlsoTo; use inline
Attributes.name("resilientAlsoTo") instead
Result:
Correct version annotation, Java API coverage, cleaner DefaultAttributes.
Tests:
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.FlowAlsoToSpec
org.apache.pekko.stream.DslConsistencySpec
org.apache.pekko.stream.javadsl.FlowTest
org.apache.pekko.stream.javadsl.SourceTest" — 222/222 passed
References:
Refs #3104
* refactor(stream): implement alsoTo(propagateCancellation=false) via
Broadcast
Motivation:
ResilientAlsoTo was a separate GraphStage duplicating Broadcast's
fan-out logic. WireTap was also modified with an unrelated backpressure
parameter.
Modification:
Add nonEagerCancelOutputs: Set[Int] parameter to Broadcast. Outputs in
this set do not trigger stage cancellation when they cancel — elements
continue flowing to remaining outputs. alsoTo(propagateCancellation=false)
now uses Broadcast(2, eagerCancel=true, nonEagerCancelOutputs=Set(1)),
eliminating ResilientAlsoTo entirely. WireTap is reverted to its
original form. Also remove DefaultAttributes.resilientAlsoTo and fix
@since to 2.0.0 across all new methods. Add Java DSL compilation tests.
Result:
Cleaner implementation built on existing Broadcast. No new GraphStage.
Same 224 tests pass.
Tests:
- 224/224 passed (FlowAlsoToSpec, DslConsistencySpec, FlowTest, SourceTest,
FlowAlsoToAllSpec)
References:
Refs #3104
* style: apply javafmt to Java test files
* fix(stream): preserve Broadcast binary compatibility and tighten require
Motivation:
The previous commit restructured Broadcast to take a 3-arg primary
constructor (Int, Boolean, Set[Int]), replacing the historical 2-arg
(Int, Boolean) primary. That changes the JVM `<init>` descriptor, which
is a binary-compatibility break for any external caller compiled against
prior Pekko versions invoking `new Broadcast(n, eager)`. Also, the
require condition allowed a non-empty nonEagerCancelOutputs with
eagerCancel=false, where the parameter is silently meaningless.
Modification:
- Restore the 2-arg `(Int, Boolean)` as the primary constructor,
preserving the `<init>(IZ)V` JVM descriptor for existing callers.
Add the 3-arg `(Int, Boolean, Set[Int])` as an auxiliary
constructor. Store nonEagerCancelOutputs in a private mutable
backing field assigned once from the auxiliary constructor; expose
it via a public `def nonEagerCancelOutputs` accessor (source- and
binary-compatible since it is a new, unreleased API).
- Strengthen require: nonEagerCancelOutputs must imply eagerCancel=true
(rejected with a descriptive message otherwise), and must be a
subset of valid output indices.
- Add two regression tests in GraphBroadcastSpec:
* "continue to remaining outputs when a non-eager output cancels"
directly exercises the new Broadcast overload.
* "reject nonEagerCancelOutputs when eagerCancel is false" verifies
the require guard.
Result:
Existing callers compiled against the old Broadcast(Int, Boolean)
constructor keep working (MiMa clean). New 3-arg constructor is
source-available. Invalid Broadcast configurations are rejected at
construction time with a clear message.
Tests:
- sbt 'stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.GraphBroadcastSpec' -> 13/13 passed
- sbt 'stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.FlowAlsoToSpec' -> 11/11 passed
- sbt 'stream-tests / Test / testOnly
org.apache.pekko.stream.DslConsistencySpec' -> 12/12 passed
- sbt '+stream / mimaReportBinaryIssues' -> clean
References:
Refs #3104
* style: apply scalafmt to GraphBroadcastSpec
References:
Refs #3104
* fix(stream): add Java Broadcast non-eager cancellation factory
Motivation:
PR #3127 added a Scala Broadcast factory overload used by
alsoTo(propagateCancellation = false), but the Java DSL did not expose the
matching factory. The Scala 3.3 DSL factory consistency check therefore failed
in CI.
Modification:
Add the Java Broadcast.create overload accepting nonEagerCancelOutputs,
cover it with a Java GraphDSL behavior test, and clarify alsoTo cancellation
propagation docs.
Result:
Java and Scala DSL factory coverage is back in parity and alsoTo
non-propagating cancellation behavior is documented.
Tests:
- rtk sbt "++ 3.3.8" "stream-tests / Test / testOnly
org.apache.pekko.stream.DslFactoriesConsistencySpec
org.apache.pekko.stream.javadsl.GraphDslTest
org.apache.pekko.stream.scaladsl.GraphBroadcastSpec
org.apache.pekko.stream.scaladsl.FlowAlsoToSpec" - passed
- rtk sbt docs/paradox - passed
- rtk sbt "+stream / mimaReportBinaryIssues" - passed
- rtk scalafmt --mode diff-ref=origin/main - passed
- rtk scalafmt --list --mode diff-ref=origin/main - passed
- rtk sbt javafmtAll with JDK 17 - passed
- rtk sbt javafmtCheckAll checkCodeStyle with JDK 17 - passed
- rtk git diff --check - passed
References:
Refs #3127
---
.../stream/operators/Source-or-Flow/alsoTo.md | 11 +-
project/StreamOperatorsIndexGenerator.scala | 1 +
.../org/apache/pekko/stream/javadsl/FlowTest.java | 10 +
.../apache/pekko/stream/javadsl/GraphDslTest.java | 22 ++
.../apache/pekko/stream/javadsl/SourceTest.java | 10 +
.../apache/pekko/stream/DslConsistencySpec.scala | 1 +
.../pekko/stream/scaladsl/FlowAlsoToSpec.scala | 288 +++++++++++++++++++++
.../pekko/stream/scaladsl/GraphBroadcastSpec.scala | 45 ++++
.../org/apache/pekko/stream/javadsl/Flow.scala | 41 +++
.../org/apache/pekko/stream/javadsl/Graph.scala | 21 ++
.../org/apache/pekko/stream/javadsl/Source.scala | 41 +++
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 24 ++
.../apache/pekko/stream/javadsl/SubSource.scala | 24 ++
.../org/apache/pekko/stream/scaladsl/Flow.scala | 54 ++++
.../pekko/stream/scaladsl/FlowWithContext.scala | 4 +
.../pekko/stream/scaladsl/FlowWithContextOps.scala | 8 +
.../org/apache/pekko/stream/scaladsl/Graph.scala | 61 ++++-
.../pekko/stream/scaladsl/SourceWithContext.scala | 4 +
18 files changed, 666 insertions(+), 4 deletions(-)
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoTo.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoTo.md
index 452e012235..430b4fedb2 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoTo.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoTo.md
@@ -14,6 +14,12 @@ Attaches the given `Sink` to this `Flow`, meaning that
elements that pass throug
Attaches the given `Sink` to this `Flow`, meaning that elements that pass
through this `Flow` will also be sent to the `Sink`.
+By default, cancellation or failure of the attached `Sink` cancels the main
stream. The `alsoTo` overload with
+`propagateCancellation = false` can be used when the attached `Sink` is a
best-effort side sink, such as logging or
+metrics, and its cancellation or failure should not terminate the main stream.
In that mode, the operator still
+backpressures when the side `Sink` is active and backpressuring, but once the
side `Sink` cancels or fails, elements
+continue to the main downstream only.
+
## Reactive Streams semantics
@@@div { .callout }
@@ -24,8 +30,9 @@ Attaches the given `Sink` to this `Flow`, meaning that
elements that pass throug
**completes** when upstream completes
-**cancels** when downstream or `Sink` cancels
+**cancels** when downstream cancels. With the default cancellation
propagation, the operator also cancels when the
+attached `Sink` cancels or fails. With `propagateCancellation = false`,
cancellation or failure of the attached `Sink`
+does not cancel the main stream.
@@@
-
diff --git a/project/StreamOperatorsIndexGenerator.scala
b/project/StreamOperatorsIndexGenerator.scala
index 1dcce00698..daf88cc4e4 100644
--- a/project/StreamOperatorsIndexGenerator.scala
+++ b/project/StreamOperatorsIndexGenerator.scala
@@ -81,6 +81,7 @@ object StreamOperatorsIndexGenerator extends AutoPlugin {
"mergeGraph",
"wireTapGraph",
"alsoToGraph",
+ "resilientAlsoToGraph",
"orElseGraph",
"divertToGraph",
"zipWithGraph",
diff --git
a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java
b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java
index 6ee8f1c1e6..8737c6d27d 100644
--- a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java
+++ b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java
@@ -1879,6 +1879,16 @@ public class FlowTest extends StreamTestJupiter {
Flow.of(Integer.class).alsoToMat(Sink.ignore(), (i, n) -> "foo");
}
+ @Test
+ public void mustBeAbleToUseAlsoToWithPropagateCancellation() {
+ final Flow<Integer, Integer, NotUsed> f =
Flow.of(Integer.class).alsoTo(Sink.ignore(), false);
+ final Flow<Integer, Integer, NotUsed> f2 =
Flow.of(Integer.class).alsoTo(Sink.ignore(), true);
+ final Flow<Integer, Integer, String> f3 =
+ Flow.of(Integer.class).alsoToMat(Sink.ignore(), false, (i, n) ->
"foo");
+ final Flow<Integer, Integer, String> f4 =
+ Flow.of(Integer.class).alsoToMat(Sink.ignore(), true, (i, n) -> "foo");
+ }
+
@Test
public void mustBeAbleToUseAlsoToAll() {
final Flow<Integer, Integer, NotUsed> f =
diff --git
a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/GraphDslTest.java
b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/GraphDslTest.java
index 869ac16898..135d67b0b2 100644
---
a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/GraphDslTest.java
+++
b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/GraphDslTest.java
@@ -17,6 +17,7 @@ import static org.junit.jupiter.api.Assertions.*;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
@@ -80,6 +81,27 @@ public class GraphDslTest extends StreamTestJupiter {
new String[] {"31", "32", "33", "34", "35", "41", "42", "43", "44",
"45"}, res);
}
+ @Test
+ public void mustKeepFlowingWhenNonEagerBroadcastOutputCancels() throws
Exception {
+ final Source<Integer, NotUsed> source = Source.from(Arrays.asList(1, 2,
3));
+ final Sink<Integer, CompletionStage<List<Integer>>> sink = Sink.seq();
+
+ final RunnableGraph<CompletionStage<List<Integer>>> graph =
+ RunnableGraph.fromGraph(
+ GraphDSL.create(
+ sink,
+ (builder, out) -> {
+ final UniformFanOutShape<Integer, Integer> broadcast =
+ builder.add(Broadcast.create(2, true,
Collections.singleton(1)));
+
builder.from(builder.add(source)).viaFanOut(broadcast).to(out);
+
builder.from(broadcast).to(builder.add(Sink.<Integer>cancelled()));
+ return ClosedShape.getInstance();
+ }));
+
+ assertEquals(
+ Arrays.asList(1, 2, 3), graph.run(system).toCompletableFuture().get(3,
TimeUnit.SECONDS));
+ }
+
@Test
@SuppressWarnings("unused")
public void demonstrateConnectErrors() {
diff --git
a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
index 519a2eeec7..465d3d350b 100644
--- a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
+++ b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
@@ -1476,6 +1476,16 @@ public class SourceTest extends StreamTestJupiter {
Source.<Integer>empty().alsoToMat(Sink.ignore(), (i, n) -> "foo");
}
+ @Test
+ public void mustBeAbleToUseAlsoToWithPropagateCancellation() {
+ final Source<Integer, NotUsed> f =
Source.<Integer>empty().alsoTo(Sink.ignore(), false);
+ final Source<Integer, NotUsed> f2 =
Source.<Integer>empty().alsoTo(Sink.ignore(), true);
+ final Source<Integer, String> f3 =
+ Source.<Integer>empty().alsoToMat(Sink.ignore(), false, (i, n) ->
"foo");
+ final Source<Integer, String> f4 =
+ Source.<Integer>empty().alsoToMat(Sink.ignore(), true, (i, n) ->
"foo");
+ }
+
@Test
public void mustBeAbleToUseAlsoToAll() {
final Source<Integer, NotUsed> f =
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala
index 1befcaefb7..2aa2c3ec01 100755
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala
@@ -90,6 +90,7 @@ class DslConsistencySpec extends AnyWordSpec with Matchers {
"concatGraph",
"prependGraph",
"alsoToGraph",
+ "resilientAlsoToGraph",
"wireTapGraph",
"orElseGraph",
"divertToGraph",
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowAlsoToSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowAlsoToSpec.scala
new file mode 100644
index 0000000000..c78b25ebb8
--- /dev/null
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowAlsoToSpec.scala
@@ -0,0 +1,288 @@
+/*
+ * 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 org.apache.pekko.stream.scaladsl
+
+import org.apache.pekko
+import pekko.stream.testkit._
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.stream.testkit.scaladsl.TestSource
+
+import scala.concurrent.duration._
+
+class FlowAlsoToSpec extends StreamSpec("""
+ pekko.stream.materializer.initial-input-buffer-size = 2
+ """) {
+
+ "alsoTo with propagateCancellation=true (default)" must {
+
+ "cancel the stream when side sink cancels" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink).runWith(mainSink)
+
+ mainProbe.request(2)
+ sideProbe.request(2)
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+ sideProbe.expectNext(1)
+
+ sideProbe.cancel()
+ pub.expectCancellation()
+ }
+
+ "forward elements to both downstreams" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink).runWith(mainSink)
+
+ mainProbe.request(3)
+ sideProbe.request(3)
+
+ pub.sendNext(1)
+ pub.sendNext(2)
+ pub.sendNext(3)
+
+ mainProbe.expectNext(1, 2, 3)
+ sideProbe.expectNext(1, 2, 3)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ sideProbe.expectComplete()
+ }
+ }
+
+ "alsoTo with propagateCancellation=false" must {
+
+ "continue main stream when side sink cancels" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(4)
+ sideProbe.request(2)
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+ sideProbe.expectNext(1)
+
+ pub.sendNext(2)
+ mainProbe.expectNext(2)
+ sideProbe.expectNext(2)
+
+ sideProbe.cancel()
+
+ pub.sendNext(3)
+ mainProbe.expectNext(3)
+
+ pub.sendNext(4)
+ mainProbe.expectNext(4)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ }
+
+ "continue main stream when side sink fails" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ val failingSideSink = Flow[Int].map { elem =>
+ if (elem == 1) throw new RuntimeException("side sink failure")
+ elem
+ }.to(sideSink)
+
+ src.alsoTo(failingSideSink, propagateCancellation =
false).runWith(mainSink)
+
+ mainProbe.request(3)
+ sideProbe.request(3)
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+
+ pub.sendNext(2)
+ mainProbe.expectNext(2)
+
+ pub.sendNext(3)
+ mainProbe.expectNext(3)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ }
+
+ "cancel side sink when main downstream cancels" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(1)
+ sideProbe.request(1)
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+ sideProbe.expectNext(1)
+
+ mainProbe.cancel()
+ pub.expectCancellation()
+ }
+
+ "forward elements to both downstreams before side cancels" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(3)
+ sideProbe.request(3)
+
+ pub.sendNext(1)
+ pub.sendNext(2)
+ pub.sendNext(3)
+
+ mainProbe.expectNext(1, 2, 3)
+ sideProbe.expectNext(1, 2, 3)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ sideProbe.expectComplete()
+ }
+
+ "complete normally when upstream completes" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(2)
+ sideProbe.request(2)
+
+ pub.sendNext(1)
+ pub.sendNext(2)
+ pub.sendComplete()
+
+ mainProbe.expectNext(1, 2).expectComplete()
+ sideProbe.expectNext(1, 2).expectComplete()
+ }
+
+ "handle side sink cancelling before any element is emitted" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(2)
+ sideProbe.request(1)
+
+ sideProbe.cancel()
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+
+ pub.sendNext(2)
+ mainProbe.expectNext(2)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ }
+
+ "propagate upstream failure to both downstreams" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(1)
+ sideProbe.request(1)
+
+ val ex = new RuntimeException("upstream boom")
+ pub.sendError(ex)
+
+ mainProbe.expectError(ex)
+ sideProbe.expectError(ex)
+ }
+
+ "backpressure when side sink is slow" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(3)
+ sideProbe.request(1)
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+ sideProbe.expectNext(1)
+
+ sideProbe.request(1)
+
+ pub.sendNext(2)
+ mainProbe.expectNext(2)
+ sideProbe.expectNext(2)
+
+ sideProbe.request(1)
+
+ pub.sendNext(3)
+ mainProbe.expectNext(3)
+ sideProbe.expectNext(3)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ sideProbe.expectComplete()
+ }
+
+ "handle side sink cancelling while pending element exists" in {
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+ val (pub, src) = TestSource[Int]().preMaterialize()
+
+ src.alsoTo(sideSink, propagateCancellation = false).runWith(mainSink)
+
+ mainProbe.request(3)
+ sideProbe.request(1)
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+ sideProbe.expectNext(1)
+
+ pub.sendNext(2)
+
+ sideProbe.cancel()
+ mainProbe.expectNext(2)
+
+ pub.sendNext(3)
+ mainProbe.expectNext(3)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ }
+ }
+}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBroadcastSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBroadcastSpec.scala
index 5817f7a633..0b9b3c3884 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBroadcastSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBroadcastSpec.scala
@@ -301,6 +301,51 @@ class GraphBroadcastSpec extends StreamSpec("""
pMain.cancel()
pIn.expectCancellation()
}
+
+ "continue to remaining outputs when a non-eager output cancels" in {
+ val (pub, src) = TestSource[Int]().preMaterialize()
+ val (mainProbe, mainSink) = TestSink[Int]().preMaterialize()
+ val (sideProbe, sideSink) = TestSink[Int]().preMaterialize()
+
+ RunnableGraph
+ .fromGraph(GraphDSL.create() { implicit b =>
+ import GraphDSL.Implicits._
+ // eagerCancel=true but side output (index 1) is non-eager: its
cancellation
+ // must not tear down the stage. Main output (index 0) keeps flowing.
+ val bcast = b.add(Broadcast[Int](2, eagerCancel = true,
nonEagerCancelOutputs = Set(1)))
+ src ~> bcast.in
+ bcast.out(0) ~> mainSink
+ bcast.out(1) ~> sideSink
+ ClosedShape
+ })
+ .run()
+
+ mainProbe.request(3)
+ sideProbe.request(2)
+
+ pub.sendNext(1)
+ mainProbe.expectNext(1)
+ sideProbe.expectNext(1)
+
+ pub.sendNext(2)
+ mainProbe.expectNext(2)
+ sideProbe.expectNext(2)
+
+ sideProbe.cancel()
+
+ pub.sendNext(3)
+ mainProbe.expectNext(3)
+
+ pub.sendComplete()
+ mainProbe.expectComplete()
+ }
+
+ "reject nonEagerCancelOutputs when eagerCancel is false" in {
+ val ex = intercept[IllegalArgumentException] {
+ Broadcast[Int](2, eagerCancel = false, nonEagerCancelOutputs = Set(1))
+ }
+ ex.getMessage should include("requires eagerCancel=true")
+ }
}
}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
index 50c8e2a0e2..525a0e25a7 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
@@ -3322,6 +3322,30 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
def alsoTo(that: Graph[SinkShape[Out], ?]): javadsl.Flow[In, Out, Mat] =
new Flow(delegate.alsoTo(that))
+ /**
+ * Attaches the given [[Sink]] to this [[Flow]], meaning that elements that
pass
+ * through will also be sent to the [[Sink]].
+ *
+ * When `propagateCancellation` is `false`, cancellation or failure of the
side [[Sink]]
+ * will not cancel the main stream. Elements will continue to flow to the
main downstream only.
+ *
+ * When `propagateCancellation` is `true` (the default), this behaves
identically to [[#alsoTo]].
+ *
+ * '''Emits when''' element is available and demand exists both from the
Sink and the downstream.
+ *
+ * '''Backpressures when''' downstream or Sink backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels (the side [[Sink]] is also
cancelled).
+ * When `propagateCancellation` is `true`,
cancellation or failure of
+ * the side [[Sink]] also cancels the downstream.
+ *
+ * @since 2.0.0
+ */
+ def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation: Boolean):
javadsl.Flow[In, Out, Mat] =
+ new Flow(delegate.alsoTo(that, propagateCancellation))
+
/**
* Attaches the given [[Sink]]s to this [[Flow]], meaning that elements that
passes
* through will also be sent to all those [[Sink]]s.
@@ -3357,6 +3381,23 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
matF: function.Function2[Mat, M2, M3]): javadsl.Flow[In, Out, M3] =
new Flow(delegate.alsoToMat(that)(combinerToScala(matF)))
+ /**
+ * Attaches the given [[Sink]] to this [[Flow]], meaning that elements that
pass
+ * through will also be sent to the [[Sink]].
+ *
+ * @see [[#alsoTo]]
+ *
+ * It is recommended to use the internally optimized `Keep.left` and
`Keep.right` combiners
+ * where appropriate instead of manually writing functions that pass through
one of the values.
+ *
+ * @since 2.0.0
+ */
+ def alsoToMat[M2, M3](
+ that: Graph[SinkShape[Out], M2],
+ propagateCancellation: Boolean,
+ matF: function.Function2[Mat, M2, M3]): javadsl.Flow[In, Out, M3] =
+ new Flow(delegate.alsoToMat(that,
propagateCancellation)(combinerToScala(matF)))
+
/**
* Attaches the given [[Sink]] to this [[Flow]], meaning that elements will
be sent to the [[Sink]]
* instead of being passed through if the predicate `when` returns `true`.
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala
index 4af7893d3f..f897bde130 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala
@@ -209,6 +209,27 @@ object Broadcast {
def create[T](outputCount: Int, eagerCancel: Boolean):
Graph[UniformFanOutShape[T, T], NotUsed] =
scaladsl.Broadcast(outputCount, eagerCancel = eagerCancel)
+ /**
+ * Create a new `Broadcast` operator with the specified input type and
per-output cancellation behavior.
+ *
+ * @param outputCount number of output ports
+ * @param eagerCancel if true, broadcast cancels upstream if any of its
downstreams cancel
+ * (except those listed in `nonEagerCancelOutputs`).
+ * @param nonEagerCancelOutputs set of output port indices whose
cancellation should not
+ * trigger upstream cancellation. These outputs
are silently
+ * removed from the broadcast when they cancel,
and elements
+ * continue flowing to the remaining outputs.
+ * @since 2.0.0
+ */
+ def create[T](
+ outputCount: Int,
+ eagerCancel: Boolean,
+ nonEagerCancelOutputs: util.Set[java.lang.Integer]):
Graph[UniformFanOutShape[T, T], NotUsed] =
+ scaladsl.Broadcast(
+ outputCount,
+ eagerCancel = eagerCancel,
+ nonEagerCancelOutputs =
nonEagerCancelOutputs.asScala.iterator.map(_.intValue()).toSet)
+
/**
* Create a new `Broadcast` operator with the specified input type.
*
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 4c8923a6fe..9893ffdcd9 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
@@ -1412,6 +1412,30 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
def alsoTo(that: Graph[SinkShape[Out], ?]): javadsl.Source[Out, Mat] =
new Source(delegate.alsoTo(that))
+ /**
+ * Attaches the given [[Sink]] to this [[Source]], meaning that elements
that pass
+ * through will also be sent to the [[Sink]].
+ *
+ * When `propagateCancellation` is `false`, cancellation or failure of the
side [[Sink]]
+ * will not cancel the main stream. Elements will continue to flow to the
main downstream only.
+ *
+ * When `propagateCancellation` is `true` (the default), this behaves
identically to [[#alsoTo]].
+ *
+ * '''Emits when''' element is available and demand exists both from the
Sink and the downstream.
+ *
+ * '''Backpressures when''' downstream or Sink backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels (the side [[Sink]] is also
cancelled).
+ * When `propagateCancellation` is `true`,
cancellation or failure of
+ * the side [[Sink]] also cancels the downstream.
+ *
+ * @since 2.0.0
+ */
+ def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation: Boolean):
javadsl.Source[Out, Mat] =
+ new Source(delegate.alsoTo(that, propagateCancellation))
+
/**
* Attaches the given [[Sink]]s to this [[Source]], meaning that elements
that passes
* through will also be sent to all those [[Sink]]s.
@@ -1447,6 +1471,23 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
matF: function.Function2[Mat, M2, M3]): javadsl.Source[Out, M3] =
new Source(delegate.alsoToMat(that)(combinerToScala(matF)))
+ /**
+ * Attaches the given [[Sink]] to this [[Source]], meaning that elements
that pass
+ * through will also be sent to the [[Sink]].
+ *
+ * @see [[#alsoTo]]
+ *
+ * It is recommended to use the internally optimized `Keep.left` and
`Keep.right` combiners
+ * where appropriate instead of manually writing functions that pass through
one of the values.
+ *
+ * @since 2.0.0
+ */
+ def alsoToMat[M2, M3](
+ that: Graph[SinkShape[Out], M2],
+ propagateCancellation: Boolean,
+ matF: function.Function2[Mat, M2, M3]): javadsl.Source[Out, M3] =
+ new Source(delegate.alsoToMat(that,
propagateCancellation)(combinerToScala(matF)))
+
/**
* Attaches the given [[Sink]] to this [[Flow]], meaning that elements will
be sent to the [[Sink]]
* instead of being passed through if the predicate `when` returns `true`.
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
index 189ad343a7..f06f63d10c 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
@@ -2329,6 +2329,30 @@ final class SubFlow[In, Out, Mat](
def alsoTo(that: Graph[SinkShape[Out], ?]): SubFlow[In, Out, Mat] =
new SubFlow(delegate.alsoTo(that))
+ /**
+ * Attaches the given [[Sink]] to this [[Flow]], meaning that elements that
pass
+ * through will also be sent to the [[Sink]].
+ *
+ * When `propagateCancellation` is `false`, cancellation or failure of the
side [[Sink]]
+ * will not cancel the main stream. Elements will continue to flow to the
main downstream only.
+ *
+ * When `propagateCancellation` is `true` (the default), this behaves
identically to [[#alsoTo]].
+ *
+ * '''Emits when''' element is available and demand exists both from the
Sink and the downstream.
+ *
+ * '''Backpressures when''' downstream or Sink backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels (the side [[Sink]] is also
cancelled).
+ * When `propagateCancellation` is `true`,
cancellation or failure of
+ * the side [[Sink]] also cancels the downstream.
+ *
+ * @since 2.0.0
+ */
+ def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation: Boolean):
SubFlow[In, Out, Mat] =
+ new SubFlow(delegate.alsoTo(that, propagateCancellation))
+
/**
* Attaches the given [[Sink]]s to this [[Flow]], meaning that elements that
passes
* through will also be sent to all those [[Sink]]s.
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
index ba572b0a86..da703a4562 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
@@ -2295,6 +2295,30 @@ final class SubSource[Out, Mat](
def alsoTo(that: Graph[SinkShape[Out], ?]): SubSource[Out, Mat] =
new SubSource(delegate.alsoTo(that))
+ /**
+ * Attaches the given [[Sink]] to this [[Source]], meaning that elements
that pass
+ * through will also be sent to the [[Sink]].
+ *
+ * When `propagateCancellation` is `false`, cancellation or failure of the
side [[Sink]]
+ * will not cancel the main stream. Elements will continue to flow to the
main downstream only.
+ *
+ * When `propagateCancellation` is `true` (the default), this behaves
identically to [[#alsoTo]].
+ *
+ * '''Emits when''' element is available and demand exists both from the
Sink and the downstream.
+ *
+ * '''Backpressures when''' downstream or Sink backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels (the side [[Sink]] is also
cancelled).
+ * When `propagateCancellation` is `true`,
cancellation or failure of
+ * the side [[Sink]] also cancels the downstream.
+ *
+ * @since 2.0.0
+ */
+ def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation: Boolean):
SubSource[Out, Mat] =
+ new SubSource(delegate.alsoTo(that, propagateCancellation))
+
/**
* Attaches the given [[Sink]]s to this [[Source]], meaning that elements
that passes
* through will also be sent to all those [[Sink]]s.
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
index 322108cd63..f9addecd4a 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
@@ -4067,6 +4067,44 @@ trait FlowOps[+Out, +Mat] {
FlowShape(bcast.in, bcast.out(0))
}
+ /**
+ * Attaches the given [[Sink]] to this [[Flow]], meaning that elements that
pass
+ * through will also be sent to the [[Sink]].
+ *
+ * When `propagateCancellation` is `false`, cancellation or failure of the
side [[Sink]]
+ * will not cancel the main stream. Elements will continue to flow to the
main downstream
+ * only. This is useful for fire-and-forget side sinks (e.g. logging) where
the side sink's
+ * availability should not affect the main business stream.
+ *
+ * When `propagateCancellation` is `true` (the default), this behaves
identically to [[#alsoTo]].
+ *
+ * It is similar to [[#wireTap]] but will backpressure instead of dropping
elements when the given [[Sink]] is not ready.
+ *
+ * '''Emits when''' element is available and demand exists both from the
Sink and the downstream.
+ *
+ * '''Backpressures when''' downstream or Sink backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels (the side [[Sink]] is also
cancelled).
+ * When `propagateCancellation` is `true`,
cancellation or failure of
+ * the side [[Sink]] also cancels the downstream.
+ *
+ * @since 2.0.0
+ */
+ def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation: Boolean):
Repr[Out] =
+ if (propagateCancellation) alsoTo(that)
+ else via(resilientAlsoToGraph(that))
+
+ protected def resilientAlsoToGraph[M](
+ that: Graph[SinkShape[Out], M]): Graph[FlowShape[Out @uncheckedVariance,
Out], M] =
+ GraphDSL.createGraph(that) { implicit b => r =>
+ import GraphDSL.Implicits._
+ val bcast = b.add(Broadcast[Out](2, eagerCancel = true,
nonEagerCancelOutputs = Set(1)))
+ bcast.out(1) ~> r
+ FlowShape(bcast.in, bcast.out(0))
+ }
+
/**
* Attaches the given [[Sink]]s to this [[Source]], meaning that elements
that pass
* through will also be sent to the [[Sink]].
@@ -4591,6 +4629,22 @@ trait FlowOpsMat[+Out, +Mat] extends FlowOps[Out, Mat] {
def alsoToMat[Mat2, Mat3](that: Graph[SinkShape[Out], Mat2])(matF: (Mat,
Mat2) => Mat3): ReprMat[Out, Mat3] =
viaMat(alsoToGraph(that))(matF)
+ /**
+ * Attaches the given [[Sink]] to this [[Flow]], meaning that elements that
pass
+ * through will also be sent to the [[Sink]].
+ *
+ * @see [[#alsoTo]]
+ *
+ * It is recommended to use the internally optimized `Keep.left` and
`Keep.right` combiners
+ * where appropriate instead of manually writing functions that pass through
one of the values.
+ *
+ * @since 2.0.0
+ */
+ def alsoToMat[Mat2, Mat3](that: Graph[SinkShape[Out], Mat2],
propagateCancellation: Boolean)(
+ matF: (Mat, Mat2) => Mat3): ReprMat[Out, Mat3] =
+ if (propagateCancellation) viaMat(alsoToGraph(that))(matF)
+ else viaMat(resilientAlsoToGraph(that))(matF)
+
/**
* Attaches the given [[Sink]] to this [[Flow]], meaning that elements will
be sent to the [[Sink]]
* instead of being passed through if the predicate `when` returns `true`.
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContext.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContext.scala
index fd7c1eafb6..40cf9ce25f 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContext.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContext.scala
@@ -149,6 +149,10 @@ final class FlowWithContext[-In, -CtxIn, +Out, +CtxOut,
+Mat](delegate: Flow[(In
override def alsoTo(that: Graph[SinkShape[Out], ?]): Repr[Out, CtxOut] =
FlowWithContext.fromTuples(delegate.alsoTo(Sink.contramapImpl(that, (in:
(Out, CtxOut)) => in._1)))
+ override def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation:
Boolean): Repr[Out, CtxOut] =
+ FlowWithContext.fromTuples(
+ delegate.alsoTo(Sink.contramapImpl(that, (in: (Out, CtxOut)) => in._1),
propagateCancellation))
+
override def alsoToContext(that: Graph[SinkShape[CtxOut], ?]): Repr[Out,
CtxOut] =
FlowWithContext.fromTuples(delegate.alsoTo(Sink.contramapImpl(that, (in:
(Out, CtxOut)) => in._2)))
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContextOps.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContextOps.scala
index 3a1193cf86..ac034481fc 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContextOps.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/FlowWithContextOps.scala
@@ -95,6 +95,14 @@ trait FlowWithContextOps[+Out, +Ctx, +Mat] {
*/
def alsoTo(that: Graph[SinkShape[Out], ?]): Repr[Out, Ctx]
+ /**
+ * Data variant of [[pekko.stream.scaladsl.FlowOps.alsoTo]] with
configurable cancellation propagation.
+ *
+ * @see [[pekko.stream.scaladsl.FlowOps.alsoTo]]
+ * @since 2.0.0
+ */
+ def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation: Boolean):
Repr[Out, Ctx]
+
/**
* Context variant of [[pekko.stream.scaladsl.FlowOps.alsoTo]]
*
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
index 415af7ea12..3136297a0d 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
@@ -609,6 +609,25 @@ object Broadcast {
*/
def apply[T](outputPorts: Int, eagerCancel: Boolean = false): Broadcast[T] =
new Broadcast(outputPorts, eagerCancel)
+
+ /**
+ * Create a new `Broadcast` with the specified number of output ports and
per-output
+ * cancellation behavior.
+ *
+ * @param outputPorts number of output ports
+ * @param eagerCancel if true, broadcast cancels upstream if any of its
downstreams cancel
+ * (except those listed in `nonEagerCancelOutputs`).
+ * @param nonEagerCancelOutputs set of output port indices whose
cancellation should not
+ * trigger upstream cancellation. These outputs
are silently
+ * removed from the broadcast when they cancel,
and elements
+ * continue flowing to the remaining outputs.
+ * @since 2.0.0
+ */
+ def apply[T](
+ outputPorts: Int,
+ eagerCancel: Boolean,
+ nonEagerCancelOutputs: Set[Int]): Broadcast[T] =
+ new Broadcast(outputPorts, eagerCancel, nonEagerCancelOutputs)
}
/**
@@ -627,8 +646,46 @@ object Broadcast {
final class Broadcast[T](val outputPorts: Int, val eagerCancel: Boolean)
extends GraphStage[UniformFanOutShape[T, T]]
with pekko.stream.TypePreservingFanOut {
- // one output might seem counter intuitive but saves us from special
handling in other places
require(outputPorts >= 1, "A Broadcast must have one or more output ports")
+
+ // Backing field for the 3-arg auxiliary constructor; default is empty (all
outputs eager).
+ // The auxiliary constructor assigns this exactly once during construction,
so the value is
+ // effectively immutable after init. Kept `var` because Scala auxiliary
constructors cannot
+ // assign `val` fields.
+ private var _nonEagerCancelOutputs: Set[Int] = Set.empty
+
+ /**
+ * Set of output port indices whose cancellation should not trigger upstream
cancellation.
+ * Only meaningful when `eagerCancel` is `true`; these outputs are silently
removed from the
+ * broadcast when they cancel, and elements continue flowing to the
remaining outputs.
+ *
+ * @since 2.0.0
+ */
+ def nonEagerCancelOutputs: Set[Int] = _nonEagerCancelOutputs
+
+ /**
+ * Create a new `Broadcast` with per-output cancellation behavior.
+ *
+ * @param nonEagerCancelOutputs set of output port indices whose
cancellation should not
+ * trigger upstream cancellation when
`eagerCancel` is `true`.
+ * These outputs are silently removed from the
broadcast when
+ * they cancel, and elements continue flowing
to the remaining
+ * outputs. Must be empty when `eagerCancel` is
`false` (where
+ * the parameter would be meaningless).
+ * @since 2.0.0
+ */
+ def this(outputPorts: Int, eagerCancel: Boolean, nonEagerCancelOutputs:
Set[Int]) = {
+ this(outputPorts, eagerCancel)
+ require(
+ eagerCancel || nonEagerCancelOutputs.isEmpty,
+ s"nonEagerCancelOutputs [$nonEagerCancelOutputs] requires
eagerCancel=true; " +
+ s"with eagerCancel=false all outputs already behave non-eagerly")
+ require(
+ nonEagerCancelOutputs.subsetOf((0 until outputPorts).toSet),
+ s"nonEagerCancelOutputs [$nonEagerCancelOutputs] must be a subset of
[${0 until outputPorts}]")
+ _nonEagerCancelOutputs = nonEagerCancelOutputs
+ }
+
val in: Inlet[T] = Inlet[T]("Broadcast.in")
val out: immutable.IndexedSeq[Outlet[T]] = Vector.tabulate(outputPorts)(i =>
Outlet[T]("Broadcast.out" + i))
override def initialAttributes = DefaultAttributes.broadcast
@@ -677,7 +734,7 @@ final class Broadcast[T](val outputPorts: Int, val
eagerCancel: Boolean)
}
override def onDownstreamFinish(cause: Throwable) = {
- if (eagerCancel) cancelStage(cause)
+ if (eagerCancel && !nonEagerCancelOutputs.contains(i))
cancelStage(cause)
else {
downstreamsRunning -= 1
if (downstreamsRunning == 0) cancelStage(cause)
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala
index 4890b40efe..e97b206696 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala
@@ -166,6 +166,10 @@ final class SourceWithContext[+Out, +Ctx, +Mat]
private[stream] (delegate: Sourc
override def alsoTo(that: Graph[SinkShape[Out], ?]): Repr[Out, Ctx] =
SourceWithContext.fromTuples(delegate.alsoTo(Sink.contramapImpl(that, (in:
(Out, Ctx)) => in._1)))
+ override def alsoTo(that: Graph[SinkShape[Out], ?], propagateCancellation:
Boolean): Repr[Out, Ctx] =
+ SourceWithContext.fromTuples(
+ delegate.alsoTo(Sink.contramapImpl(that, (in: (Out, Ctx)) => in._1),
propagateCancellation))
+
override def alsoToContext(that: Graph[SinkShape[Ctx], ?]): Repr[Out, Ctx] =
SourceWithContext.fromTuples(delegate.alsoTo(Sink.contramapImpl(that, (in:
(Out, Ctx)) => in._2)))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
