This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko.git
The following commit(s) were added to refs/heads/main by this push:
new b818925edc Replace SubstreamCancelStrategy with SupervisionDecider for
Split
b818925edc is described below
commit b818925edc28f14939a10b7c468b3a7bdeb3946b
Author: Matthew de Detrich <[email protected]>
AuthorDate: Sat Mar 18 12:15:46 2023 +0100
Replace SubstreamCancelStrategy with SupervisionDecider for Split
---
.../paradox/project/migration-guide-1.0.x-1.1.x.md | 72 ++++++++++++++++++++++
docs/src/main/paradox/project/migration-guides.md | 10 ++-
.../stream/operators/Source-or-Flow/splitAfter.md | 6 +-
.../stream/operators/Source-or-Flow/splitWhen.md | 6 +-
.../pekko/stream/scaladsl/FlowSplitAfterSpec.scala | 15 +++--
.../pekko/stream/scaladsl/FlowSplitWhenSpec.scala | 36 ++++++++---
.../pekko/stream/SubstreamCancelStrategy.scala | 10 +++
.../pekko/stream/impl/fusing/StreamOfStreams.scala | 46 ++++++++------
.../org/apache/pekko/stream/javadsl/Flow.scala | 6 ++
.../org/apache/pekko/stream/javadsl/Source.scala | 6 ++
.../org/apache/pekko/stream/scaladsl/Flow.scala | 56 ++++++++++++++---
11 files changed, 218 insertions(+), 51 deletions(-)
diff --git a/docs/src/main/paradox/project/migration-guide-1.0.x-1.1.x.md
b/docs/src/main/paradox/project/migration-guide-1.0.x-1.1.x.md
new file mode 100644
index 0000000000..7e7bd630c9
--- /dev/null
+++ b/docs/src/main/paradox/project/migration-guide-1.0.x-1.1.x.md
@@ -0,0 +1,72 @@
+---
+project.description: Migrating to Apache Pekko 1.1.
+---
+# Migration Guide 1.0.x to 1.1.x
+
+Apache Pekko 1.1.x is binary backwards compatible with 1.0.x with the ordinary
exceptions listed in the
+@ref:[Binary Compatibility Rules](../common/binary-compatibility-rules.md).
+
+This means that updating an application from Apache Pekko 1.0.x to 1.1.x
should be a smooth process, and
+that libraries built for Apache Pekko 1.0.x can also be used with Apache Pekko
1.1.x. For example
+Apache Pekko HTTP 1.0.0 and Apache Pekko Management 1.0.3 can be used with
Apache Pekko 1.1.x dependencies.
+You may have to add explicit dependencies to the new Apache Pekko version in
your build.
+
+That said, there are some changes to configuration and behavior that should be
considered, so
+reading this migration guide and testing your application thoroughly is
recommended.
+
+## `Source.splitWhen`/`Flow.splitWhen`/`Source.splitAfter`/`Flow.splitAfter`
behavioural change
+
+In Apache Pekko 1.0.x, the `splitWhen`/`splitAfter` stream operators on
`Source`/`Flow` had an optional
+`SubstreamCancelStrategy` parameter which defaulted to
`SubstreamCancelStrategy.drain`. In Apache Pekko
+1.1.x the usage of `SubstreamCancelStrategy` has been deprecated and instead
`splitWhen`/`splitAfter`
+inherits the already existing `Supervision` strategy `Attribute` to achieve
the same effect with the
+following translation
+
+* `Supervision.resume` behaves the same way as `SubstreamCancelStrategy.drain`
+* `Supervision.stop` behaves the same way as
`SubstreamCancelStrategy.propagate`
+* `Supervision.restart` doesn't have an equivalent `SubstreamCancelStrategy`.
Since it doesn't make semantic
+sense for `SubFlow`s it behaves the same way as `Supervision.stop`.
+
+The potential behavioural change results from the fact that in Apache Pekko
Streams the default `Supervision`
+strategy is `Supervision.stop` whereas the default `SubstreamCancelStrategy`
for Apache Pekko 1.0.x is
+`SubstreamCancelStrategy.drain` (which translates to `Supervision.resume` in
Apache Pekko 1.1.x). This means
+that when you upgrade from Apache Pekko 1.0.x to 1.1.x its possible that
previously if `SubFlow`s
+resulted in errors the parent stream would continue to operate whereas in
Apache Pekko 1.1.x the stream would cancel.
+
+If you would like to keep the previous Apache Pekko 1.0.x behaviour you can
just specify the
+`Supervision.resumingDecider` attribute on your stream in the standard way,
i.e. you would change
+
+@@@div { .group-java }
+```java
+source.splitAfter(somePredicate);
+```
+@@@
+
+@@@div { .group-scala }
+```scala
+source.splitAfter(somePredicate)
+```
+@@@
+
+to
+
+@@@div { .group-java }
+```java
+source
+ .splitAfter(somePredicate)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider));
+```
+@@@
+
+@@@div { .group-scala }
+```scala
+source
+ .splitAfter(_ == somePredicate)
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+```
+@@@
+
+If you already happen to have already explicitly defined a
`SubstreamCancelStrategy` in the
+`splitWhen`/`splitAfter` operators then there won't be any behavioural change
albeit you
+will get deprecation warnings on compilation so its recommended to migrate
your code to use
+the equivalent `SupervisionStrategy` as described earlier.
diff --git a/docs/src/main/paradox/project/migration-guides.md
b/docs/src/main/paradox/project/migration-guides.md
index 7f03436dbd..ab204e82be 100644
--- a/docs/src/main/paradox/project/migration-guides.md
+++ b/docs/src/main/paradox/project/migration-guides.md
@@ -6,7 +6,7 @@ project.description: Apache Pekko version migration guides.
Apache Pekko is based on the latest version of Akka in the v2.6.x series. If
migrating from an earlier version of Akka,
please [migrate to Akka
2.6](https://doc.akka.io/docs/akka/current/project/migration-guides.html)
before migrating to Pekko.
-## Migration to Apache Pekko
+## Migration from Akka to Apache Pekko
These migration notes are designed for users migrating from Akka 2.6 to Pekko
1.0 and assume a basic level of Akka experience. Please feel free to submit an
issue or a patch if you feel like the notes can be improved.
@@ -44,3 +44,11 @@ These migration notes are designed for users migrating from
Akka 2.6 to Pekko 1.
It appears that data persisted with "akka-persistence" is usable with
"pekko-persistence" and vice versa (@ref:[Persistence](../persistence.md)).
We currently do not expect that Akka and Pekko nodes combined will be able to
form a @ref:[cluster](../cluster-usage.md).
+
+@@toc { depth=1 }
+
+@@@ index
+
+* [migration-guide-1.0.x-1.1.x](migration-guide-1.0.x-1.1.x.md)
+
+@@@
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/splitAfter.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/splitAfter.md
index 7ced7b40a4..7534422125 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/splitAfter.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/splitAfter.md
@@ -6,9 +6,11 @@ End the current substream whenever a predicate returns `true`,
starting a new su
## Signature
-@apidoc[Source.splitAfter](Source) {
scala="#splitAfter(substreamCancelStrategy:org.apache.pekko.stream.SubstreamCancelStrategy)(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitAfter(org.apache.pekko.stream.SubstreamCancelStrategy,org.apache.pekko.japi.function.Predicate)"
}
-@apidoc[Flow.splitAfter](Flow) {
scala="#splitAfter(substreamCancelStrategy:org.apache.pekko.stream.SubstreamCancelStrategy)(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitAfter(org.apache.pekko.stream.SubstreamCancelStrategy,org.apache.pekko.japi.function.Predicate)"
}
+@apidoc[Source.splitAfter](Source) {
scala="#splitAfter(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitAfter(org.apache.pekko.japi.function.Predicate)" }
+@apidoc[Flow.splitAfter](Flow) {
scala="#splitAfter(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitAfter(org.apache.pekko.japi.function.Predicate)" }
+The `splitAfter` operator adheres to the ActorAttributes.SupervisionStrategy
attribute with the caveat that
+`Supervision.restart` behaves the same way as `Supervision.resume` since
`Supervision.restart` for `SubFlow`s semantically doesn't make sense.
## Description
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/splitWhen.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/splitWhen.md
index 891b34dc6f..3b5dd5b9e7 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/splitWhen.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/splitWhen.md
@@ -6,9 +6,11 @@ Split off elements into a new substream whenever a predicate
function return `tr
## Signature
-@apidoc[Source.splitWhen](Source) {
scala="#splitWhen(substreamCancelStrategy:org.apache.pekko.stream.SubstreamCancelStrategy)(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitWhen(org.apache.pekko.stream.SubstreamCancelStrategy,org.apache.pekko.japi.function.Predicate)"
}
-@apidoc[Flow.splitWhen](Flow) {
scala="#splitWhen(substreamCancelStrategy:org.apache.pekko.stream.SubstreamCancelStrategy)(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitWhen(org.apache.pekko.stream.SubstreamCancelStrategy,org.apache.pekko.japi.function.Predicate)"
}
+@apidoc[Source.splitWhen](Source) {
scala="#splitWhen(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitWhen(org.apache.pekko.japi.function.Predicate)" }
+@apidoc[Flow.splitWhen](Flow) {
scala="#splitWhen(p:Out=>Boolean):org.apache.pekko.stream.scaladsl.SubFlow[Out,Mat,FlowOps.this.Repr,FlowOps.this.Closed]"
java="#splitWhen(org.apache.pekko.japi.function.Predicate)" }
+The `splitWhen` operator adheres to the ActorAttributes.SupervisionStrategy
attribute with the caveat that
+`Supervision.restart` behaves the same way as `Supervision.resume` since
`Supervision.restart` for `SubFlow`s semantically doesn't make sense.
## Description
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitAfterSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitAfterSpec.scala
index 1dc340e377..16b5e30675 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitAfterSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitAfterSpec.scala
@@ -22,7 +22,7 @@ import org.apache.pekko
import pekko.NotUsed
import pekko.stream._
import pekko.stream.StreamSubscriptionTimeoutTerminationMode
-import pekko.stream.Supervision.resumingDecider
+import pekko.stream.Supervision.{ resumingDecider, Decider }
import pekko.stream.impl.SubscriptionTimeoutException
import pekko.stream.testkit.StreamSpec
import pekko.stream.testkit.TestPublisher
@@ -58,13 +58,16 @@ class FlowSplitAfterSpec extends StreamSpec("""
def cancel(): Unit = subscription.cancel()
}
- class SubstreamsSupport(
- splitAfter: Int = 3,
+ class SubstreamsSupport(splitAfter: Int = 3,
elementCount: Int = 6,
- substreamCancelStrategy: SubstreamCancelStrategy =
SubstreamCancelStrategy.drain) {
+ decider: Decider = Supervision.resumingDecider) {
val source = Source(1 to elementCount)
- val groupStream = source.splitAfter(substreamCancelStrategy)(_ ==
splitAfter).lift.runWith(Sink.asPublisher(false))
+ val groupStream = source
+ .splitAfter(_ == splitAfter)
+ .lift
+ .withAttributes(ActorAttributes.supervisionStrategy(decider))
+ .runWith(Sink.asPublisher(false))
val masterSubscriber = TestSubscriber.manualProbe[Source[Int, _]]()
groupStream.subscribe(masterSubscriber)
@@ -272,7 +275,7 @@ class FlowSplitAfterSpec extends StreamSpec("""
}
"support eager cancellation of master stream on cancelling substreams" in {
- new SubstreamsSupport(splitAfter = 5, elementCount = 8,
SubstreamCancelStrategy.propagate) {
+ new SubstreamsSupport(splitAfter = 5, elementCount = 8,
Supervision.stoppingDecider) {
val s1 = StreamPuppet(expectSubFlow().runWith(Sink.asPublisher(false)))
s1.cancel()
masterSubscriber.expectComplete()
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitWhenSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitWhenSpec.scala
index d2a748725e..5eaaa1a9d3 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitWhenSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitWhenSpec.scala
@@ -23,7 +23,7 @@ import org.apache.pekko
import pekko.Done
import pekko.NotUsed
import pekko.stream._
-import pekko.stream.Supervision.resumingDecider
+import pekko.stream.Supervision.Decider
import pekko.stream.impl.SubscriptionTimeoutException
import pekko.stream.impl.fusing.Split
import pekko.stream.testkit._
@@ -51,13 +51,16 @@ class FlowSplitWhenSpec extends StreamSpec("""
def cancel(): Unit = subscription.cancel()
}
- class SubstreamsSupport(
- splitWhen: Int = 3,
+ class SubstreamsSupport(splitWhen: Int = 3,
elementCount: Int = 6,
- substreamCancelStrategy: SubstreamCancelStrategy =
SubstreamCancelStrategy.drain) {
+ decider: Decider = Supervision.resumingDecider) {
val source = Source(1 to elementCount)
- val groupStream = source.splitWhen(substreamCancelStrategy)(_ ==
splitWhen).lift.runWith(Sink.asPublisher(false))
+ val groupStream = source
+ .splitWhen(_ == splitWhen)
+ .lift
+ .withAttributes(ActorAttributes.supervisionStrategy(decider))
+ .runWith(Sink.asPublisher(false))
val masterSubscriber = TestSubscriber.manualProbe[Source[Int, _]]()
groupStream.subscribe(masterSubscriber)
@@ -162,6 +165,7 @@ class FlowSplitWhenSpec extends StreamSpec("""
.fromPublisher(inputs)
.splitWhen(_ == 2)
.lift
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
.map(_.runWith(Sink.fromSubscriber(substream)))
.runWith(Sink.fromSubscriber(masterStream))
@@ -176,7 +180,13 @@ class FlowSplitWhenSpec extends StreamSpec("""
inputs.expectCancellation()
val inputs2 = TestPublisher.probe[Int]()
- Source.fromPublisher(inputs2).splitWhen(_ ==
2).lift.map(_.runWith(Sink.cancelled)).runWith(Sink.cancelled)
+ Source
+ .fromPublisher(inputs2)
+ .splitWhen(_ == 2)
+ .lift
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .map(_.runWith(Sink.cancelled))
+ .runWith(Sink.cancelled)
inputs2.expectCancellation()
@@ -184,7 +194,12 @@ class FlowSplitWhenSpec extends StreamSpec("""
val masterStream3 = TestSubscriber.probe[Source[Int, Any]]()
- Source.fromPublisher(inputs3).splitWhen(_ ==
2).lift.runWith(Sink.fromSubscriber(masterStream3))
+ Source
+ .fromPublisher(inputs3)
+ .splitWhen(_ == 2)
+ .lift
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .runWith(Sink.fromSubscriber(masterStream3))
masterStream3.request(1)
inputs3.sendNext(1)
@@ -267,6 +282,7 @@ class FlowSplitWhenSpec extends StreamSpec("""
Source(1 to 100)
.splitWhen(_ => true)
.lift
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
.mapAsync(1)(_.runWith(Sink.head)) // Please note that this line
*also* implicitly asserts nonempty substreams
.grouped(200)
.runWith(Sink.head),
@@ -281,7 +297,7 @@ class FlowSplitWhenSpec extends StreamSpec("""
// `lift` doesn't cut here because it will prevent the behavior we'd
like to see.
// In fact, this test is somewhat useless, as a user cannot trigger
double materialization using
// the public splitWhen => SubFlow API.
- .via(Split.when(_ => true, SubstreamCancelStrategy.drain))
+ .via(Split.when(_ => true))
.map { source =>
// run twice, but make sure we return the result of the
materialization that ran second
source.runWith(Sink.ignore).flatMap(_ =>
source.runWith(Sink.ignore))
@@ -320,7 +336,7 @@ class FlowSplitWhenSpec extends StreamSpec("""
.fromPublisher(publisherProbeProbe)
.splitWhen(elem => if (elem == 3) throw exc else elem % 3 == 0)
.lift
- .withAttributes(ActorAttributes.supervisionStrategy(resumingDecider))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
.runWith(Sink.asPublisher(false))
val subscriber = TestSubscriber.manualProbe[Source[Int, NotUsed]]()
publisher.subscribe(subscriber)
@@ -374,7 +390,7 @@ class FlowSplitWhenSpec extends StreamSpec("""
}
"support eager cancellation of master stream on cancelling substreams" in {
- new SubstreamsSupport(splitWhen = 5, elementCount = 8,
SubstreamCancelStrategy.propagate) {
+ new SubstreamsSupport(splitWhen = 5, elementCount = 8,
Supervision.stoppingDecider) {
val s1 = StreamPuppet(getSubFlow().runWith(Sink.asPublisher(false)))
s1.cancel()
masterSubscriber.expectComplete()
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/SubstreamCancelStrategy.scala
b/stream/src/main/scala/org/apache/pekko/stream/SubstreamCancelStrategy.scala
index e260a0d280..563ee91f61 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/SubstreamCancelStrategy.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/SubstreamCancelStrategy.scala
@@ -15,9 +15,14 @@ package org.apache.pekko.stream
import SubstreamCancelStrategies._
+import scala.annotation.nowarn
+
/**
* Represents a strategy that decides how to deal with substream events.
*/
+@deprecated(
+ "Use .withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider))
rather than a SubstreamCancelStrategy",
+ since = "1.1.0")
sealed abstract class SubstreamCancelStrategy
private[pekko] object SubstreamCancelStrategies {
@@ -25,14 +30,19 @@ private[pekko] object SubstreamCancelStrategies {
/**
* INTERNAL API
*/
+ @nowarn("msg=deprecated")
private[pekko] case object Propagate extends SubstreamCancelStrategy
/**
* INTERNAL API
*/
+ @nowarn("msg=deprecated")
private[pekko] case object Drain extends SubstreamCancelStrategy
}
+@deprecated(
+ "Use .withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider))
rather than a SubstreamCancelStrategy",
+ since = "1.1.0")
object SubstreamCancelStrategy {
/**
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala
index 8e5bdc2b77..94abf377ef 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala
@@ -15,7 +15,7 @@ package org.apache.pekko.stream.impl.fusing
import java.util.Collections
import java.util.concurrent.atomic.AtomicReference
-import scala.annotation.tailrec
+import scala.annotation.{ nowarn, tailrec }
import scala.collection.immutable
import scala.concurrent.duration.FiniteDuration
import scala.util.control.NonFatal
@@ -27,6 +27,7 @@ import pekko.stream.ActorAttributes.StreamSubscriptionTimeout
import pekko.stream.ActorAttributes.SupervisionStrategy
import pekko.stream.Attributes.SourceLocation
import pekko.stream.impl.{ Buffer => BufferImpl }
+import pekko.stream.Supervision.Decider
import pekko.stream.impl.ActorSubscriberMessage
import pekko.stream.impl.ActorSubscriberMessage.OnError
import pekko.stream.impl.Stages.DefaultAttributes
@@ -467,24 +468,25 @@ import pekko.util.ccompat.JavaConverters._
/** Splits after the current element. The current element will be the last
element in the current substream. */
case object SplitAfter extends SplitDecision
- def when[T](
- p: T => Boolean,
- substreamCancelStrategy: SubstreamCancelStrategy): Graph[FlowShape[T,
Source[T, NotUsed]], NotUsed] =
- new Split(Split.SplitBefore, p, substreamCancelStrategy)
+ @nowarn("msg=deprecated")
+ def cancelStrategyToDecider(substreamCancelStrategy:
SubstreamCancelStrategy): Decider =
+ substreamCancelStrategy match {
+ case SubstreamCancelStrategies.Propagate => Supervision.stoppingDecider
+ case SubstreamCancelStrategies.Drain => Supervision.resumingDecider
+ }
+
+ def when[T](p: T => Boolean): Graph[FlowShape[T, Source[T, NotUsed]],
NotUsed] = {
+ new Split(Split.SplitBefore, p)
+ }
- def after[T](
- p: T => Boolean,
- substreamCancelStrategy: SubstreamCancelStrategy): Graph[FlowShape[T,
Source[T, NotUsed]], NotUsed] =
- new Split(Split.SplitAfter, p, substreamCancelStrategy)
+ def after[T](p: T => Boolean): Graph[FlowShape[T, Source[T, NotUsed]],
NotUsed] =
+ new Split(Split.SplitAfter, p)
}
/**
* INTERNAL API
*/
-@InternalApi private[pekko] final class Split[T](
- val decision: Split.SplitDecision,
- val p: T => Boolean,
- val substreamCancelStrategy: SubstreamCancelStrategy)
+@InternalApi private[pekko] final class Split[T](val decision:
Split.SplitDecision, val p: T => Boolean)
extends GraphStage[FlowShape[T, Source[T, NotUsed]]] {
val in: Inlet[T] = Inlet("Split.in")
@@ -492,24 +494,28 @@ import pekko.util.ccompat.JavaConverters._
override val shape: FlowShape[T, Source[T, NotUsed]] = FlowShape(in, out)
- private val propagateSubstreamCancel = substreamCancelStrategy match {
- case SubstreamCancelStrategies.Propagate => true
- case SubstreamCancelStrategies.Drain => false
- }
-
- override protected def initialAttributes: Attributes =
DefaultAttributes.split and SourceLocation.forLambda(p)
+ override def initialAttributes: Attributes = DefaultAttributes.split and
SourceLocation.forLambda(p)
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
new TimerGraphStageLogic(shape) {
import Split._
private val SubscriptionTimer = "SubstreamSubscriptionTimer"
+ private lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+
private val timeout: FiniteDuration =
inheritedAttributes.mandatoryAttribute[ActorAttributes.StreamSubscriptionTimeout].timeout
private var substreamSource: SubSourceOutlet[T] = null
private var substreamWaitingToBePushed = false
private var substreamCancelled = false
+ def propagateSubstreamCancel(ex: Throwable): Boolean =
+ decider(ex) match {
+ case Supervision.Stop => true
+ case Supervision.Resume => false
+ case Supervision.Restart => false
+ }
+
setHandler(
out,
new OutHandler {
@@ -606,7 +612,7 @@ import pekko.util.ccompat.JavaConverters._
override def onDownstreamFinish(cause: Throwable): Unit = {
substreamCancelled = true
- if (isClosed(in) || propagateSubstreamCancel) {
+ if (isClosed(in) || propagateSubstreamCancel(cause)) {
cancelStage(cause)
} else {
// Start draining
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 3e2330abf3..8aaf3a45ee 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
@@ -2414,6 +2414,9 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
*
* @see [[#splitWhen]]
*/
+ @deprecated(
+ "Use
.withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather
than a SubstreamCancelStrategy",
+ since = "1.1.0")
def splitWhen(substreamCancelStrategy: SubstreamCancelStrategy)(p:
function.Predicate[Out]): SubFlow[In, Out, Mat] =
new SubFlow(delegate.splitWhen(substreamCancelStrategy)(p.test))
@@ -2472,6 +2475,9 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
*
* @see [[#splitAfter]]
*/
+ @deprecated(
+ "Use
.withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather
than a SubstreamCancelStrategy",
+ since = "1.1.0")
def splitAfter(substreamCancelStrategy: SubstreamCancelStrategy)(p:
function.Predicate[Out]): SubFlow[In, Out, Mat] =
new SubFlow(delegate.splitAfter(substreamCancelStrategy)(p.test))
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 edaff38f1b..2e6db54589 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
@@ -3905,6 +3905,9 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
*
* @see [[#splitWhen]]
*/
+ @deprecated(
+ "Use
.withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather
than a SubstreamCancelStrategy",
+ since = "1.1.0")
def splitWhen(substreamCancelStrategy: SubstreamCancelStrategy)(p:
function.Predicate[Out]): SubSource[Out, Mat] =
new SubSource(delegate.splitWhen(substreamCancelStrategy)(p.test))
@@ -3962,6 +3965,9 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
*
* @see [[#splitAfter]]
*/
+ @deprecated(
+ "Use
.withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather
than a SubstreamCancelStrategy",
+ since = "1.1.0")
def splitAfter(substreamCancelStrategy: SubstreamCancelStrategy)(p:
function.Predicate[Out]): SubSource[Out, Mat] =
new SubSource(delegate.splitAfter(substreamCancelStrategy)(p.test))
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 a73822fec1..6e0f205c68 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
@@ -2338,15 +2338,25 @@ trait FlowOps[+Out, +Mat] {
*
* See also [[FlowOps.splitAfter]].
*/
+ @deprecated(
+ "Use
.withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather
than a SubstreamCancelStrategy",
+ since = "1.1.0")
def splitWhen(substreamCancelStrategy: SubstreamCancelStrategy)(
p: Out => Boolean): SubFlow[Out, Mat, Repr, Closed] = {
val merge = new SubFlowImpl.MergeBack[Out, Repr] {
override def apply[T](flow: Flow[Out, T, NotUsed], breadth: Int):
Repr[T] =
- via(Split.when(p, substreamCancelStrategy)).map(_.via(flow)).via(new
FlattenMerge(breadth))
+ via(Split
+ .when(p)
+
.withAttributes(ActorAttributes.supervisionStrategy(Split.cancelStrategyToDecider(substreamCancelStrategy))))
+ .map(_.via(flow))
+ .via(new FlattenMerge(breadth))
}
- val finish: (Sink[Out, NotUsed]) => Closed = s =>
- via(Split.when(p, substreamCancelStrategy))
+ val finish: Sink[Out, NotUsed] => Closed = s =>
+ via(
+ Split
+ .when(p)
+
.withAttributes(ActorAttributes.supervisionStrategy(Split.cancelStrategyToDecider(substreamCancelStrategy))))
.to(Sink.foreach(_.runWith(s)(GraphInterpreter.currentInterpreter.materializer)))
new SubFlowImpl(Flow[Out], merge, finish)
@@ -2359,8 +2369,17 @@ trait FlowOps[+Out, +Mat] {
*
* @see [[#splitWhen]]
*/
- def splitWhen(p: Out => Boolean): SubFlow[Out, Mat, Repr, Closed] =
- splitWhen(SubstreamCancelStrategy.drain)(p)
+ def splitWhen(p: Out => Boolean): SubFlow[Out, Mat, Repr, Closed] = {
+ val merge = new SubFlowImpl.MergeBack[Out, Repr] {
+ override def apply[T](flow: Flow[Out, T, NotUsed], breadth: Int):
Repr[T] =
+ via(Split.when(p)).map(_.via(flow)).via(new FlattenMerge(breadth))
+ }
+
+ val finish: Sink[Out, NotUsed] => Closed = s =>
+
via(Split.when(p)).to(Sink.foreach(_.runWith(s)(GraphInterpreter.currentInterpreter.materializer)))
+
+ new SubFlowImpl(Flow[Out], merge, finish)
+ }
/**
* This operation applies the given predicate to all incoming elements and
@@ -2407,14 +2426,24 @@ trait FlowOps[+Out, +Mat] {
*
* See also [[FlowOps.splitWhen]].
*/
+ @deprecated(
+ "Use
.withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather
than a SubstreamCancelStrategy",
+ since = "1.1.0")
def splitAfter(substreamCancelStrategy: SubstreamCancelStrategy)(
p: Out => Boolean): SubFlow[Out, Mat, Repr, Closed] = {
val merge = new SubFlowImpl.MergeBack[Out, Repr] {
override def apply[T](flow: Flow[Out, T, NotUsed], breadth: Int):
Repr[T] =
- via(Split.after(p, substreamCancelStrategy)).map(_.via(flow)).via(new
FlattenMerge(breadth))
+ via(Split
+ .after(p)
+
.withAttributes(ActorAttributes.supervisionStrategy(Split.cancelStrategyToDecider(substreamCancelStrategy))))
+ .map(_.via(flow))
+ .via(new FlattenMerge(breadth))
}
- val finish: (Sink[Out, NotUsed]) => Closed = s =>
- via(Split.after(p, substreamCancelStrategy))
+ val finish: Sink[Out, NotUsed] => Closed = s =>
+ via(
+ Split
+ .after(p)
+
.withAttributes(ActorAttributes.supervisionStrategy(Split.cancelStrategyToDecider(substreamCancelStrategy))))
.to(Sink.foreach(_.runWith(s)(GraphInterpreter.currentInterpreter.materializer)))
new SubFlowImpl(Flow[Out], merge, finish)
}
@@ -2426,8 +2455,15 @@ trait FlowOps[+Out, +Mat] {
*
* @see [[#splitAfter]]
*/
- def splitAfter(p: Out => Boolean): SubFlow[Out, Mat, Repr, Closed] =
- splitAfter(SubstreamCancelStrategy.drain)(p)
+ def splitAfter(p: Out => Boolean): SubFlow[Out, Mat, Repr, Closed] = {
+ val merge = new SubFlowImpl.MergeBack[Out, Repr] {
+ override def apply[T](flow: Flow[Out, T, NotUsed], breadth: Int):
Repr[T] =
+ via(Split.after(p)).map(_.via(flow)).via(new FlattenMerge(breadth))
+ }
+ val finish: Sink[Out, NotUsed] => Closed = s =>
+
via(Split.after(p)).to(Sink.foreach(_.runWith(s)(GraphInterpreter.currentInterpreter.materializer)))
+ new SubFlowImpl(Flow[Out], merge, finish)
+ }
/**
* Transform each input element into a `Source` of output elements that is
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
