chore: Clean up some duplicated code.

--------- Co-authored-by: Matthew de Detrich <[email protected]>
apache · Jan 20, 2024 · 637d72a · 637d72a
1 parent b1ec854
commit 637d72a
Showing 1 changed file with 8 additions and 6 deletions.
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
@@ -193,9 +193,8 @@ import pekko.util.ccompat._
         } catch {
           case NonFatal(ex) =>
             decider(ex) match {
-              case Supervision.Stop    => failStage(ex)
-              case Supervision.Resume  => if (!hasBeenPulled(in)) pull(in)
-              case Supervision.Restart => if (!hasBeenPulled(in)) pull(in)
+              case Supervision.Stop => failStage(ex)
+              case _                => pull(in)
             }
         }
 
@@ -267,9 +266,12 @@ private[stream] object Collect {
         } catch {
           case NonFatal(ex) =>
             decider(ex) match {
-              case Supervision.Stop    => failStage(ex)
-              case Supervision.Resume  => if (!hasBeenPulled(in)) pull(in)
-              case Supervision.Restart => if (!hasBeenPulled(in)) pull(in)
+              case Supervision.Stop => failStage(ex)
+              case _                =>
+                // The !hasBeenPulled(in) check is not required here since it
+                // isn't possible to do an additional pull(in) due to the nature
+                // of how collect works
+                pull(in)
             }
         }