From 83e0f63055a4b2402ca3c66dace2c6e966f1e888 Mon Sep 17 00:00:00 2001 From: Lachlan O'Dea Date: Fri, 21 Apr 2023 09:31:18 +1000 Subject: [PATCH 1/4] Channel that outputs to a subscriber. Using a channel to drive a subscriber instead of a sink has the advantage that we can pass upstream errors through to the subscriber's `onError` method without needing the caller to manually install an error handler. While a channel is not as convenient to use as a sink, it is more convenient that the error handler + sink pair. --- docs/index.md | 15 +-- project/build.properties | 2 +- .../interop/reactivestreams/Adapters.scala | 50 +++++++- .../zio/interop/reactivestreams/package.scala | 19 +++ .../ChannelToSubscriberSpec.scala | 113 ++++++++++++++++++ .../PublisherToStreamSpec.scala | 2 +- 6 files changed, 186 insertions(+), 15 deletions(-) create mode 100644 zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala diff --git a/docs/index.md b/docs/index.md index b18924e..876d7c8 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. If the subscriber cancels its subscription, the channel fails with unit. + +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..ab2e8e6 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, Unit, 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, Unit, Nothing, Unit] + )(chunk: Chunk[I]): ZChannel[Any, Throwable, Chunk[I], Any, Unit, 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(()), // canceled + _ => ZChannel.unit + ) + } *> keepReading + def handleError(t: Throwable): ZChannel[Any, Throwable, Chunk[I], Any, Nothing, Nothing, Unit] = + ZChannel.succeed { + if (!subscription.isCanceled) + sub.onError(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, Unit, 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..6209432 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,21 @@ 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 propagated to the subscriber's `onError` method. If + * the subscriber cancels its subscription, the channel fails with unit. + * + * @param subscriber + * The reactive streams subscriber to output to. + */ + def toSubscriber[I](subscriber: Subscriber[I])(implicit + trace: Trace + ): ZChannel[Any, Throwable, Chunk[I], Any, Unit, 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..e4d7c37 --- /dev/null +++ b/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala @@ -0,0 +1,113 @@ +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, Unit, Nothing, Unit] = + ZChannel.toSubscriber(subscriber) + val input = ZStream(1, 2, 3).concat(ZStream(100, 200)) + val stream: ZStream[Any, Unit, 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, Unit, Nothing, Unit] = + ZChannel.toSubscriber(subscriber) + val input = ZStream(1, 2, 3, 4, 5, 6) + val stream: ZStream[Any, Unit, 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") { + val subscriber = new TestSubscriber(1, 1) + val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Unit, Nothing, Unit] = + ZChannel.toSubscriber(subscriber) + val exception = new IllegalStateException("boom") + val input = ZStream(1, 2, 3) + val stream: ZStream[Any, Unit, Nothing] = + input.concat(ZStream.fail(exception)).concat(ZStream(100, 200)).pipeThroughChannel(channel) + for { + expected <- input.runCollect + _ <- stream.runDrain + actual <- subscriber.getState + } yield { + val (subscribe, values, error, complete) = actual + assertTrue(values == expected && subscribe && error.contains(exception) && !complete) + } + }, + 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, Unit, Nothing, Unit] = + ZChannel.toSubscriber(subscriber) + val input = ZStream(1, 2, 3, 4, 5) + val stream: ZStream[Any, Unit, 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 == ()) + } + } + ) +} 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 From 4d9e53d3f1a9ec5c316b1be5f867372c8c55f669 Mon Sep 17 00:00:00 2001 From: Lachlan O'Dea Date: Fri, 21 Apr 2023 10:37:13 +1000 Subject: [PATCH 2/4] Update readme. --- README.md | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 78d1c2d..88ee76c 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. If the subscriber cancels its subscription, the channel fails with unit. + +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 From 104608765f677d0c71af52fc31551ca8e6bd1d6b Mon Sep 17 00:00:00 2001 From: Lachlan O'Dea Date: Fri, 21 Apr 2023 10:56:35 +1000 Subject: [PATCH 3/4] Fix test failures. --- .../src/main/scala/zio/interop/reactivestreams/Adapters.scala | 2 +- .../zio/interop/reactivestreams/ChannelToSubscriberSpec.scala | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 ab2e8e6..98d5b0c 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 @@ -58,7 +58,7 @@ object Adapters { val subscription = new DemandTrackingSubscription(sub) ZIO.succeed(sub.onSubscribe(subscription)).as { def handleInput( - keepReading: ZChannel[Any, Throwable, Chunk[I], Any, Unit, Nothing, Unit] + keepReading: => ZChannel[Any, Throwable, Chunk[I], Any, Unit, Nothing, Unit] )(chunk: Chunk[I]): ZChannel[Any, Throwable, Chunk[I], Any, Unit, Nothing, Unit] = ZChannel.unwrap { ZIO 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 index e4d7c37..7c68095 100644 --- a/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala +++ b/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala @@ -106,7 +106,7 @@ object ChannelToSubscriberSpec extends ZIOSpecDefault { actual <- subscriber.getState } yield { val (subscribe, values, error, complete) = actual - assertTrue(values == expected && subscribe && error.isEmpty && !complete && errorValue == ()) + assertTrue(values == expected && subscribe && error.isEmpty && !complete && errorValue == (())) } } ) From 0b2d2c1d587bcf4eb25546eef3b0630832ee9cf7 Mon Sep 17 00:00:00 2001 From: Lachlan O'Dea Date: Sat, 22 Apr 2023 14:41:34 +1000 Subject: [PATCH 4/4] Propagate upstream errors to downstream. --- README.md | 2 +- docs/index.md | 2 +- .../interop/reactivestreams/Adapters.scala | 14 +++---- .../zio/interop/reactivestreams/package.scala | 7 ++-- .../ChannelToSubscriberSpec.scala | 38 ++++++++++--------- 5 files changed, 34 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index 88ee76c..aa0d771 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ streamFromPublisher.run(Sink.collectAll[Integer]) ### Channel that outputs to a Subscriber -`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. If the subscriber cancels its subscription, the channel fails with unit. +`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. diff --git a/docs/index.md b/docs/index.md index 876d7c8..24b2283 100644 --- a/docs/index.md +++ b/docs/index.md @@ -55,7 +55,7 @@ streamFromPublisher.run(Sink.collectAll[Integer]) ### Channel that outputs to a Subscriber -`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. If the subscriber cancels its subscription, the channel fails with unit. +`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. 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 98d5b0c..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 @@ -51,15 +51,15 @@ object Adapters { def subscriberToChannel[I](subscriber: => Subscriber[I])(implicit trace: Trace - ): ZChannel[Any, Throwable, Chunk[I], Any, Unit, Nothing, Unit] = unsafe { implicit unsafe => + ): 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, Unit, Nothing, Unit] - )(chunk: Chunk[I]): ZChannel[Any, Throwable, Chunk[I], Any, Unit, Nothing, Unit] = + 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 => @@ -69,21 +69,21 @@ object Adapters { } } .fold( - _ => ZChannel.fail(()), // canceled + _ => ZChannel.fail(None), // canceled _ => ZChannel.unit ) } *> keepReading - def handleError(t: Throwable): ZChannel[Any, Throwable, Chunk[I], Any, Nothing, Nothing, Unit] = + 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, Unit, Nothing, Unit] = ZChannel + lazy val chan: ZChannel[Any, Throwable, Chunk[I], Any, Option[Throwable], Nothing, Unit] = ZChannel .readWith(handleInput(chan), handleError, handleDone) chan } 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 6209432..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 @@ -65,15 +65,16 @@ package object reactivestreams { /** A channel that outputs to a reactive streams subscriber. * - * The upstream can fail with any `Throwable`, which will be propagated to the subscriber's `onError` method. If - * the subscriber cancels its subscription, the channel fails with unit. + * 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, Unit, Nothing, Unit] = + ): 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 index 7c68095..584c1d4 100644 --- a/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala +++ b/zio-interop-reactivestreams/src/test/scala/zio/interop/reactivestreams/ChannelToSubscriberSpec.scala @@ -39,10 +39,10 @@ object ChannelToSubscriberSpec extends ZIOSpecDefault { 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, Unit, Nothing, Unit] = + 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, Unit, Nothing] = input.pipeThroughChannel(channel) + 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")) @@ -55,10 +55,10 @@ object ChannelToSubscriberSpec extends ZIOSpecDefault { }, test("works with limited subscriber demand") { val subscriber = new TestSubscriber(2, 1) - val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Unit, Nothing, Unit] = + 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, Unit, Nothing] = input.pipeThroughChannel(channel) + 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 @@ -68,21 +68,25 @@ object ChannelToSubscriberSpec extends ZIOSpecDefault { assertTrue(values == expected && subscribe && error.isEmpty && complete) } }, - test("signals upstream errors to the subscriber") { + test("signals upstream errors to the subscriber and the downstream") { val subscriber = new TestSubscriber(1, 1) - val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Unit, Nothing, Unit] = + 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, Unit, Nothing] = + val stream: ZStream[Any, Option[Throwable], Nothing] = input.concat(ZStream.fail(exception)).concat(ZStream(100, 200)).pipeThroughChannel(channel) for { - expected <- input.runCollect - _ <- stream.runDrain - actual <- subscriber.getState + 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) + assertTrue( + values == expected && subscribe && error.contains(exception) && !complete && resultError.is( + _.some + ) == exception + ) } }, test("reports cancellation by the subscriber") { @@ -96,17 +100,17 @@ object ChannelToSubscriberSpec extends ZIOSpecDefault { super.onNext(t) } } - val channel: ZChannel[Any, Throwable, Chunk[Int], Any, Unit, Nothing, Unit] = + 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, Unit, Nothing] = input.pipeThroughChannel(channel) + 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 == (())) + assertTrue(values == expected && subscribe && error.isEmpty && !complete && errorValue.isEmpty) } } )