diff --git a/README.md b/README.md index 78d1c2d..aa0d771 100644 --- a/README.md +++ b/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 diff --git a/docs/index.md b/docs/index.md index b18924e..24b2283 100644 --- a/docs/index.md +++ b/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 diff --git a/project/build.properties b/project/build.properties index c8fcab5..46e43a9 100644 --- a/project/build.properties +++ b/project/build.properties @@ -1 +1 @@ -sbt.version=1.6.2 +sbt.version=1.8.2 diff --git a/zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/Adapters.scala index 22e1efe..66cedb1 100644 --- a/zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/Adapters.scala +++ b/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) diff --git a/zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/package.scala b/zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/package.scala index 6ac7b4a..6935d5c 100644 --- a/zio-interop-reactivestreams/src/main/scala/zio/interop/reactivestreams/package.scala +++ b/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) + + } + } diff --git a/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala b/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala new file mode 100644 index 0000000..584c1d4 --- /dev/null +++ b/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) + } + } + ) +} diff --git a/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/PublisherToStreamSpec.scala b/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/PublisherToStreamSpec.scala index bef3763..c51affa 100644 --- a/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/PublisherToStreamSpec.scala +++ b/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