This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch feat/alsoTo-propagateCancellation-3104 in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 326d4d64cff697b42c807bb48ff9fcab171ddc30 Author: 虎鸣 <[email protected]> AuthorDate: Tue Jun 23 01:36:37 2026 +0800 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 --- project/StreamOperatorsIndexGenerator.scala | 1 + .../apache/pekko/stream/DslConsistencySpec.scala | 1 + .../pekko/stream/scaladsl/FlowAlsoToSpec.scala | 288 +++++++++++++++++++++ .../org/apache/pekko/stream/impl/Stages.scala | 1 + .../org/apache/pekko/stream/javadsl/Flow.scala | 41 +++ .../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 | 106 +++++++- .../pekko/stream/scaladsl/SourceWithContext.scala | 4 + 13 files changed, 596 insertions(+), 1 deletion(-) 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/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/src/main/scala/org/apache/pekko/stream/impl/Stages.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala index 78584949e4..41a46bb8a3 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala @@ -103,6 +103,7 @@ import pekko.stream.Attributes._ val onErrorComplete = name("onErrorComplete") val broadcast = name("broadcast") val wireTap = name("wireTap") + val resilientAlsoTo = name("resilientAlsoTo") val balance = name("balance") val zip = name("zip") val zipLatest = name("zipLatest") 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 5a889b6189..39a71e8674 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 @@ -3282,6 +3282,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 1.2.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. @@ -3317,6 +3341,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 1.2.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/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala index e60a445e62..5a0ede7f6f 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 @@ -1406,6 +1406,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 1.2.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. @@ -1441,6 +1465,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 1.2.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 0082bf73d5..6268b278e8 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 @@ -2289,6 +2289,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 1.2.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 18ad5712bc..4c57e58b45 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 @@ -2255,6 +2255,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 1.2.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 4ae16e46ce..16501af7a6 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 @@ -4019,6 +4019,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 1.2.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 stage = b.add(new ResilientAlsoTo[Out]) + stage.out1 ~> r + FlowShape(stage.in, stage.out0) + } + /** * Attaches the given [[Sink]]s to this [[Source]], meaning that elements that pass * through will also be sent to the [[Sink]]. @@ -4536,6 +4574,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 1.2.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 9fafa15e64..eaef4cea2f 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 @@ -94,6 +94,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 1.2.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 6cde3ffa9f..58671bb7b9 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 @@ -30,7 +30,7 @@ import pekko.stream.impl._ import pekko.stream.impl.Stages.DefaultAttributes import pekko.stream.impl.fusing.GraphStages import pekko.stream.scaladsl.Partition.PartitionOutOfBoundsException -import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler } +import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler, StageLogging } import pekko.util.ConstantFun /** @@ -781,6 +781,110 @@ private[stream] final class WireTap[T] extends GraphStage[FanOutShape2[T, T, T]] override def toString = "WireTap" } +/** + * INTERNAL API + * + * A `Broadcast`-like stage with two outputs where cancellation of the side output (out1) + * does not cancel the stage. Elements continue flowing to the main output (out0) after + * the side output cancels or fails. Cancellation of the main output cancels the stage. + * + * Backpressures when either output backpressures (same contract as `alsoTo` / `Broadcast`). + */ +private[pekko] final class ResilientAlsoTo[T] extends GraphStage[FanOutShape2[T, T, T]] { + val in: Inlet[T] = Inlet[T]("ResilientAlsoTo.in") + val outMain: Outlet[T] = Outlet[T]("ResilientAlsoTo.outMain") + val outSide: Outlet[T] = Outlet[T]("ResilientAlsoTo.outSide") + override def initialAttributes: Attributes = DefaultAttributes.resilientAlsoTo + override val shape: FanOutShape2[T, T, T] = new FanOutShape2(in, outMain, outSide) + + override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = + new GraphStageLogic(shape) with InHandler with OutHandler with StageLogging { + + private var pendingElement: Option[T] = None + private var mainReady = false + private var sideReady = false + + override def onPush(): Unit = { + val elem = grab(in) + if (isClosed(outSide)) { + push(outMain, elem) + } else if (mainReady && sideReady) { + push(outMain, elem) + push(outSide, elem) + mainReady = false + sideReady = false + } else { + pendingElement = Some(elem) + } + } + + override def onPull(): Unit = { + mainReady = true + tryPushAndPull() + } + + override def onDownstreamFinish(cause: Throwable): Unit = + cancelStage(cause) + + private def tryPushAndPull(): Unit = { + pendingElement match { + case Some(elem) => + if (isClosed(outSide)) { + push(outMain, elem) + pendingElement = None + mainReady = false + } else if (mainReady && sideReady) { + push(outMain, elem) + push(outSide, elem) + pendingElement = None + mainReady = false + sideReady = false + } + case None => + if (mainReady && (sideReady || isClosed(outSide)) && !hasBeenPulled(in)) + pull(in) + } + } + + setHandler( + outSide, + new OutHandler { + override def onPull(): Unit = { + sideReady = true + tryPushAndPull() + } + + override def onDownstreamFinish(cause: Throwable): Unit = { + if (!cause.isInstanceOf[SubscriptionWithCancelException.NonFailureCancellation]) + log.warning("ResilientAlsoTo: side sink failed, continuing main stream: {}", cause.getMessage) + else + log.debug("ResilientAlsoTo: side sink cancelled") + pendingElement match { + case Some(elem) => + if (mainReady) { + push(outMain, elem) + mainReady = false + } + pendingElement = None + case None => + } + sideReady = false + setHandler(in, + new InHandler { + override def onPush(): Unit = + push(outMain, grab(in)) + }) + if (mainReady && !hasBeenPulled(in)) + pull(in) + } + }) + + setHandlers(in, outMain, this) + } + + override def toString = "ResilientAlsoTo" +} + object Partition { // FIXME make `PartitionOutOfBoundsException` a `final` class when possible case class PartitionOutOfBoundsException(msg: String) extends IndexOutOfBoundsException(msg) with NoStackTrace 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]
