Wait for Dispatcher in StreamPublisher

typelevel · Nov 24, 2023 · b5e91af · b5e91af
1 parent 5c766b9
commit b5e91af
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 5 deletions.
diff --git a/core/shared/src/main/scala/fs2/interop/flow/StreamPublisher.scala b/core/shared/src/main/scala/fs2/interop/flow/StreamPublisher.scala
@@ -72,12 +72,12 @@ private[flow] sealed abstract class StreamPublisher[F[_], A] private (
 private[flow] object StreamPublisher {
   private final class DispatcherStreamPublisher[F[_], A](
       stream: Stream[F, A],
-      dispatcher: Dispatcher[F]
+      startDispatcher: Dispatcher[F]
   )(implicit
       F: Async[F]
   ) extends StreamPublisher[F, A](stream) {
     override protected final def runSubscription(subscribe: F[Unit]): Unit =
-      dispatcher.unsafeRunAndForget(subscribe)
+      startDispatcher.unsafeRunAndForget(subscribe)
   }
 
   private final class IORuntimeStreamPublisher[A](
@@ -94,7 +94,7 @@ private[flow] object StreamPublisher {
   )(implicit
       F: Async[F]
   ): Resource[F, StreamPublisher[F, A]] =
-    Dispatcher.parallel[F](await = false).map { startDispatcher =>
+    Dispatcher.parallel[F](await = true).map { startDispatcher =>
       new DispatcherStreamPublisher(stream, startDispatcher)
     }
 

diff --git a/core/shared/src/main/scala/fs2/interop/flow/package.scala b/core/shared/src/main/scala/fs2/interop/flow/package.scala
@@ -136,7 +136,9 @@ package object flow {
   /** Creates a [[Publisher]] from a [[Stream]].
     *
     * The stream is only ran when elements are requested.
-    * Closing the [[Resource]] means gracefully shutting down all active subscriptions.
+    * Closing the [[Resource]] means not accepting new subscriptions,
+    * but waiting for all active ones to finish consuming.
+    * Canceling the [[Resource.use]] means gracefully shutting down all active subscriptions.
     * Thus, no more elements will be published.
     *
     * @note This [[Publisher]] can be reused for multiple [[Subscribers]],

diff --git a/core/shared/src/main/scala/fs2/interop/flow/syntax.scala b/core/shared/src/main/scala/fs2/interop/flow/syntax.scala
@@ -67,7 +67,9 @@ object syntax {
     /** Creates a [[Publisher]] from a [[Stream]].
       *
       * The stream is only ran when elements are requested.
-      * Closing the [[Resource]] means gracefully shutting down all active subscriptions.
+      * Closing the [[Resource]] means not accepting new subscriptions,
+      * but waiting for all active ones to finish consuming.
+      * Canceling the [[Resource.use]] means gracefully shutting down all active subscriptions.
       * Thus, no more elements will be published.
       *
       * @note This [[Publisher]] can be reused for multiple [[Subscribers]],