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

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.
   *
@@ -5541,6 +5541,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.pipeToProcessor(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.unsafePipeToProcessor(pipe = self, chunkSize)
   }
 
   /** Provides operations on pure pipes for syntactic convenience. */

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[flow] 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[flow] 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[flow] 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
 

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

@@ -23,7 +23,7 @@ package fs2
 package interop
 package flow
 
-import cats.effect.kernel.Async
+import cats.effect.Async
 
 import java.util.Objects.requireNonNull
 import java.util.concurrent.Flow.{Subscriber, Subscription}
@@ -330,18 +330,24 @@ private[flow] object StreamSubscriber {
   /** Instantiates a new [[StreamSubscriber]] for the given buffer size. */
   def apply[F[_], A](
       chunkSize: Int
-  )(implicit F: Async[F]): F[StreamSubscriber[F, A]] = {
-    require(chunkSize > 0, "The buffer size MUST be positive")
+  )(implicit
+      F: Async[F]
+  ): F[StreamSubscriber[F, A]] =
+    F.delay(unsafe(chunkSize))
 
-    F.delay {
-      val currentState =
-        new AtomicReference[(State, () => Unit)]((State.Uninitialized(cb = None), noop))
+  private[fs2] def unsafe[F[_], A](
+      chunkSize: Int
+  )(implicit
+      F: Async[F]
+  ): StreamSubscriber[F, A] = {
+    require(chunkSize > 0, "The buffer size MUST be positive")
 
-      new StreamSubscriber[F, A](
-        chunkSize,
-        currentState
+    new StreamSubscriber[F, A](
+      chunkSize,
+      currentState = new AtomicReference[(State, () => Unit)](
+        (State.Uninitialized(cb = None), noop)
       )
-    }
+    )
   }
 
   private sealed abstract class StreamSubscriberException(msg: String, cause: Throwable = null)

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

@@ -22,11 +22,10 @@
 package fs2
 package interop
 
-import cats.effect.IO
-import cats.effect.kernel.{Async, Resource}
+import cats.effect.{Async, IO, Resource}
 import cats.effect.unsafe.IORuntime
 
-import java.util.concurrent.Flow.{Publisher, Subscriber, defaultBufferSize}
+import java.util.concurrent.Flow.{Publisher, Processor, Subscriber, defaultBufferSize}
 
 /** Implementation of the reactive-streams protocol for fs2; based on Java Flow.
   *
@@ -199,6 +198,69 @@ package object flow {
   ): Stream[F, Nothing] =
     StreamSubscription.subscribe(stream, subscriber)
 
+  /** Creates a [[Processor]] from a [[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.
+    *
+    * @see [[unsafePipeToProcessor]] for an unsafe version that returns a plain [[Processor]].
+    *
+    * @param pipe The [[Pipe]] which represents the [[Processor]] 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 pipeToProcessor[F[_], I, O](
+      pipe: Pipe[F, I, O],
+      chunkSize: Int
+  )(implicit
+      F: Async[F]
+  ): Resource[F, Processor[I, O]] =
+    StreamProcessor.fromPipe(pipe, chunkSize)
+
+  /** Creates a [[Processor]] from a [[Pipe]].
+    *
+    * You are required to manually subscribe this [[Processor]] to an upstream [[Publisher]], and have at least one downstream [[Subscriber]] subscribe to the [[Consumer]].
+    *
+    * @see [[pipeToProcessor]] for a safe version that returns a [[Resource]].
+    *
+    * @param pipe The [[Pipe]] which represents the [[Processor]] 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 unsafePipeToProcessor[I, O](
+      pipe: Pipe[IO, I, O],
+      chunkSize: Int
+  )(implicit
+      runtime: IORuntime
+  ): Processor[I, O] =
+    StreamProcessor.unsafeFromPipe(pipe, 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 processorToPipe[F[_]]: syntax.FromProcessorPartiallyApplied[F] =
+    new syntax.FromProcessorPartiallyApplied[F](dummy = true)
+
   /** A default value for the `chunkSize` argument,
     * that may be used in the absence of other constraints;
     * we encourage choosing an appropriate value consciously.

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

@@ -23,9 +23,9 @@ package fs2
 package interop
 package flow
 
-import cats.effect.kernel.{Async, Resource}
+import cats.effect.{Async, Resource}
 
-import java.util.concurrent.Flow.{Publisher, Subscriber}
+import java.util.concurrent.Flow.{Processor, Publisher, Subscriber}
 
 object syntax {
   implicit final class PublisherOps[A](private val publisher: Publisher[A]) extends AnyVal {
@@ -57,4 +57,14 @@ object syntax {
         F.delay(publisher.subscribe(subscriber))
       }
   }
+
+  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 ProcessorPipe(processor, chunkSize)
+  }
 }