add zioToPublisher conversion

README.md

@@ -9,9 +9,18 @@ This library provides an interoperability layer between ZIO and reactive streams
 
 ## Reactive Streams `Producer` and `Subscriber`
 
-**ZIO** integrates with [Reactive Streams](http://reactivestreams.org) by providing conversions from `zio.stream.Stream` to `org.reactivestreams.Publisher`
-and from `zio.stream.Sink` to `org.reactivestreams.Subscriber` and vice versa. Simply import `import zio.interop.reactivestreams._` to make the
-conversions available.
+**ZIO** integrates with [Reactive Streams](http://reactivestreams.org) by providing various conversions:
+
+| from | to                               |
+| :--- |:---------------------------------|
+| `zio.stream.ZStream` | `org.reactivestreams.Publisher`  |
+| `zio.ZIO` | `org.reactivestreams.Publisher`  |
+| `zio.stream.Sink` | `org.reactivestreams.Subscriber` |
+| `org.reactivestreams.Publisher` | `zio.stream.ZStream`             |
+| `org.reactivestreams.Subscriber` | `zio.stream.ZSink`               |
+| `org.reactivestreams.Subscriber` | `zio.stream.ZChannel`            |
+
+Simply import `import zio.interop.reactivestreams._` to make the conversions available.
 
 ## Examples
 
@@ -23,7 +32,7 @@ import zio._
 import zio.interop.reactivestreams._
 import zio.stream._
 
-val runtime = new DefaultRuntime {}
+val runtime = Runtime.default
 ```
 
 We use the following `Publisher` and `Subscriber` for the examples: 
@@ -65,6 +74,25 @@ runtime.unsafeRun(
 )
 ```
 
+### Subscriber to Channel
+
+A `Subscriber` can be converted into a `ZChannel` that supports `Throwable` as input error. Converting a `Subscriber` into a `ZChannel` may be convenient if the input `ZStream` can output throwables. In that case, no additional side channel for signalling failures is needed.
+
+```scala mdoc
+val asChannel = subscriber.toZIOChannel
+val stream    = ZStream.range(3, 13) ++ ZStream.fail(new Exception("boom"))
+
+val exit = runtime.unsafeRun(
+  asChannel.flatMap { channel =>
+    stream.pipeThroughChannel(channel).runDrain.exit
+  }
+)
+println(exit)
+
+An exception is passed to `Subscriber.onError` and is also reflected in the exit value when running the stream.
+```
+
+
 ### Stream to Publisher
 
 ```scala mdoc
@@ -76,6 +104,19 @@ runtime.unsafeRun(
 )
 ```
 
+### ZIO to Publisher
+
+It is also possible to publish a single `ZIO`. In that case, the publisher emits at most one value.
+
+```scala mdoc
+val z = ZIO.succeed(1)
+runtime.unsafeRun(
+  z.toPublisher.flatMap { publisher =>
+    ZIO.succeed(publisher.subscribe(subscriber))
+  }
+)
+```
+
 ### Sink to Subscriber
 
 `toSubscriber` returns a `Subscriber` and an `IO` which completes with the result of running the 

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

@@ -8,6 +8,7 @@ import zio.internal.RingBuffer
 import zio.stream._
 import zio.stream.ZStream.Pull
 
+import java.util.concurrent.CancellationException
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
 
@@ -33,6 +34,81 @@ object Adapters {
       }
     }
 
+  def streamToPublisher2[R, E <: Throwable, O](
+    stream: => ZStream[R, E, O]
+  )(implicit trace: Trace): ZIO[R, Nothing, Publisher[O]] =
+    ZIO.runtime.map { runtime => subscriber =>
+      if (subscriber == null) {
+        throw new NullPointerException("Subscriber must not be null.")
+      } else {
+        runtime.unsafeRunAsync(
+          subscribeAndRun(subscriber) { consumer =>
+            stream.runForeachChunk(processChunk(consumer, _))
+          }
+        )
+      }
+    }
+
+  def zioToPublisher[R, E <: Throwable, O](
+    zio: => ZIO[R, E, O]
+  )(implicit trace: Trace): URIO[R, Publisher[O]] =
+    ZIO.runtime.map { runtime => subscriber =>
+      if (subscriber == null) {
+        throw new NullPointerException("Subscriber must not be null.")
+      } else {
+        runtime.unsafeRunAsync(
+          subscribeAndRun(subscriber)(consumer =>
+            // Use <* instead of <& for now (cf. https://github.com/zio/zio/issues/6888)
+            (zio <* consumer.offer(1)).flatMap(o => ZIO.succeed(consumer.next(o)))
+          )
+        )
+      }
+    }
+
+  def zioToPublisher2[R, E <: Throwable, O](
+    zio: => ZIO[R, E, O]
+  )(implicit trace: Trace): URIO[R, Publisher[O]] =
+    ZIO.runtime.map { runtime => subscriber =>
+      if (subscriber == null) {
+        throw new NullPointerException("Subscriber must not be null.")
+      } else {
+        val cancellationHookRef = new AtomicReference[FiberId => Exit[Any, Unit]]
+
+        val subscription = new Subscription {
+
+          override def cancel(): Unit = {
+            val hook = cancellationHookRef.compareAndExchange(null, _ => Exit.interrupt(FiberId.None))
+            if (hook != null) {
+              hook(FiberId.None)
+              ()
+            }
+          }
+
+          override def request(n: Long): Unit =
+            if (n <= 0) throw new RuntimeException("demand must be > 0")
+            else {
+              if (cancellationHookRef.get == null) {
+                val latch = Promise.unsafeMake[Unit, Unit](FiberId.None)
+                val hook =
+                  runtime.unsafeRunAsyncCancelable(
+                    latch.await *>
+                      zio
+                        .foldZIO(
+                          e => ZIO.succeed(subscriber.onError(e)),
+                          a => ZIO.succeed(subscriber.onNext(a)) *> ZIO.succeed(subscriber.onComplete())
+                        )
+                  )(_ => ())
+                if (cancellationHookRef.compareAndSet(null, hook)) latch.unsafeDone(ZIO.succeedNow(()))
+                else latch.unsafeDone(ZIO.fail(()))
+              }
+            }
+
+        }
+
+        subscriber.onSubscribe(subscription)
+      }
+    }
+
   def subscriberToSink[E <: Throwable, I](
     subscriber: => Subscriber[I]
   )(implicit trace: Trace): ZIO[Scope, Nothing, (E => UIO[Unit], ZSink[Any, Nothing, I, I, Unit])] = {
@@ -45,6 +121,52 @@ object Adapters {
     } yield (error.fail(_) *> fiber.join, demandUnfoldSink(sub, subscription))
   }
 
+  def subscriberToChannel[I](
+    subscriber: => Subscriber[I]
+  )(implicit trace: Trace): UIO[ZChannel[Any, Throwable, Chunk[I], Any, Throwable, Chunk[Unit], Any]] =
+    for {
+      subscriber          <- ZIO.succeed(subscriber)
+      consumer            <- ZIO.succeed(new ConsumerImpl(subscriber))
+      cancellationPromise <- Promise.make[Throwable, Any]
+
+      subscription = new Subscription {
+                       override def request(n: Long): Unit = consumer.request(n)
+                       override def cancel(): Unit =
+                         cancellationPromise.unsafeDone(
+                           ZIO.fail(new CancellationException("Subscription was cancelled"))
+                         )
+                     }
+
+      _ <- ZIO.succeed(subscriber.onSubscribe(subscription))
+
+    } yield {
+
+      lazy val process: ZChannel[Any, Throwable, Chunk[I], Any, Throwable, Chunk[Unit], Any] =
+        ZChannel
+          .readWithCause[Any, Throwable, Chunk[I], Any, Throwable, Chunk[Unit], Any](
+            in => ZChannel.fromZIO(processChunk(consumer, in)) *> process,
+            halt => {
+              val throwable = halt.dieOption match {
+                case Some(throwable) => throwable
+                case _ =>
+                  halt.failureOption match {
+                    case Some(throwable) => throwable
+                    case _               => new InterruptedException("ZChannel was interrupted")
+                  }
+              }
+              subscriber.onError(throwable)
+              ZChannel.failCause(halt)
+            },
+            _ => {
+              subscriber.onComplete()
+              ZChannel.succeed(())
+            }
+          )
+          .interruptWhen(cancellationPromise)
+
+      process
+    }
+
   def publisherToStream[O](
     publisher: => Publisher[O],
     bufferSize: => Int
@@ -293,6 +415,117 @@ object Adapters {
       state.getAndSet(canceled).toNotify.foreach { case (_, p) => p.unsafeDone(ZIO.fail(())) }
   }
 
+  trait Consumer[-O] {
+
+    /** Offers a number of outputs.
+      *
+      * @param n
+      *   The number of offered elements; must be > 0
+      * @return
+      *   Returns the accepted number elements which is in the range 0 < accepted <= n. The `next` method must be called
+      *   exactly the accepted number of times before `offer` is called the next time.
+      */
+    def offer(n: Int): Task[Int]
+
+    /** Signals the next output.
+      * @param o
+      */
+    def next(o: O): Unit
+  }
+
+  private class ConsumerImpl[-O](subscriber: Subscriber[O]) extends Consumer[O] {
+    import ConsumerImpl._
+
+    val state = new AtomicReference(Requesting(0): State)
+
+    override def offer(n: Int): Task[Int] = n match {
+      case n if n > 0 =>
+        var result: () => Task[Int] = null
+        state.updateAndGet {
+          case Requesting(r) if r > 0 =>
+            val accepted = Math.min(n.toLong, r)
+            result = () => ZIO.succeedNow(accepted.toInt)
+            Requesting(r - accepted)
+          case Requesting(_) =>
+            val p = Promise.unsafeMake[Nothing, Int](FiberId.None)
+            result = () => p.await
+            Offering(n, p)
+          case state @ Offering(o, previousOfferPromise) =>
+            // we are already offering and get another offer
+            // -> reject the offer and keep the current state
+            result = () => ZIO.fail(new IllegalStateException("There is already an outstanding offer"))
+            state
+        }
+        result()
+      case _ =>
+        ZIO.fail(new IllegalArgumentException(s"offer must be greater than 0 - offer: $n"))
+    }
+
+    def request(n: Long): Unit = {
+      if (n <= 0) subscriber.onError(new IllegalArgumentException("non-positive subscription request"))
+      var notification: () => Unit = () => ()
+      state.getAndUpdate {
+        case Requesting(r) =>
+          notification = () => ()
+          if (Long.MaxValue - n > r) {
+            Requesting(r + n)
+          } else {
+            Requesting(Long.MaxValue)
+          }
+        case Offering(o, p) =>
+          val accepted = Math.min(n, o.toLong)
+          notification = () => p.unsafeDone(ZIO.succeedNow(accepted.toInt))
+          Requesting(n - accepted)
+      }
+      notification()
+    }
+
+    override def next(o: O): Unit = subscriber.onNext(o)
+  }
+
+  object ConsumerImpl {
+    sealed trait State
+    case class Requesting(n: Long)                        extends State
+    case class Offering(n: Int, p: Promise[Nothing, Int]) extends State
+  }
+
+  def subscribeAndRun[R, O](subscriber: Subscriber[O])(
+    run: Consumer[O] => RIO[R, Unit]
+  ): URIO[R, Unit] =
+    for {
+      consumer <- ZIO.succeed(new ConsumerImpl(subscriber))
+
+      fiber <- run(consumer)
+                 .tapBoth(
+                   e => ZIO.succeed(subscriber.onError(e)),
+                   _ => ZIO.succeed(subscriber.onComplete())
+                 )
+                 .forkDaemon
+
+      runtime <- ZIO.runtime[R]
+
+      subscription = new Subscription {
+                       override def request(n: Long): Unit = consumer.request(n)
+
+                       override def cancel(): Unit =
+                         runtime.unsafeRunAsync(fiber.interrupt)
+                     }
+
+      _ <- ZIO.succeed(subscriber.onSubscribe(subscription))
+    } yield {
+      ()
+    }
+
+  def processChunk[I](consumer: Consumer[I], chunk: Chunk[I]): Task[Unit] = ZIO
+    .iterate(chunk)(!_.isEmpty) { chunk =>
+      consumer.offer(chunk.size).flatMap { acceptedCount =>
+        ZIO
+          .foreach(chunk.take(acceptedCount))(a => ZIO.succeed(consumer.next(a)))
+          .as(chunk.drop(acceptedCount))
+      }
+    }
+    .unit
+
   private def fromPull[R, E, A](zio: ZIO[R with Scope, Nothing, ZIO[R, Option[E], Chunk[A]]])(implicit
     trace: Trace
   ): ZStream[R, E, A] =

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

@@ -2,9 +2,8 @@ package zio.interop
 
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
-import zio.{ Scope, UIO, Task, ZIO, Trace }
-import zio.stream.ZSink
-import zio.stream.ZStream
+import zio.{ Chunk, Scope, Task, Trace, UIO, ZIO }
+import zio.stream.{ ZChannel, ZSink, ZStream }
 
 package object reactivestreams {
 
@@ -17,6 +16,14 @@ package object reactivestreams {
       Adapters.streamToPublisher(stream)
   }
 
+  /** Creates a `Publisher` from a `ZIO` that publishes the ZIO's value. Every time the `Publisher` is subscribed to, a
+    * new instance of the `ZIO` is run.
+    */
+  final implicit class zioToPublisher[R, E <: Throwable, O](private val zio: ZIO[R, E, O]) extends AnyVal {
+    def toPublisher(implicit trace: Trace): ZIO[R, Nothing, Publisher[O]] =
+      Adapters.zioToPublisher(zio)
+  }
+
   final implicit class sinkToSubscriber[R, E <: Throwable, A, L, Z](private val sink: ZSink[R, E, A, L, Z]) {
 
     /** Create a `Subscriber` from a `Sink`. The returned Task will eventually return the result of running the
@@ -57,6 +64,11 @@ package object reactivestreams {
       trace: Trace
     ): ZIO[Scope, Nothing, (E => UIO[Unit], ZSink[Any, Nothing, I, I, Unit])] =
       Adapters.subscriberToSink(subscriber)
+
+    def toZIOChannel(implicit
+      trace: Trace
+    ): UIO[ZChannel[Any, Throwable, Chunk[I], Any, Throwable, Chunk[Unit], Any]] =
+      Adapters.subscriberToChannel(subscriber)
   }
 
 }