Terminate the sending fork in mapParScope after an error

softwaremill · Oct 30, 2023 · 8f08365 · 8f08365
1 parent f3214bc
commit 8f08365
Showing 1 changed file with 1 addition and 2 deletions.
diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala
@@ -158,8 +158,7 @@ trait SourceOps[+T] { this: Source[T] =>
         repeatWhile {
           inProgress.receive() match
             case f: Fork[Option[U]] @unchecked =>
-              f.join().foreach(c2.send)
-              true
+              f.join().map(c2.send).isDefined
             case ChannelClosed.Done =>
               closeScope.countDown()
               c2.done()