Add interop.flow.pipeToProcessor

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

@@ -0,0 +1,44 @@
+package fs2
+package interop
+package flow
+
+import java.util.concurrent.Flow
+import cats.effect.{Async, Resource}
+
+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 <: Object]): 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
+    )
+}

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

@@ -278,7 +278,9 @@ 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]] = {
+  )(implicit
+      F: Async[F]
+  ): F[StreamSubscriber[F, A]] = {
     require(chunkSize > 0, "The buffer size MUST be positive")
 
     F.delay {

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

@@ -22,10 +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
 import java.util.concurrent.Flow.{Publisher, Subscriber, defaultBufferSize}
 
 /** Implementation of the reactive-streams protocol for fs2; based on Java Flow.
@@ -199,6 +199,15 @@ package object flow {
   ): Stream[F, Nothing] =
     StreamSubscription.subscribe(stream, subscriber)
 
+  /** TODO. */
+  def pipeToProcessor[F[_], I, O](
+      pipe: Pipe[F, I, O],
+      chunkSize: Int
+  )(implicit
+      F: Async[F]
+  ): Resource[F, Flow.Processor[I, O]] =
+    StreamProcessor.fromPipe(pipe, chunkSize)
+
   /** A default value for the `chunkSize` argument,
     * that may be used in the absence of other constraints;
     * we encourage choosing an appropriate value consciously.