From dac6c7398d4bef475dd7e78c4c6af4fd1f9bd8ad Mon Sep 17 00:00:00 2001 From: He-Pin Date: Sat, 21 Dec 2024 18:12:58 +0800 Subject: [PATCH] chore: Remove SupervisedGraphStageLogic --- ...vised-group-stage-logic.backwards.excludes | 1 + .../apache/pekko/stream/impl/fusing/Ops.scala | 29 +------------------ 2 files changed, 2 insertions(+), 28 deletions(-) create mode 100644 stream/src/main/mima-filters/1.0.x.backwards.excludes/pr-1619-remove-supervised-group-stage-logic.backwards.excludes diff --git a/stream/src/main/mima-filters/1.0.x.backwards.excludes/pr-1619-remove-supervised-group-stage-logic.backwards.excludes b/stream/src/main/mima-filters/1.0.x.backwards.excludes/pr-1619-remove-supervised-group-stage-logic.backwards.excludes new file mode 100644 index 00000000000..1de591945b9 --- /dev/null +++ b/stream/src/main/mima-filters/1.0.x.backwards.excludes/pr-1619-remove-supervised-group-stage-logic.backwards.excludes @@ -0,0 +1 @@ + ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.impl.fusing.SupervisedGraphStageLogic") \ No newline at end of file diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala index d892a049526..1b23fc96c98 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala @@ -27,7 +27,7 @@ import scala.util.control.Exception.Catcher import org.apache.pekko import pekko.actor.{ ActorRef, Terminated } -import pekko.annotation.{ DoNotInherit, InternalApi } +import pekko.annotation.InternalApi import pekko.event._ import pekko.event.Logging.LogLevel import pekko.stream.{ Supervision, _ } @@ -206,33 +206,6 @@ import pekko.util.ccompat._ override def toString = "DropWhile" } -/** - * INTERNAL API - */ -@DoNotInherit private[pekko] abstract class SupervisedGraphStageLogic(inheritedAttributes: Attributes, shape: Shape) - extends GraphStageLogic(shape) { - private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider - - def withSupervision[T](f: () => T): Option[T] = - try { - Some(f()) - } catch { - case NonFatal(ex) => - decider(ex) match { - case Supervision.Stop => onStop(ex) - case Supervision.Resume => onResume(ex) - case Supervision.Restart => onRestart(ex) - } - None - } - - def onResume(t: Throwable): Unit - - def onStop(t: Throwable): Unit = failStage(t) - - def onRestart(t: Throwable): Unit = onResume(t) -} - private[stream] object Collect { // Cached function that can be used with PartialFunction.applyOrElse to ensure that A) the guard is only applied once, // and the caller can check the returned value with Collect.notApplied to query whether the PF was applied or not.