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 7ad07e692c refactor: remove deprecated SubstreamCancelStrategy and
related splitWhen/splitAfter overloads (#3292)
7ad07e692c is described below
commit 7ad07e692cdae86a001b508c6c6c5329e5914ae4
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Mon Jul 6 05:25:40 2026 +0800
refactor: remove deprecated SubstreamCancelStrategy and related
splitWhen/splitAfter overloads (#3292)
Motivation:
SubstreamCancelStrategy and the splitWhen/splitAfter overloads that
accepted it were deprecated since Pekko 1.1.0 in favor of using
.withAttributes(ActorAttributes.supervisionStrategy(decider)).
Modification:
- Delete SubstreamCancelStrategy.scala (class, companion object, and
internal SubstreamCancelStrategies)
- Remove splitWhen(SubstreamCancelStrategy) and
splitAfter(SubstreamCancelStrategy)
from scaladsl Flow/Source, javadsl Flow/Source, and FlowOps
- Remove Split.cancelStrategyToDecider helper from StreamOfStreams.scala
- Clean up unused imports (nowarn, Supervision.Decider)
- Update docs test to use the non-deprecated splitWhen/splitAfter API
- Add MiMa binary compatibility exclusion filters for 2.0.x
Result:
Deprecated API removed. Users should use
.withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider))
instead.
Tests:
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.FlowSplitWhenSpec
org.apache.pekko.stream.scaladsl.FlowSplitAfterSpec" - passed, 24 tests
- sbt "stream / mimaReportBinaryIssues" - passed
References: None - deprecated API cleanup
---
.../test/scala/docs/stream/SubstreamDocSpec.scala | 5 +-
.../remove-substream-cancel-strategy.excludes | 32 +++++
.../pekko/stream/SubstreamCancelStrategy.scala | 57 --------
.../pekko/stream/impl/fusing/StreamOfStreams.scala | 10 +-
.../org/apache/pekko/stream/javadsl/Flow.scala | 32 +----
.../org/apache/pekko/stream/javadsl/Source.scala | 26 ----
.../org/apache/pekko/stream/scaladsl/Flow.scala | 145 ---------------------
7 files changed, 37 insertions(+), 270 deletions(-)
diff --git a/docs/src/test/scala/docs/stream/SubstreamDocSpec.scala
b/docs/src/test/scala/docs/stream/SubstreamDocSpec.scala
index f3cf41e9d6..42f2b9ff88 100644
--- a/docs/src/test/scala/docs/stream/SubstreamDocSpec.scala
+++ b/docs/src/test/scala/docs/stream/SubstreamDocSpec.scala
@@ -14,7 +14,6 @@
package docs.stream
import org.apache.pekko.stream.scaladsl.{ Sink, Source }
-import org.apache.pekko.stream.SubstreamCancelStrategy
import org.apache.pekko.testkit.PekkoSpec
class SubstreamDocSpec extends PekkoSpec {
@@ -42,9 +41,9 @@ class SubstreamDocSpec extends PekkoSpec {
"generate substreams by splitWhen and splitAfter" in {
// #splitWhenAfter
- Source(1 to 10).splitWhen(SubstreamCancelStrategy.drain)(_ == 3)
+ Source(1 to 10).splitWhen(_ == 3)
- Source(1 to 10).splitAfter(SubstreamCancelStrategy.drain)(_ == 3)
+ Source(1 to 10).splitAfter(_ == 3)
// #splitWhenAfter
// #wordCount
diff --git
a/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-substream-cancel-strategy.excludes
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-substream-cancel-strategy.excludes
new file mode 100644
index 0000000000..68ac07e681
--- /dev/null
+++
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-substream-cancel-strategy.excludes
@@ -0,0 +1,32 @@
+# 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.
+
+# Remove deprecated SubstreamCancelStrategy and related splitWhen/splitAfter
overloads (deprecated since Pekko 1.1.0)
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.SubstreamCancelStrategy*")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.SubstreamCancelStrategies*")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.javadsl.Flow.splitWhen")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.javadsl.Flow.splitAfter")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.javadsl.Source.splitWhen")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.javadsl.Source.splitAfter")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.Flow.splitWhen")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.Flow.splitAfter")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.FlowOps.splitWhen")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.FlowOps.splitAfter")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.GraphDSL#Implicits#PortOpsImpl.splitWhen")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.GraphDSL#Implicits#PortOpsImpl.splitAfter")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.Source.splitWhen")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.scaladsl.Source.splitAfter")
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/SubstreamCancelStrategy.scala
b/stream/src/main/scala/org/apache/pekko/stream/SubstreamCancelStrategy.scala
deleted file mode 100644
index 7f74ac764e..0000000000
---
a/stream/src/main/scala/org/apache/pekko/stream/SubstreamCancelStrategy.scala
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * license agreements; and to You under the Apache License, version 2.0:
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * This file is part of the Apache Pekko project, which was derived from Akka.
- */
-
-/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
- */
-
-package org.apache.pekko.stream
-
-import scala.annotation.nowarn
-
-import SubstreamCancelStrategies._
-
-/**
- * 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 {
-
- /**
- * 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 {
-
- /**
- * Cancel the stream of streams if any substream is cancelled.
- */
- def propagate: SubstreamCancelStrategy = Propagate
-
- /**
- * Drain substream on cancellation in order to prevent stalling of the
stream of streams.
- */
- def drain: SubstreamCancelStrategy = Drain
-}
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 fdb0e1ed15..e750eca615 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
@@ -16,7 +16,7 @@ package org.apache.pekko.stream.impl.fusing
import java.util.Collections
import java.util.concurrent.atomic.AtomicReference
-import scala.annotation.{ nowarn, tailrec }
+import scala.annotation.tailrec
import scala.collection.immutable
import scala.concurrent.Future
import scala.concurrent.duration.FiniteDuration
@@ -31,7 +31,6 @@ import pekko.stream._
import pekko.stream.ActorAttributes.StreamSubscriptionTimeout
import pekko.stream.ActorAttributes.SupervisionStrategy
import pekko.stream.Attributes.SourceLocation
-import pekko.stream.Supervision.Decider
import pekko.stream.impl.{ Buffer => BufferImpl, FailedSource,
JavaStreamSource }
import pekko.stream.impl.ActorSubscriberMessage
import pekko.stream.impl.ActorSubscriberMessage.OnError
@@ -586,13 +585,6 @@ import pekko.util.OptionVal
/** Splits after the current element. The current element will be the last
element in the current substream. */
case object SplitAfter extends SplitDecision
- @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)
}
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 0b80d72ebb..2f5d6bc70b 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
@@ -2909,27 +2909,13 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
*
* '''Completes when''' upstream completes
*
- * '''Cancels when''' downstream cancels and substreams cancel on
`SubstreamCancelStrategy.drain()`, downstream
- * cancels or any substream cancels on `SubstreamCancelStrategy.propagate()`
+ * '''Cancels when''' downstream cancels and substreams cancel
*
* See also [[Flow.splitAfter]].
*/
def splitWhen(p: function.Predicate[Out]): SubFlow[In, Out, Mat] =
new SubFlow(delegate.splitWhen(p.test))
- /**
- * This operation applies the given predicate to all incoming elements and
- * emits them to a stream of output streams, always beginning a new one with
- * the current element if the given predicate returns true for it.
- *
- * @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))
-
/**
* This operation applies the given predicate to all incoming elements and
* emits them to a stream of output streams. It *ends* the current substream
when the
@@ -2970,27 +2956,13 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
*
* '''Completes when''' upstream completes
*
- * '''Cancels when''' downstream cancels and substreams cancel on
`SubstreamCancelStrategy.drain`, downstream
- * cancels or any substream cancels on `SubstreamCancelStrategy.propagate`
+ * '''Cancels when''' downstream cancels and substreams cancel
*
* See also [[Flow.splitWhen]].
*/
def splitAfter(p: function.Predicate[Out]): SubFlow[In, Out, Mat] =
new SubFlow(delegate.splitAfter(p.test))
- /**
- * This operation applies the given predicate to all incoming elements and
- * emits them to a stream of output streams. It *ends* the current substream
when the
- * predicate is true.
- *
- * @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))
-
/**
* Transform each input element into a `Source` of output elements that is
* then flattened into the output stream by concatenation,
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 ae3a668983..1b10205833 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
@@ -4397,19 +4397,6 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
def splitWhen(p: function.Predicate[Out]): SubSource[Out, Mat] =
new SubSource(delegate.splitWhen(p.test))
- /**
- * This operation applies the given predicate to all incoming elements and
- * emits them to a stream of output streams, always beginning a new one with
- * the current element if the given predicate returns true for it.
- *
- * @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))
-
/**
* This operation applies the given predicate to all incoming elements and
* emits them to a stream of output streams. It *ends* the current substream
when the
@@ -4457,19 +4444,6 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
def splitAfter(p: function.Predicate[Out]): SubSource[Out, Mat] =
new SubSource(delegate.splitAfter(p.test))
- /**
- * This operation applies the given predicate to all incoming elements and
- * emits them to a stream of output streams. It *ends* the current substream
when the
- * predicate is true.
- *
- * @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))
-
/**
* Transform each input element into a `Source` of output elements that is
* then flattened into the output stream by concatenation,
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 70a8f263ec..109765e67b 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
@@ -2751,84 +2751,6 @@ trait FlowOps[+Out, +Mat] {
*/
def groupBy[K](maxSubstreams: Int, f: Out => K): SubFlow[Out, Mat, Repr,
Closed] = groupBy(maxSubstreams, f, false)
- /**
- * This operation applies the given predicate to all incoming elements and
- * emits them to a stream of output streams, always beginning a new one with
- * the current element if the given predicate returns true for it. This means
- * that for the following series of predicate values, three substreams will
- * be produced with lengths 1, 2, and 3:
- *
- * {{{
- * false, // element goes into first substream
- * true, false, // elements go into second substream
- * true, false, false // elements go into third substream
- * }}}
- *
- * In case the *first* element of the stream matches the predicate, the first
- * substream emitted by splitWhen will start from that element. For example:
- *
- * {{{
- * true, false, false // first substream starts from the split-by element
- * true, false // subsequent substreams operate the same way
- * }}}
- *
- * The object returned from this method is not a normal [[Source]] or
[[Flow]],
- * it is a [[SubFlow]]. This means that after this operator all
transformations
- * are applied to all encountered substreams in the same fashion. Substream
mode
- * is exited either by closing the substream (i.e. connecting it to a
[[Sink]])
- * or by merging the substreams back together; see the `to` and `mergeBack`
methods
- * on [[SubFlow]] for more information.
- *
- * It is important to note that the substreams also propagate back-pressure
as
- * any other stream, which means that blocking one substream will block the
`splitWhen`
- * operator itself—and thereby all substreams—once all internal or
- * explicit buffers are filled.
- *
- * If the split predicate `p` throws an exception and the supervision
decision
- * is [[pekko.stream.Supervision.Stop]] the stream and substreams will be
completed
- * with failure.
- *
- * If the split predicate `p` throws an exception and the supervision
decision
- * is [[pekko.stream.Supervision.Resume]] or
[[pekko.stream.Supervision.Restart]]
- * the element is dropped and the stream and substreams continue.
- *
- * '''Emits when''' an element for which the provided predicate is true,
opening and emitting
- * a new substream for subsequent element
- *
- * '''Backpressures when''' there is an element pending for the next
substream, but the previous
- * is not fully consumed yet, or the substream backpressures
- *
- * '''Completes when''' upstream completes
- *
- * '''Cancels when''' downstream cancels and substreams cancel on
`SubstreamCancelStrategy.drain`, downstream
- * cancels or any substream cancels on `SubstreamCancelStrategy.propagate`
- *
- * 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)
-
.withAttributes(ActorAttributes.supervisionStrategy(Split.cancelStrategyToDecider(substreamCancelStrategy))))
- .map(_.via(flow))
- .via(new FlattenMerge(breadth))
- }
-
- 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)
- }
-
/**
* This operation applies the given predicate to all incoming elements and
* emits them to a stream of output streams, always beginning a new one with
@@ -2848,73 +2770,6 @@ trait FlowOps[+Out, +Mat] {
new SubFlowImpl(Flow[Out], merge, finish)
}
- /**
- * This operation applies the given predicate to all incoming elements and
- * emits them to a stream of output streams. It *ends* the current substream
when the
- * predicate is true. This means that for the following series of predicate
values,
- * three substreams will be produced with lengths 2, 2, and 3:
- *
- * {{{
- * false, true, // elements go into first substream
- * false, true, // elements go into second substream
- * false, false, true // elements go into third substream
- * }}}
- *
- * The object returned from this method is not a normal [[Source]] or
[[Flow]],
- * it is a [[SubFlow]]. This means that after this operator all
transformations
- * are applied to all encountered substreams in the same fashion. Substream
mode
- * is exited either by closing the substream (i.e. connecting it to a
[[Sink]])
- * or by merging the substreams back together; see the `to` and `mergeBack`
methods
- * on [[SubFlow]] for more information.
- *
- * It is important to note that the substreams also propagate back-pressure
as
- * any other stream, which means that blocking one substream will block the
`splitAfter`
- * operator itself—and thereby all substreams—once all internal or
- * explicit buffers are filled.
- *
- * If the split predicate `p` throws an exception and the supervision
decision
- * is [[pekko.stream.Supervision.Stop]] the stream and substreams will be
completed
- * with failure.
- *
- * If the split predicate `p` throws an exception and the supervision
decision
- * is [[pekko.stream.Supervision.Resume]] or
[[pekko.stream.Supervision.Restart]]
- * the element is dropped and the stream and substreams continue.
- *
- * '''Emits when''' an element passes through. When the provided predicate
is true it emits the element
- * and opens a new substream for subsequent element
- *
- * '''Backpressures when''' there is an element pending for the next
substream, but the previous
- * is not fully consumed yet, or the substream backpressures
- *
- * '''Completes when''' upstream completes
- *
- * '''Cancels when''' downstream cancels and substreams cancel on
`SubstreamCancelStrategy.drain`, downstream
- * cancels or any substream cancels on `SubstreamCancelStrategy.propagate`
- *
- * 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)
-
.withAttributes(ActorAttributes.supervisionStrategy(Split.cancelStrategyToDecider(substreamCancelStrategy))))
- .map(_.via(flow))
- .via(new FlattenMerge(breadth))
- }
- 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)
- }
-
/**
* This operation applies the given predicate to all incoming elements and
* emits them to a stream of output streams. It *ends* the current substream
when the
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
