diff --git a/README.md b/README.md index d94e281..8ddcbf2 100644 --- a/README.md +++ b/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 diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index a2c55ff..4e60d1a 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/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] = diff --git a/src/main/scala/zio/interop/reactivestreams/package.scala b/src/main/scala/zio/interop/reactivestreams/package.scala index 6ac7b4a..7d8ed58 100644 --- a/src/main/scala/zio/interop/reactivestreams/package.scala +++ b/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) } } diff --git a/src/test/scala/zio/interop/reactivestreams/SubscriberToChannelSpec.scala b/src/test/scala/zio/interop/reactivestreams/SubscriberToChannelSpec.scala new file mode 100644 index 0000000..8af71f0 --- /dev/null +++ b/src/test/scala/zio/interop/reactivestreams/SubscriberToChannelSpec.scala @@ -0,0 +1,129 @@ +package zio.interop.reactivestreams + +import org.reactivestreams.tck.TestEnvironment +import org.reactivestreams.tck.TestEnvironment.ManualSubscriberWithSubscriptionSupport +import zio.stream.ZStream +import zio.test.Assertion._ +import zio.test.TestAspect.nonFlaky +import zio.test._ +import zio.{ IO, UIO, ZIO, durationInt } + +import java.util.concurrent.CancellationException +import scala.jdk.CollectionConverters._ + +object SubscriberToChannelSpec extends ZIOSpecDefault { + override def spec = + suite("Converting a `Subscriber` to a `Channel`")( + test("works on the happy path") { + makeSubscriber.flatMap { probe => + probe.underlying.toZIOChannel.flatMap { channel => + for { + fiber <- ZStream.fromIterable(seq).pipeThroughChannel(channel).runDrain.fork + _ <- probe.request(length + 1) + elements <- probe.nextElements(length).exit + completion <- probe.expectCompletion.exit + _ <- fiber.join + } yield assert(elements)(succeeds(equalTo(seq))) && assert(completion)(succeeds(isUnit)) + } + } + }, + test("works on the happy path 2") { + makeSubscriber.flatMap { probe => + probe.underlying.toZIOChannel.flatMap { channel => + for { + fiber <- ZStream.fromIterable(seq).pipeThroughChannel(channel).runDrain.fork + _ <- probe.request(length) + elements <- probe.nextElements(length).exit + completion <- probe.expectCompletion.exit + _ <- fiber.join + } yield assert(elements)(succeeds(equalTo(seq))) && assert(completion)(succeeds(isUnit)) + } + } + }, + test("transports errors") { + makeSubscriber.flatMap { probe => + probe.underlying.toZIOChannel.flatMap { channel => + for { + fiber <- (ZStream.fromIterable(seq) ++ ZStream.fail(e)).pipeThroughChannel(channel).runDrain.fork + _ <- probe.request(length + 1) + elements <- probe.nextElements(length).exit + err <- probe.expectError.exit + exit <- fiber.await + } yield assert(elements)(succeeds(equalTo(seq))) && assert(err)(succeeds(equalTo(e))) && assert(exit)( + fails(equalTo(e)) + ) + } + } + }, + test("transports errors 2") { + makeSubscriber.flatMap { probe => + probe.underlying.toZIOChannel.flatMap { channel => + for { + fiber <- ZStream.fail(e).pipeThroughChannel(channel).runDrain.fork + err <- probe.expectError.exit + exit <- fiber.await + } yield assert(err)(succeeds(equalTo(e))) && assert(exit)(fails(equalTo(e))) + } + } + }, + test("transports errors 3") { + makeSubscriber.flatMap { probe => + for { + fiber <- probe.underlying.toZIOChannel.flatMap { channel => + ZStream.fail(e).pipeThroughChannel(channel).runDrain + }.fork + exit <- fiber.await + err <- probe.expectError.exit + } yield assert(err)(succeeds(equalTo(e))) && assert(exit)(fails(equalTo(e))) + } + } @@ nonFlaky(10), + test("transports errors only once") { + ZIO.scoped[Any] { + for { + probe <- makeSubscriber + channel <- probe.underlying.toZIOChannel + _ <- ZStream.fail(e).pipeThroughChannel(channel).runDrain.fork + err <- probe.expectError.exit + err2 <- probe.expectError.timeout(100.millis).exit + } yield assert(err)(succeeds(equalTo(e))) && assert(err2)(fails(anything)) + } + }, + test("cancellation causes a CancellationException") { + for { + probe <- makeSubscriber + channel <- probe.underlying.toZIOChannel + fiber <- ZStream.never.pipeThroughChannel(channel).runDrain.fork + _ = probe.underlying.cancel() + exit <- fiber.await + } yield assert(exit)(fails(isSubtype[CancellationException](anything))) + }, + test("interruption causes an InterruptedException") { + for { + probe <- makeSubscriber + channel <- probe.underlying.toZIOChannel + fiber <- ZStream.never.pipeThroughChannel(channel).runDrain.fork + _ <- fiber.interrupt + exit <- probe.expectError.exit + } yield assert(exit)(fails(isSubtype[InterruptedException](anything))) + } + ) + + val seq: List[Int] = List.range(0, 31) + val length: Long = seq.length.toLong + val e: Throwable = new RuntimeException("boom") + + case class Probe[T](underlying: ManualSubscriberWithSubscriptionSupport[T]) { + def request(n: Long): UIO[Unit] = + ZIO.succeed(underlying.request(n)) + def nextElements(n: Long): IO[Throwable, List[T]] = + ZIO.attemptBlockingInterrupt(underlying.nextElements(n.toLong).asScala.toList) + def expectError: IO[Throwable, Throwable] = + ZIO.attemptBlockingInterrupt(underlying.expectError(classOf[Throwable])) + def expectCompletion: IO[Throwable, Unit] = + ZIO.attemptBlockingInterrupt(underlying.expectCompletion()) + } + + val makeSubscriber = + ZIO.succeed(new ManualSubscriberWithSubscriptionSupport[Int](new TestEnvironment(2000))).map(Probe.apply) + +} diff --git a/src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala b/src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala new file mode 100644 index 0000000..81703be --- /dev/null +++ b/src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala @@ -0,0 +1,125 @@ +package zio.interop.reactivestreams + +import org.reactivestreams.{ Publisher, Subscriber, Subscription } +import org.reactivestreams.tck.PublisherVerification.PublisherTestRun +import org.reactivestreams.tck.{ PublisherVerification, TestEnvironment } +import org.testng.SkipException +import org.testng.annotations.Test +import zio.{ Promise, ZIO } +import zio.test.Assertion._ +import zio.test._ + +import java.lang.reflect.InvocationTargetException + +object ZioToPublisherSpec extends ZIOSpecDefault { + override def spec = + suite("Converting a `ZIO` to a `Publisher`")( + suite("passes all required and optional TCK tests that are applicable to streams of length 1")(tests: _*), + test("interrupts evaluation on cancellation") { + for { + interruptedPromise <- Promise.make[Nothing, Unit] + subscriptionPromise <- Promise.make[Nothing, Subscription] + z = ZIO.never.as(0).onInterrupt(interruptedPromise.complete(ZIO.succeed(()))) + publisher <- z.toPublisher + subscriber = new Subscriber[Int] { + override def onSubscribe(s: Subscription): Unit = + subscriptionPromise.unsafeDone(ZIO.succeed(s)) + override def onNext(t: Int): Unit = ??? + override def onError(t: Throwable): Unit = ??? + override def onComplete(): Unit = ??? + } + _ <- ZIO.succeed(publisher.subscribe(subscriber)) + subscription <- subscriptionPromise.await + _ <- ZIO.succeed(subscription.cancel()) + unit <- interruptedPromise.await + } yield { + assert(unit)(equalTo(())) + } + } + ) + + def makePV(runtime: zio.Runtime[Any]) = + new PublisherVerification[Int](new TestEnvironment(2000, 500), 2000L) { + + override def maxElementsFromPublisher(): Long = 1 + + override def activePublisherTest( + elements: Long, + completionSignalRequired: Boolean, + body: PublisherTestRun[Int] + ): Unit = + if (elements < 1) { + throw new SkipException( + String.format( + "Unable to run this test, as required elements nr: %d is lower than supported by given producer: %d", + elements, + 1 + ) + ); + } else { + super.activePublisherTest(elements, completionSignalRequired, body) + } + + override def optionalActivePublisherTest( + elements: Long, + completionSignalRequired: Boolean, + body: PublisherTestRun[Int] + ): Unit = + if (elements < 1) { + throw new SkipException( + String.format( + "Unable to run this test, as required elements nr: %d is lower than supported by given producer: %d", + elements, + 1 + ) + ); + } else { + super.optionalActivePublisherTest(elements, completionSignalRequired, body) + } + + def createPublisher(elements: Long): Publisher[Int] = + if (elements == 1) { + runtime.unsafeRun( + ZIO.succeed(1).toPublisher + ) + } else { + throw new IllegalArgumentException("Only publishers for one value are possible.") + } + + override def createFailedPublisher(): Publisher[Int] = + runtime.unsafeRun( + ZIO + .fail(new RuntimeException("boom!")) + .map(_.asInstanceOf[Int]) + .toPublisher + ) + } + + val tests = + classOf[PublisherVerification[Int]] + .getMethods() + .toList + .filter { method => + method + .getAnnotations() + .exists(annotation => classOf[Test].isAssignableFrom(annotation.annotationType())) + } + .collect { + case method if method.getName().startsWith("untested") => + test(method.getName())(assert(())(anything)) @@ TestAspect.ignore + case method => + test(method.getName())( + for { + runtime <- ZIO.runtime[Any] + pv = makePV(runtime) + _ <- ZIO.succeed(pv.setUp()) + r <- ZIO + .attemptBlockingInterrupt(method.invoke(pv)) + .unit + .refineOrDie { case e: InvocationTargetException => e.getTargetException() } + .catchSome { case _: SkipException => ZIO.succeed(()) } + .exit + } yield assert(r)(succeeds(isUnit)) + ) + } +}