Merge pull request #3449 from BalmungSan/flow-processor

Add `interop.flow.pipeToProcessor` & `interop.flow.processorToPipe`

Author: mpilquist

core/shared/src/main/scala/fs2/Stream.scala

@@ -40,7 +40,7 @@ import fs2.internal._
 import org.typelevel.scalaccompat.annotation._
 import Pull.StreamPullOps
 
-import java.util.concurrent.Flow.{Publisher, Subscriber}
+import java.util.concurrent.Flow.{Publisher, Processor, Subscriber}
 
 /** A stream producing output of type `O` and which may evaluate `F` effects.
   *
@@ -2853,7 +2853,7 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
     * @param subscriber the [[Subscriber]] that will receive the elements of the stream.
     */
   def subscribe[F2[x] >: F[x]: Async, O2 >: O](subscriber: Subscriber[O2]): Stream[F2, Nothing] =
-    interop.flow.subscribeAsStream[F2, O2](this, subscriber)
+    interop.flow.StreamSubscription.subscribe[F2, O2](this, subscriber)
 
   /** Emits all elements of the input except the first one.
     *
@@ -3001,7 +3001,7 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
 
   /** @see [[toPublisher]] */
   def toPublisherResource[F2[x] >: F[x]: Async, O2 >: O]: Resource[F2, Publisher[O2]] =
-    interop.flow.toPublisher(this)
+    interop.flow.StreamPublisher(this)
 
   /** Translates effect type from `F` to `G` using the supplied `FunctionK`.
     */
@@ -3883,6 +3883,56 @@ object Stream extends StreamLowPriority {
     await
   }
 
+  /** Creates a [[Stream]] from a `subscribe` function;
+    * analogous to a `Publisher`, but effectual.
+    *
+    * This function is useful when you actually need to provide a subscriber to a third-party.
+    *
+    * @example {{{
+    * scala> import cats.effect.IO
+    * scala> import java.util.concurrent.Flow.{Publisher, Subscriber}
+    * scala>
+    * scala> def thirdPartyLibrary(subscriber: Subscriber[Int]): Unit = {
+    *      |  def somePublisher: Publisher[Int] = ???
+    *      |  somePublisher.subscribe(subscriber)
+    *      | }
+    * scala>
+    * scala> // Interop with the third party library.
+    * scala> Stream.fromPublisher[IO, Int](chunkSize = 16) { subscriber =>
+    *      |   IO.println("Subscribing!") >>
+    *      |   IO.delay(thirdPartyLibrary(subscriber)) >>
+    *      |   IO.println("Subscribed!")
+    *      | }
+    * res0: Stream[IO, Int] = Stream(..)
+    * }}}
+    *
+    * @note The subscribe function will not be executed until the stream is run.
+    *
+    * @see the overload that only requires a [[Publisher]].
+    *
+    * @param chunkSize setup the number of elements asked each time from the [[Publisher]].
+    *                  A high number may be useful if the publisher is triggering from IO,
+    *                  like requesting elements from a database.
+    *                  A high number will also lead to more elements in memory.
+    *                  The stream will not emit new element until,
+    *                  either the `Chunk` is filled or the publisher finishes.
+    * @param subscribe The effectual function that will be used to initiate the consumption process,
+    *                  it receives a [[Subscriber]] that should be used to subscribe to a [[Publisher]].
+    *                  The `subscribe` operation must be called exactly once.
+    */
+  def fromPublisher[F[_], A](
+      chunkSize: Int
+  )(
+      subscribe: Subscriber[A] => F[Unit]
+  )(implicit
+      F: Async[F]
+  ): Stream[F, A] =
+    Stream
+      .eval(interop.flow.StreamSubscriber[F, A](chunkSize))
+      .flatMap { subscriber =>
+        subscriber.stream(subscribe(subscriber))
+      }
+
   /** Creates a [[Stream]] from a [[Publisher]].
     *
     * @example {{{
@@ -3900,8 +3950,6 @@ object Stream extends StreamLowPriority {
     *
     * @note The [[Publisher]] will not receive a [[Subscriber]] until the stream is run.
     *
-    * @see the `toStream` extension method added to `Publisher`
-    *
     * @param publisher The [[Publisher]] to consume.
     * @param chunkSize setup the number of elements asked each time from the [[Publisher]].
     *                  A high number may be useful if the publisher is triggering from IO,
@@ -3911,7 +3959,7 @@ object Stream extends StreamLowPriority {
     *                  either the `Chunk` is filled or the publisher finishes.
     */
   def fromPublisher[F[_]]: interop.flow.syntax.FromPublisherPartiallyApplied[F] =
-    interop.flow.fromPublisher
+    new interop.flow.syntax.FromPublisherPartiallyApplied(dummy = true)
 
   /** Like `emits`, but works for any G that has a `Foldable` instance.
     */
@@ -4697,7 +4745,7 @@ object Stream extends StreamLowPriority {
     def unsafeToPublisher()(implicit
         runtime: IORuntime
     ): Publisher[A] =
-      interop.flow.unsafeToPublisher(self)
+      interop.flow.StreamPublisher.unsafe(self)
   }
 
   /** Projection of a `Stream` providing various ways to get a `Pull` from the `Stream`. */
@@ -5541,6 +5589,47 @@ object Stream extends StreamLowPriority {
     /** Transforms the right input of the given `Pipe2` using a `Pipe`. */
     def attachR[I0, O2](p: Pipe2[F, I0, O, O2]): Pipe2[F, I0, I, O2] =
       (l, r) => p(l, self(r))
+
+    /** Creates a flow [[Processor]] from this [[Pipe]].
+      *
+      * You are required to manually subscribe this [[Processor]] to an upstream [[Publisher]], and have at least one downstream [[Subscriber]] subscribe to the [[Consumer]].
+      *
+      * 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.
+      *
+      * @param chunkSize setup the number of elements asked each time from the upstream [[Publisher]].
+      *                  A high number may be useful if the publisher is triggering from IO,
+      *                  like requesting elements from a database.
+      *                  A high number will also lead to more elements in memory.
+      */
+    def toProcessor(
+        chunkSize: Int
+    )(implicit
+        F: Async[F]
+    ): Resource[F, Processor[I, O]] =
+      interop.flow.StreamProcessor.fromPipe(pipe = self, chunkSize)
+  }
+
+  /** Provides operations on IO pipes for syntactic convenience. */
+  implicit final class IOPipeOps[I, O](private val self: Pipe[IO, I, O]) extends AnyVal {
+
+    /** Creates a [[Processor]] from this [[Pipe]].
+      *
+      * You are required to manually subscribe this [[Processor]] to an upstream [[Publisher]], and have at least one downstream [[Subscriber]] subscribe to the [[Consumer]].
+      *
+      * @param chunkSize setup the number of elements asked each time from the upstream [[Publisher]].
+      *                  A high number may be useful if the publisher is triggering from IO,
+      *                  like requesting elements from a database.
+      *                  A high number will also lead to more elements in memory.
+      */
+    def unsafeToProcessor(
+        chunkSize: Int
+    )(implicit
+        runtime: IORuntime
+    ): Processor[I, O] =
+      interop.flow.StreamProcessor.unsafeFromPipe(pipe = self, chunkSize)
   }
 
   /** Provides operations on pure pipes for syntactic convenience. */

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

@@ -19,6 +19,9 @@
  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  */
 
+import java.util.concurrent.Flow.Processor
+import cats.effect.Async
+
 package object fs2 {
 
   /** A stream transformation represented as a function from stream to stream.
@@ -27,6 +30,36 @@ package object fs2 {
     */
   type Pipe[F[_], -I, +O] = Stream[F, I] => Stream[F, O]
 
+  object Pipe {
+    final class FromProcessorPartiallyApplied[F[_]](private val dummy: Boolean) extends AnyVal {
+      def apply[I, O](
+          processor: Processor[I, O],
+          chunkSize: Int
+      )(implicit
+          F: Async[F]
+      ): Pipe[F, I, O] =
+        new interop.flow.ProcessorPipe(processor, chunkSize)
+    }
+
+    /** Creates a [[Pipe]] from the given [[Processor]].
+      *
+      * The input stream won't be consumed until you request elements from the output stream,
+      * and thus the processor is not initiated until then.
+      *
+      * @note The [[Pipe]] can be reused multiple times as long as the [[Processor]] can be reused.
+      * Each invocation of the pipe will create and manage its own internal [[Publisher]] and [[Subscriber]],
+      * and use them to subscribe to and from the [[Processor]] respectively.
+      *
+      * @param [[processor]] the [[Processor]] that represents the [[Pipe]] logic.
+      * @param chunkSize setup the number of elements asked each time from the upstream [[Publisher]].
+      *                  A high number may be useful if the publisher is triggering from IO,
+      *                  like requesting elements from a database.
+      *                  A high number will also lead to more elements in memory.
+      */
+    def fromProcessor[F[_]]: FromProcessorPartiallyApplied[F] =
+      new FromProcessorPartiallyApplied[F](dummy = true)
+  }
+
   /** A stream transformation that combines two streams in to a single stream,
     * represented as a function from two streams to a single stream.
     *

core/shared/src/main/scala/fs2/interop/flow/ProcessorPipe.scala

@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2013 Functional Streams for Scala
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ * the Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package fs2
+package interop
+package flow
+
+import cats.syntax.all.*
+
+import java.util.concurrent.Flow
+import cats.effect.Async
+
+private[fs2] final class ProcessorPipe[F[_], I, O](
+    processor: Flow.Processor[I, O],
+    chunkSize: Int
+)(implicit
+    F: Async[F]
+) extends Pipe[F, I, O] {
+  override def apply(stream: Stream[F, I]): Stream[F, O] =
+    (
+      Stream.resource(StreamPublisher[F, I](stream)),
+      Stream.eval(StreamSubscriber[F, O](chunkSize))
+    ).flatMapN { (publisher, subscriber) =>
+      val initiateUpstreamProduction = F.delay(publisher.subscribe(processor))
+      val initiateDownstreamConsumption = F.delay(processor.subscribe(subscriber))
+
+      subscriber.stream(
+        subscribe = initiateUpstreamProduction >> initiateDownstreamConsumption
+      )
+    }
+}

core/shared/src/main/scala/fs2/interop/flow/StreamProcessor.scala

@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2013 Functional Streams for Scala
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to
+ * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ * the Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package fs2
+package interop
+package flow
+
+import java.util.concurrent.Flow
+import cats.effect.{Async, IO, Resource}
+import cats.effect.unsafe.IORuntime
+
+private[fs2] final class StreamProcessor[F[_], I, O](
+    streamSubscriber: StreamSubscriber[F, I],
+    streamPublisher: StreamPublisher[F, O]
+) extends Flow.Processor[I, O] {
+  override def onSubscribe(subscription: Flow.Subscription): Unit =
+    streamSubscriber.onSubscribe(subscription)
+
+  override def onNext(i: I): Unit =
+    streamSubscriber.onNext(i)
+
+  override def onError(ex: Throwable): Unit =
+    streamSubscriber.onError(ex)
+
+  override def onComplete(): Unit =
+    streamSubscriber.onComplete()
+
+  override def subscribe(subscriber: Flow.Subscriber[? >: O]): Unit =
+    streamPublisher.subscribe(subscriber)
+}
+
+private[fs2] object StreamProcessor {
+  def fromPipe[F[_], I, O](
+      pipe: Pipe[F, I, O],
+      chunkSize: Int
+  )(implicit
+      F: Async[F]
+  ): Resource[F, StreamProcessor[F, I, O]] =
+    for {
+      streamSubscriber <- Resource.eval(StreamSubscriber[F, I](chunkSize))
+      inputStream = streamSubscriber.stream(subscribe = F.unit)
+      outputStream = pipe(inputStream)
+      streamPublisher <- StreamPublisher(outputStream)
+    } yield new StreamProcessor(
+      streamSubscriber,
+      streamPublisher
+    )
+
+  def unsafeFromPipe[I, O](
+      pipe: Pipe[IO, I, O],
+      chunkSize: Int
+  )(implicit
+      runtime: IORuntime
+  ): StreamProcessor[IO, I, O] = {
+    val streamSubscriber = StreamSubscriber.unsafe[IO, I](chunkSize)
+    val inputStream = streamSubscriber.stream(subscribe = IO.unit)
+    val outputStream = pipe(inputStream)
+    val streamPublisher = StreamPublisher.unsafe(outputStream)
+    new StreamProcessor(
+      streamSubscriber,
+      streamPublisher
+    )
+  }
+}

core/shared/src/main/scala/fs2/interop/flow/StreamPublisher.scala

@@ -23,8 +23,7 @@ package fs2
 package interop
 package flow
 
-import cats.effect.IO
-import cats.effect.kernel.{Async, Resource}
+import cats.effect.{Async, IO, Resource}
 import cats.effect.std.Dispatcher
 import cats.effect.unsafe.IORuntime
 
@@ -42,7 +41,7 @@ import scala.util.control.NoStackTrace
   *
   * @see [[https://github.com/reactive-streams/reactive-streams-jvm#1-publisher-code]]
   */
-private[flow] sealed abstract class StreamPublisher[F[_], A] private (
+private[fs2] sealed abstract class StreamPublisher[F[_], A] private (
     stream: Stream[F, A]
 )(implicit
     F: Async[F]
@@ -65,7 +64,7 @@ private[flow] sealed abstract class StreamPublisher[F[_], A] private (
   }
 }
 
-private[flow] object StreamPublisher {
+private[fs2] object StreamPublisher {
   private final class DispatcherStreamPublisher[F[_], A](
       stream: Stream[F, A],
       dispatcher: Dispatcher[F]