Add a channel that outputs to a subscriber.

README.md

@@ -17,7 +17,7 @@ This library provides an interoperability layer between ZIO and reactive streams
 In order to use this library, we need to add the following line in our `build.sbt` file:
 
 ```scala
-libraryDependencies += "dev.zio" %% "zio-interop-reactive-streams" % "2.0.0"
+libraryDependencies += "dev.zio" %% "zio-interop-reactive-streams" % "2.0.1"
 ```
 
 ## Examples
@@ -53,19 +53,16 @@ val streamFromPublisher = publisher.toZIOStream(qSize = 16)
 streamFromPublisher.run(Sink.collectAll[Integer])
 ```
 
-### Subscriber to Sink
+### Channel that outputs to a Subscriber
 
-When running a `Stream` to a `Subscriber`, a side channel is needed for signalling failures.
-For this reason `toZIOSink` returns a tuple of a callback and a `Sink`. The callback must be used to signal `Stream` failure. The type parameter on `toZIOSink` is the error type of *the Stream*.
+`ZChannel.toSubscriber` creates a channel that outputs to a `Subscriber`. The upstream can fail with any `Throwable`, which will be signaled to the subscriber's `onError` method and cause the channel to fail with `Some(throwable)`. If the subscriber cancels its subscription, the channel fails with `None`.
+
+To use the channel as the destination for a stream, one method is to use `pipeThroughChannel` to get the effect of signalling the subscriber, and `runDrain` to run the resulting stream.
 
 ```scala
-val asSink = subscriber.toZIOSink[Throwable]
+val subscriberChannel = ZChannel.toSubscriber(subscriber)
 val failingStream = ZStream.range(3, 13) ++ ZStream.fail(new RuntimeException("boom!"))
-ZIO.scoped {
-  asSink.flatMap { case (signalError, sink) => // FIXME
-    failingStream.run(sink).catchAll(signalError)
-  }
-}
+failingStream.pipeThroughChannel(subscriberChannel).runDrain
 ```
 
 ### Stream to Publisher

docs/index.md

@@ -53,19 +53,16 @@ val streamFromPublisher = publisher.toZIOStream(qSize = 16)
 streamFromPublisher.run(Sink.collectAll[Integer])
 ```
 
-### Subscriber to Sink
+### Channel that outputs to a Subscriber
 
-When running a `Stream` to a `Subscriber`, a side channel is needed for signalling failures.
-For this reason `toZIOSink` returns a tuple of a callback and a `Sink`. The callback must be used to signal `Stream` failure. The type parameter on `toZIOSink` is the error type of *the Stream*.
+`ZChannel.toSubscriber` creates a channel that outputs to a `Subscriber`. The upstream can fail with any `Throwable`, which will be signaled to the subscriber's `onError` method and cause the channel to fail with `Some(throwable)`. If the subscriber cancels its subscription, the channel fails with `None`.
+
+To use the channel as the destination for a stream, one method is to use `pipeThroughChannel` to get the effect of signalling the subscriber, and `runDrain` to run the resulting stream.
 
 ```scala
-val asSink = subscriber.toZIOSink[Throwable]
+val subscriberChannel = ZChannel.toSubscriber(subscriber)
 val failingStream = ZStream.range(3, 13) ++ ZStream.fail(new RuntimeException("boom!"))
-ZIO.scoped {
-  asSink.flatMap { case (signalError, sink) => // FIXME
-    failingStream.run(sink).catchAll(signalError)
-  }
-}
+failingStream.pipeThroughChannel(subscriberChannel).runDrain
 ```
 
 ### Stream to Publisher

project/build.properties

@@ -1 +1 @@
-sbt.version=1.6.2
+sbt.version=1.8.2

zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/Adapters.scala

@@ -49,6 +49,48 @@ object Adapters {
       } yield (error.fail(_) *> fiber.join, demandUnfoldSink(sub, subscription))
     }
 
+  def subscriberToChannel[I](subscriber: => Subscriber[I])(implicit
+    trace: Trace
+  ): ZChannel[Any, Throwable, Chunk[I], Any, Option[Throwable], Nothing, Unit] = unsafe { implicit unsafe =>
+    ZChannel.unwrap {
+      ZIO.suspendSucceed {
+        val sub          = subscriber
+        val subscription = new DemandTrackingSubscription(sub)
+        ZIO.succeed(sub.onSubscribe(subscription)).as {
+          def handleInput(
+            keepReading: => ZChannel[Any, Throwable, Chunk[I], Any, Option[Throwable], Nothing, Unit]
+          )(chunk: Chunk[I]): ZChannel[Any, Throwable, Chunk[I], Any, Option[Throwable], Nothing, Unit] =
+            ZChannel.unwrap {
+              ZIO
+                .iterate(chunk)(!_.isEmpty) { chunk =>
+                  subscription.offer(chunk.size).flatMap { acceptedCount =>
+                    val (send, remain) = chunk.splitAt(acceptedCount)
+                    ZIO.foreachDiscard(send)(a => ZIO.succeed(sub.onNext(a))).as(remain)
+                  }
+                }
+                .fold(
+                  _ => ZChannel.fail(None), // canceled
+                  _ => ZChannel.unit
+                )
+            } *> keepReading
+          def handleError(t: Throwable): ZChannel[Any, Throwable, Chunk[I], Any, Option[Throwable], Nothing, Unit] =
+            ZChannel.succeed {
+              if (!subscription.isCanceled)
+                sub.onError(t)
+            } *> ZChannel.fail(Some(t))
+          val handleDone: Any => ZChannel[Any, Throwable, Chunk[I], Any, Nothing, Nothing, Unit] = _ =>
+            ZChannel.succeed {
+              if (!subscription.isCanceled)
+                sub.onComplete()
+            }
+          lazy val chan: ZChannel[Any, Throwable, Chunk[I], Any, Option[Throwable], Nothing, Unit] = ZChannel
+            .readWith(handleInput(chan), handleError, handleDone)
+          chan
+        }
+      }
+    }
+  }
+
   def publisherToStream[O](
     publisher: => Publisher[O],
     bufferSize: => Int
@@ -77,7 +119,7 @@ object Adapters {
       pull = p.await.flatMap { case (subscription, q) =>
                process(subscription, q, () => subscriber.await(), () => subscriber.isDone, bufferSize)
              }
-               .catchAll(e => ZIO.succeedNow(Pull.fail(e)))
+               .catchAll(e => ZIO.succeed(Pull.fail(e)))
       fiber <- fromPull(pull).run(sink).forkScoped
     } yield (subscriber, fiber.join)
 
@@ -178,7 +220,7 @@ object Adapters {
               if (shouldCancel)
                 s.cancel()
               else
-                p.unsafe.done(ZIO.succeedNow((s, q)))
+                p.unsafe.done(ZIO.succeed((s, q)))
             }
 
           override def onNext(t: A): Unit =
@@ -267,7 +309,7 @@ object Adapters {
         case State(requestedCount, _) =>
           val newRequestedCount = Math.max(requestedCount - n, 0L)
           val accepted          = Math.min(requestedCount, n.toLong).toInt
-          result = ZIO.succeedNow(accepted)
+          result = ZIO.succeed(accepted)
           requested(newRequestedCount)
       }
       result
@@ -285,7 +327,7 @@ object Adapters {
           val newRequestedCount = requestedCount + n
           val accepted          = Math.min(offered.toLong, newRequestedCount)
           val remaining         = newRequestedCount - accepted
-          notification = () => toNotify.unsafe.done(ZIO.succeedNow(accepted.toInt))
+          notification = () => toNotify.unsafe.done(ZIO.succeed(accepted.toInt))
           requested(remaining)
         case State(requestedCount, _) if ((Long.MaxValue - n) > requestedCount) =>
           requested(requestedCount + n)

zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/package.scala

@@ -5,6 +5,8 @@ import org.reactivestreams.Subscriber
 import zio.{ Scope, UIO, Task, ZIO, Trace }
 import zio.stream.ZSink
 import zio.stream.ZStream
+import zio.stream.ZChannel
+import zio.Chunk
 
 package object reactivestreams {
 
@@ -59,4 +61,22 @@ package object reactivestreams {
       Adapters.subscriberToSink(subscriber)
   }
 
+  final implicit class ZChannelInterop(private val zchannel: ZChannel.type) extends AnyVal {
+
+    /** A channel that outputs to a reactive streams subscriber.
+      *
+      * The upstream can fail with any `Throwable`, which will be signalled to the subscriber's `onError` method, and
+      * the channel fails with `Some(throwable)`. If the subscriber cancels its subscription, the channel fails with
+      * `None`.
+      *
+      * @param subscriber
+      *   The reactive streams subscriber to output to.
+      */
+    def toSubscriber[I](subscriber: Subscriber[I])(implicit
+      trace: Trace
+    ): ZChannel[Any, Throwable, Chunk[I], Any, Option[Throwable], Nothing, Unit] =
+      Adapters.subscriberToChannel(subscriber)
+
+  }
+
 }

zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala

@@ -0,0 +1,117 @@
+package zio.interop.reactivestreams
+
+import zio._
+import zio.test._
+import zio.stream._
+import org.reactivestreams.Subscriber
+import org.reactivestreams.Subscription
+
+object ChannelToSubscriberSpec extends ZIOSpecDefault {
+
+  private class TestSubscriber(initialRequest: Int = 1, onNextRequest: Int = 1) extends Subscriber[Int] {
+
+    protected var subscription: Subscription = _
+    private var subscribed                   = false
+    private var values                       = Chunk.empty[Int]
+    private var error                        = Option.empty[Throwable]
+    private var complete                     = false
+
+    override def onSubscribe(s: Subscription): Unit = {
+      subscription = s
+      subscribed = true
+      s.request(initialRequest.toLong)
+    }
+
+    override def onError(t: Throwable): Unit =
+      error = Some(t)
+
+    override def onComplete(): Unit =
+      complete = true
+
+    override def onNext(t: Int): Unit = {
+      values = values :+ t
+      subscription.request(onNextRequest.toLong)
+    }
+
+    final def getState: UIO[(Boolean, Chunk[Int], Option[Throwable], Boolean)] =
+      ZIO.succeed((subscribed, values, error, complete))
+  }
+  override def spec = suite("Channel writing to a subscriber spec")(
+    test("works with a basic subscriber") {
+      val subscriber = new TestSubscriber(100, 1)
+      val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Option[Throwable], Nothing, Unit] =
+        ZChannel.toSubscriber(subscriber)
+      val input                                            = ZStream(1, 2, 3).concat(ZStream(100, 200))
+      val stream: ZStream[Any, Option[Throwable], Nothing] = input.pipeThroughChannel(channel)
+      for {
+        expected <- input.runCollect
+        _        <- ZIO.succeed(println("start"))
+        _        <- stream.runDrain
+        actual   <- subscriber.getState
+      } yield {
+        val (subscribe, values, error, complete) = actual
+        assertTrue(values == expected && subscribe && error.isEmpty && complete)
+      }
+    },
+    test("works with limited subscriber demand") {
+      val subscriber = new TestSubscriber(2, 1)
+      val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Option[Throwable], Nothing, Unit] =
+        ZChannel.toSubscriber(subscriber)
+      val input                                            = ZStream(1, 2, 3, 4, 5, 6)
+      val stream: ZStream[Any, Option[Throwable], Nothing] = input.pipeThroughChannel(channel)
+      for {
+        expected <- input.runCollect
+        _        <- stream.runDrain
+        actual   <- subscriber.getState
+      } yield {
+        val (subscribe, values, error, complete) = actual
+        assertTrue(values == expected && subscribe && error.isEmpty && complete)
+      }
+    },
+    test("signals upstream errors to the subscriber and the downstream") {
+      val subscriber = new TestSubscriber(1, 1)
+      val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Option[Throwable], Nothing, Unit] =
+        ZChannel.toSubscriber(subscriber)
+      val exception = new IllegalStateException("boom")
+      val input     = ZStream(1, 2, 3)
+      val stream: ZStream[Any, Option[Throwable], Nothing] =
+        input.concat(ZStream.fail(exception)).concat(ZStream(100, 200)).pipeThroughChannel(channel)
+      for {
+        expected    <- input.runCollect
+        resultError <- stream.runDrain.flip
+        actual      <- subscriber.getState
+      } yield {
+        val (subscribe, values, error, complete) = actual
+        assertTrue(
+          values == expected && subscribe && error.contains(exception) && !complete && resultError.is(
+            _.some
+          ) == exception
+        )
+      }
+    },
+    test("reports cancellation by the subscriber") {
+      val subscriber = new TestSubscriber(1, 1) {
+        private var count = 0
+        override def onNext(t: Int): Unit = {
+          count += 1
+          if (count > 2)
+            subscription.cancel()
+          else
+            super.onNext(t)
+        }
+      }
+      val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Option[Throwable], Nothing, Unit] =
+        ZChannel.toSubscriber(subscriber)
+      val input                                            = ZStream(1, 2, 3, 4, 5)
+      val stream: ZStream[Any, Option[Throwable], Nothing] = input.pipeThroughChannel(channel)
+      for {
+        expected   <- input.take(2).runCollect
+        errorValue <- stream.runDrain.flip
+        actual     <- subscriber.getState
+      } yield {
+        val (subscribe, values, error, complete) = actual
+        assertTrue(values == expected && subscribe && error.isEmpty && !complete && errorValue.isEmpty)
+      }
+    }
+  )
+}

zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/PublisherToStreamSpec.scala

@@ -107,7 +107,7 @@ object PublisherToStreamSpec extends ZIOSpecDefault {
                              }
               probe = new Publisher[Int] {
                         override def subscribe(subscriber: Subscriber[_ >: Int]): Unit =
-                          subscriberP.unsafe.done(ZIO.succeedNow(subscriber))
+                          subscriberP.unsafe.done(ZIO.succeed(subscriber))
                       }
               fiber      <- probe.toZIOStream(bufferSize).runDrain.fork
               subscriber <- subscriberP.await