From 4908dd9c4d50768c0ea183eff734684a5dafe41e Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Thu, 19 May 2022 21:04:30 +0200 Subject: [PATCH 1/9] add zioToPublisher conversion --- README.md | 13 +++ .../interop/reactivestreams/Adapters.scala | 25 +++++ .../zio/interop/reactivestreams/package.scala | 8 ++ .../reactivestreams/ZioToPublisherSpec.scala | 104 ++++++++++++++++++ 4 files changed, 150 insertions(+) create mode 100644 src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala diff --git a/README.md b/README.md index d94e281..2b3b7d2 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,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..4b6410e 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -33,6 +33,31 @@ object Adapters { } } + def zioToPublisher[R, E <: Throwable, O]( + zio: => ZIO[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 { + val subscription = new DemandTrackingSubscription(subscriber) + runtime.unsafeRunAsync( + for { + _ <- ZIO.succeed(subscriber.onSubscribe(subscription)) + // Do we need to fork here (like in streamToPublisher, above)? + _ <- for { + // the TCK requires that a failing publisher reports an error even before demand was signalled + // -> run the value evaluation and offer in parallel + // -> this has the added benefit that cancelling the subscription will interrupt an ongoing value evaluation + o <- zio.tapError(e => ZIO.succeed(subscriber.onError(e))).mapError(_ => ()) <& subscription.offer(1) + _ <- ZIO.succeed(subscriber.onNext(o)) + _ <- ZIO.succeed(subscriber.onComplete()) + } yield () + } yield () + ) + } + } + def subscriberToSink[E <: Throwable, I]( subscriber: => Subscriber[I] )(implicit trace: Trace): ZIO[Scope, Nothing, (E => UIO[Unit], ZSink[Any, Nothing, I, I, Unit])] = { diff --git a/src/main/scala/zio/interop/reactivestreams/package.scala b/src/main/scala/zio/interop/reactivestreams/package.scala index 6ac7b4a..cf4c19f 100644 --- a/src/main/scala/zio/interop/reactivestreams/package.scala +++ b/src/main/scala/zio/interop/reactivestreams/package.scala @@ -17,6 +17,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 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..0091ebf --- /dev/null +++ b/src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala @@ -0,0 +1,104 @@ +package zio.interop.reactivestreams + +import org.reactivestreams.Publisher +import org.reactivestreams.tck.PublisherVerification.PublisherTestRun +import org.reactivestreams.tck.{ PublisherVerification, TestEnvironment } +import org.testng.SkipException +import org.testng.annotations.Test +import zio.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: _*) + ) + + 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)) + ) + } +} From 417034d48ea7160e62a8158572596b55b80faf25 Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Mon, 30 May 2022 21:41:07 +0200 Subject: [PATCH 2/9] WIP: Use fiber interruption to implement subscription cancellation --- .../interop/reactivestreams/Adapters.scala | 130 ++++++++++++++++-- .../reactivestreams/ZioToPublisherSpec.scala | 27 +++- 2 files changed, 141 insertions(+), 16 deletions(-) diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 4b6410e..45075e6 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -40,20 +40,10 @@ object Adapters { if (subscriber == null) { throw new NullPointerException("Subscriber must not be null.") } else { - val subscription = new DemandTrackingSubscription(subscriber) runtime.unsafeRunAsync( - for { - _ <- ZIO.succeed(subscriber.onSubscribe(subscription)) - // Do we need to fork here (like in streamToPublisher, above)? - _ <- for { - // the TCK requires that a failing publisher reports an error even before demand was signalled - // -> run the value evaluation and offer in parallel - // -> this has the added benefit that cancelling the subscription will interrupt an ongoing value evaluation - o <- zio.tapError(e => ZIO.succeed(subscriber.onError(e))).mapError(_ => ()) <& subscription.offer(1) - _ <- ZIO.succeed(subscriber.onNext(o)) - _ <- ZIO.succeed(subscriber.onComplete()) - } yield () - } yield () + subscribeAndRun(subscriber)(consumer => + (zio <& consumer.offer(1).debug("offered")).flatMap(o => ZIO.succeed(consumer.next(o))) + ) ) } } @@ -318,6 +308,120 @@ 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[Throwable, 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 an keep the current state + result = () => ZIO.fail(new IllegalStateException("")) + 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) + + def cancel(): Unit = + state.get() match { +// case Offering(_, p) => p.unsafeDone(ZIO.fail(new Exception("cancelled"))) + case _ => + } + } + + object ConsumerImpl { + val zero = ZIO.succeedNow(0) + + sealed trait State + case class Requesting(n: Long) extends State + case class Offering(n: Int, p: Promise[Throwable, Int]) extends State + } + + def subscribeAndRun[R, O](subscriber: Subscriber[O])( + run: Consumer[O] => RIO[R, Unit] + ): ZIO[R, Throwable, Unit] = + for { + consumer <- ZIO.succeed(new ConsumerImpl(subscriber)) + + fiber <- run(consumer) + .tapBoth( + e => ZIO.succeed(subscriber.onError(e)).debug("r2"), + _ => ZIO.succeed(subscriber.onComplete()).debug("r3") + ) + .fork + + runtime <- ZIO.runtime[R] + + subscription = new Subscription { + override def request(n: Long): Unit = consumer.request(n) + + override def cancel(): Unit = + try { + // consumer.cancel() // allows to propagate cancellation to pending offers + runtime.unsafeRun(fiber.interrupt.debug("interrupted").fork.debug("forkedInterruped").unit) + } catch { + case t: Throwable => t.printStackTrace(); throw t + } + } + + _ <- ZIO.succeed(subscriber.onSubscribe(subscription)) + } yield { + () + } + 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/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala b/src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala index 0091ebf..81703be 100644 --- a/src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala +++ b/src/test/scala/zio/interop/reactivestreams/ZioToPublisherSpec.scala @@ -1,11 +1,11 @@ package zio.interop.reactivestreams -import org.reactivestreams.Publisher +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.ZIO +import zio.{ Promise, ZIO } import zio.test.Assertion._ import zio.test._ @@ -14,7 +14,28 @@ 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: _*) + 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]) = From 80ba685cc952f61af27a54e510c545b59c7e0944 Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Wed, 1 Jun 2022 08:40:53 +0200 Subject: [PATCH 3/9] Use <* instead of <& because of interruption; simplify implementation --- .../interop/reactivestreams/Adapters.scala | 28 ++++++------------- 1 file changed, 9 insertions(+), 19 deletions(-) diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 45075e6..89e7ec3 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -35,14 +35,15 @@ object Adapters { def zioToPublisher[R, E <: Throwable, O]( zio: => ZIO[R, E, O] - )(implicit trace: Trace): ZIO[R, Nothing, Publisher[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 => - (zio <& consumer.offer(1).debug("offered")).flatMap(o => ZIO.succeed(consumer.next(o))) + // Use <* instead of <& for now (cf. https://github.com/zio/zio/issues/6888) + (zio <* consumer.offer(1)).flatMap(o => ZIO.succeed(consumer.next(o))) ) ) } @@ -345,7 +346,7 @@ object Adapters { Offering(n, p) case state @ Offering(o, previousOfferPromise) => // we are already offering and get another offer - // -> reject the offer an keep the current state + // -> reject the offer and keep the current state result = () => ZIO.fail(new IllegalStateException("")) state } @@ -374,12 +375,6 @@ object Adapters { } override def next(o: O): Unit = subscriber.onNext(o) - - def cancel(): Unit = - state.get() match { -// case Offering(_, p) => p.unsafeDone(ZIO.fail(new Exception("cancelled"))) - case _ => - } } object ConsumerImpl { @@ -392,16 +387,16 @@ object Adapters { def subscribeAndRun[R, O](subscriber: Subscriber[O])( run: Consumer[O] => RIO[R, Unit] - ): ZIO[R, Throwable, Unit] = + ): URIO[R, Unit] = for { consumer <- ZIO.succeed(new ConsumerImpl(subscriber)) fiber <- run(consumer) .tapBoth( - e => ZIO.succeed(subscriber.onError(e)).debug("r2"), - _ => ZIO.succeed(subscriber.onComplete()).debug("r3") + e => ZIO.succeed(subscriber.onError(e)), + _ => ZIO.succeed(subscriber.onComplete()) ) - .fork + .forkDaemon runtime <- ZIO.runtime[R] @@ -409,12 +404,7 @@ object Adapters { override def request(n: Long): Unit = consumer.request(n) override def cancel(): Unit = - try { - // consumer.cancel() // allows to propagate cancellation to pending offers - runtime.unsafeRun(fiber.interrupt.debug("interrupted").fork.debug("forkedInterruped").unit) - } catch { - case t: Throwable => t.printStackTrace(); throw t - } + runtime.unsafeRunAsync(fiber.interrupt) } _ <- ZIO.succeed(subscriber.onSubscribe(subscription)) From b67fe57c3998ce1f661cf8b3854b191a0b9a8491 Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Wed, 1 Jun 2022 08:47:34 +0200 Subject: [PATCH 4/9] Add exception message --- src/main/scala/zio/interop/reactivestreams/Adapters.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 89e7ec3..3c7f000 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -347,7 +347,7 @@ object Adapters { 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("")) + result = () => ZIO.fail(new IllegalStateException("There is already an outstanding offer")) state } result() From b3c3f83e053fc9bc778b06b3048356c0331971d7 Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Wed, 1 Jun 2022 08:57:01 +0200 Subject: [PATCH 5/9] Remove unused member --- src/main/scala/zio/interop/reactivestreams/Adapters.scala | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 3c7f000..a3a9654 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -378,8 +378,6 @@ object Adapters { } object ConsumerImpl { - val zero = ZIO.succeedNow(0) - sealed trait State case class Requesting(n: Long) extends State case class Offering(n: Int, p: Promise[Throwable, Int]) extends State From 8c534d5785f1999d3344150c6a38647667b018e7 Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Wed, 1 Jun 2022 12:58:48 +0200 Subject: [PATCH 6/9] Constrain promise to be non-failing --- src/main/scala/zio/interop/reactivestreams/Adapters.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index a3a9654..4e204d7 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -341,7 +341,7 @@ object Adapters { result = () => ZIO.succeedNow(accepted.toInt) Requesting(r - accepted) case Requesting(_) => - val p = Promise.unsafeMake[Throwable, Int](FiberId.None) + val p = Promise.unsafeMake[Nothing, Int](FiberId.None) result = () => p.await Offering(n, p) case state @ Offering(o, previousOfferPromise) => @@ -379,8 +379,8 @@ object Adapters { object ConsumerImpl { sealed trait State - case class Requesting(n: Long) extends State - case class Offering(n: Int, p: Promise[Throwable, Int]) extends 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])( From 7037718b3245347404c480ff434e7dc5fe025fbd Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Wed, 1 Jun 2022 18:17:51 +0200 Subject: [PATCH 7/9] Add alternative implementations of zioToPublisher and of streamToPublisher --- .../interop/reactivestreams/Adapters.scala | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 4e204d7..33c8058 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -33,6 +33,29 @@ 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 { chunk => + 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)) + } + } + } + } + ) + } + } + def zioToPublisher[R, E <: Throwable, O]( zio: => ZIO[R, E, O] )(implicit trace: Trace): URIO[R, Publisher[O]] = @@ -49,6 +72,50 @@ object Adapters { } } + 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])] = { From 5d70b4bfb6bb1f4875eaf1d5068f3ddaf9af62ec Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Sun, 5 Jun 2022 22:51:04 +0200 Subject: [PATCH 8/9] add subscriberToChannel adapter --- .../interop/reactivestreams/Adapters.scala | 67 +++++++-- .../zio/interop/reactivestreams/package.scala | 10 +- .../SubscriberToChannelSpec.scala | 129 ++++++++++++++++++ 3 files changed, 194 insertions(+), 12 deletions(-) create mode 100644 src/test/scala/zio/interop/reactivestreams/SubscriberToChannelSpec.scala diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 33c8058..18eeddb 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 @@ -42,15 +43,7 @@ object Adapters { } else { runtime.unsafeRunAsync( subscribeAndRun(subscriber) { consumer => - stream.runForeachChunk { chunk => - 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)) - } - } - } + stream.runForeachChunk(processChunk(consumer, _)) } ) } @@ -128,6 +121,52 @@ object Adapters { } yield (error.fail(_) *> fiber.join, demandUnfoldSink(sub, subscription)) } + def subscriberToChannel[I]( + subscriber: => Subscriber[I] + )(implicit trace: Trace): ZIO[Any, Nothing, 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 @@ -477,6 +516,16 @@ object Adapters { () } + 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 cf4c19f..7befcb8 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 { @@ -65,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 + ): ZIO[Scope, Nothing, 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) + +} From 2d2f61d35a59a77582e1bdc14491475e9277b7cb Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Mon, 6 Jun 2022 11:59:06 +0200 Subject: [PATCH 9/9] WIP: Improve docu; SubscriberToChannelSpec "interruption causes an InterruptionException" still fails --- README.md | 36 ++++++++++++++++--- .../interop/reactivestreams/Adapters.scala | 2 +- .../zio/interop/reactivestreams/package.scala | 2 +- 3 files changed, 34 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 2b3b7d2..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 diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 18eeddb..4e60d1a 100644 --- a/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/src/main/scala/zio/interop/reactivestreams/Adapters.scala @@ -123,7 +123,7 @@ object Adapters { def subscriberToChannel[I]( subscriber: => Subscriber[I] - )(implicit trace: Trace): ZIO[Any, Nothing, ZChannel[Any, Throwable, Chunk[I], Any, Throwable, Chunk[Unit], Any]] = + )(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)) diff --git a/src/main/scala/zio/interop/reactivestreams/package.scala b/src/main/scala/zio/interop/reactivestreams/package.scala index 7befcb8..7d8ed58 100644 --- a/src/main/scala/zio/interop/reactivestreams/package.scala +++ b/src/main/scala/zio/interop/reactivestreams/package.scala @@ -67,7 +67,7 @@ package object reactivestreams { def toZIOChannel(implicit trace: Trace - ): ZIO[Scope, Nothing, ZChannel[Any, Throwable, Chunk[I], Any, Throwable, Chunk[Unit], Any]] = + ): UIO[ZChannel[Any, Throwable, Chunk[I], Any, Throwable, Chunk[Unit], Any]] = Adapters.subscriberToChannel(subscriber) }