Prevent deadlock when publisher is canceled

Author: dimitriho

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

core/shared/src/main/scala/fs2/concurrent/Topic.scala

@@ -208,7 +208,7 @@ object Topic {
         }
 
         def publish: Pipe[F, A, Nothing] = { in =>
-          (in ++ Stream.exec(close.void))
+          in.onFinalize(close.void)
             .evalMap(publish1)
             .takeWhile(_.isRight)
             .drain

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
+  }
 }

core/shared/src/test/scala/fs2/concurrent/TopicSuite.scala

@@ -185,4 +185,25 @@ class TopicSuite extends Fs2Suite {
 
     TestControl.executeEmbed(program) // will fail if program is deadlocked
   }
+
+  test("publisher cancellation does not deadlock") {
+    val program =
+      Topic[IO, String]
+        .flatMap { topic =>
+          val publisher =
+            Stream
+              .constant("1")
+              .covary[IO]
+              .evalTap(_ => IO.canceled)
+              .through(topic.publish)
+
+          Stream
+            .resource(topic.subscribeAwait(1))
+            .flatMap(subscriber => subscriber.concurrently(publisher))
+            .compile
+            .drain
+        }
+
+    TestControl.executeEmbed(program) // will fail if program is deadlocked
+  }
 }