Fix uncaught decider exception in Split with Supervision.resumingDecider

apache · Mar 18, 2024 · df3b9fc · df3b9fc
1 parent 5376692
commit df3b9fc
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 7 deletions.
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
@@ -211,8 +211,6 @@ class FlowSplitAfterSpec extends StreamSpec("""
     }
 
     "resume stream when splitAfter function throws" in {
-      info("Supervision is not supported fully by GraphStages yet")
-      pending
       val publisherProbeProbe = TestPublisher.manualProbe[Int]()
       val exc = TE("test")
       val publisher = Source
@@ -250,10 +248,10 @@ class FlowSplitAfterSpec extends StreamSpec("""
       upstreamSubscription.sendNext(6)
       substreamPuppet1.expectNext(6)
       substreamPuppet1.expectComplete()
+      upstreamSubscription.sendNext(7)
       val substream2 = subscriber.expectNext()
       val substreamPuppet2 = StreamPuppet(substream2.runWith(Sink.asPublisher(false)))
       substreamPuppet2.request(10)
-      upstreamSubscription.sendNext(7)
       substreamPuppet2.expectNext(7)
 
       upstreamSubscription.sendComplete()

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
@@ -327,9 +327,6 @@ class FlowSplitWhenSpec extends StreamSpec("""
     }
 
     "resume stream when splitWhen function throws" in {
-      info("Supervision is not supported fully by GraphStages yet")
-      pending
-
       val publisherProbeProbe = TestPublisher.manualProbe[Int]()
       val exc = TE("test")
       val publisher = Source

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
@@ -640,7 +640,12 @@ import pekko.util.ccompat.JavaConverters._
             else substreamSource.push(elem)
           }
         } catch {
-          case NonFatal(ex) => onUpstreamFailure(ex)
+          case NonFatal(ex) =>
+            decider(ex) match {
+              case Supervision.Resume  => pull(in)
+              case Supervision.Stop    => onUpstreamFailure(ex)
+              case Supervision.Restart => onUpstreamFailure(ex) // TODO implement restart?
+            }
         }
       }