fixup! Prevent deadlock when publisher is canceled

typelevel · Nov 30, 2024 · bcb9c48 · bcb9c48
1 parent d9bd035
commit bcb9c48
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 1 deletion.
diff --git a/core/shared/src/main/scala/fs2/concurrent/Channel.scala b/core/shared/src/main/scala/fs2/concurrent/Channel.scala
@@ -151,7 +151,7 @@ object Channel {
       new Channel[F, A] {
 
         def sendAll: Pipe[F, A, Nothing] = { in =>
-          (in ++ Stream.exec(close.void))
+          in.onFinalize(close.void)
             .evalMap(send)
             .takeWhile(_.isRight)
             .drain

diff --git a/core/shared/src/test/scala/fs2/concurrent/ChannelSuite.scala b/core/shared/src/test/scala/fs2/concurrent/ChannelSuite.scala
@@ -323,4 +323,19 @@ class ChannelSuite extends Fs2Suite {
     racingSendOperations(channel)
   }
 
+  test("stream should terminate when sendAll is interrupted") {
+    val program =
+      Channel
+        .bounded[IO, Unit](1)
+        .flatMap { ch =>
+          val producer =
+            Stream
+              .eval(IO.canceled)
+              .through(ch.sendAll)
+
+          ch.stream.concurrently(producer).compile.drain
+        }
+
+    TestControl.executeEmbed(program) // will fail if program is deadlocked
+  }
 }