This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch remove-substream-cancel-strategy in repository https://gitbox.apache.org/repos/asf/pekko.git
commit ab1006aec3722e7970c2a64413cecb37d536649e Author: 虎鸣 <[email protected]> AuthorDate: Sun Jul 5 16:41:27 2026 +0800 refactor: remove deprecated SubstreamCancelStrategy and related splitWhen/splitAfter overloads 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]
