Prevent deadlock when publisher is canceled

typelevel · Nov 21, 2024 · d9bd035 · d9bd035
1 parent abe9b03
commit d9bd035
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 1 deletion.
diff --git a/core/shared/src/main/scala/fs2/concurrent/Topic.scala b/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

diff --git a/core/shared/src/test/scala/fs2/concurrent/TopicSuite.scala b/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
+  }
 }