fix: signal error without Promise in subscriberToSink (#298)

* fix: signal error without Promise in subscriberToSink * fix: 3.1.0 compilation
zio · Dec 19, 2021 · 15e7e03 · 15e7e03
1 parent 3c6afc9
commit 15e7e03
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 12 deletions.
diff --git a/src/main/scala/zio/interop/reactivestreams/Adapters.scala b/src/main/scala/zio/interop/reactivestreams/Adapters.scala
@@ -35,16 +35,20 @@ object Adapters {
 
   def subscriberToSink[E <: Throwable, I](
     subscriber: => Subscriber[I]
-  )(implicit trace: ZTraceElement): ZManaged[Any, Nothing, (Promise[E, Nothing], ZSink[Any, Nothing, I, I, Unit])] = {
+  )(implicit trace: ZTraceElement): ZManaged[Any, Nothing, (E => UIO[Unit], ZSink[Any, Nothing, I, I, Unit])] = {
     val sub = subscriber
     for {
-      runtime     <- ZIO.runtime[Any].toManaged
-      demand      <- Queue.unbounded[Long].toManaged
-      error       <- Promise.make[E, Nothing].toManaged
-      subscription = createSubscription(sub, demand, runtime)
-      _           <- ZManaged.succeed(sub.onSubscribe(subscription))
-      _           <- error.await.catchAll(t => UIO(sub.onError(t)) *> demand.shutdown).toManaged.fork
-    } yield (error, demandUnfoldSink(sub, demand))
+      runtime       <- ZIO.runtime[Any].toManaged
+      demand        <- Queue.unbounded[Long].toManaged
+      subscription   = createSubscription(sub, demand, runtime)
+      _             <- ZManaged.succeed(sub.onSubscribe(subscription))
+      errorSignaled <- Promise.makeManaged[Nothing, Boolean]
+    } yield {
+      val signalError =
+        (e: E) => ZIO.whenZIO(errorSignaled.complete(UIO.succeedNow(true)))(UIO(sub.onError(e)) *> demand.shutdown).unit
+
+      (signalError, demandUnfoldSink(sub, demand))
+    }
   }
 
   def publisherToStream[O](

diff --git a/src/main/scala/zio/interop/reactivestreams/package.scala b/src/main/scala/zio/interop/reactivestreams/package.scala
@@ -3,7 +3,7 @@ package zio.interop
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
 import zio.IO
-import zio.Promise
+import zio.UIO
 import zio.ZIO
 import zio.ZManaged
 import zio.ZTraceElement
@@ -60,7 +60,7 @@ package object reactivestreams {
       */
     def toSink[E <: Throwable](implicit
       trace: ZTraceElement
-    ): ZManaged[Any, Nothing, (Promise[E, Nothing], ZSink[Any, Nothing, I, I, Unit])] =
+    ): ZManaged[Any, Nothing, (E => UIO[Unit], ZSink[Any, Nothing, I, I, Unit])] =
       Adapters.subscriberToSink(subscriber)
   }
 

diff --git a/src/test/scala/zio/interop/reactivestreams/SubscriberToSinkSpec.scala b/src/test/scala/zio/interop/reactivestreams/SubscriberToSinkSpec.scala
@@ -2,10 +2,12 @@ package zio.interop.reactivestreams
 
 import org.reactivestreams.tck.TestEnvironment
 import org.reactivestreams.tck.TestEnvironment.ManualSubscriberWithSubscriptionSupport
+import zio.durationInt
 import zio.IO
 import zio.Task
 import zio.UIO
 import zio.stream.Stream
+import zio.stream.ZStream
 import zio.test.Assertion._
 import zio.test._
 
@@ -33,16 +35,42 @@ object SubscriberToSinkSpec extends DefaultRunnableSpec {
         makeSubscriber.flatMap(probe =>
           probe.underlying
             .toSink[Throwable]
-            .use { case (error, sink) =>
+            .use { case (signalError, sink) =>
               for {
-                fiber    <- (Stream.fromIterable(seq) ++ Stream.fail(e)).run(sink).catchAll(t => error.fail(t)).fork
+                fiber    <- (Stream.fromIterable(seq) ++ Stream.fail(e)).run(sink).catchAll(signalError).fork
                 _        <- probe.request(length + 1)
                 elements <- probe.nextElements(length).exit
                 err      <- probe.expectError.exit
                 _        <- fiber.join
               } yield assert(elements)(succeeds(equalTo(seq))) && assert(err)(succeeds(equalTo(e)))
             }
         )
+      },
+      test("transports errors 2") {
+        makeSubscriber.flatMap(probe =>
+          probe.underlying
+            .toSink[Throwable]
+            .use { case (signalError, sink) =>
+              for {
+                _   <- ZStream.fail(e).run(sink).catchAll(signalError)
+                err <- probe.expectError.exit
+              } yield assert(err)(succeeds(equalTo(e)))
+            }
+        )
+      },
+      test("transports errors only once") {
+        makeSubscriber.flatMap(probe =>
+          probe.underlying
+            .toSink[Throwable]
+            .use { case (signalError, sink) =>
+              for {
+                _    <- ZStream.fail(e).run(sink).catchAll(signalError)
+                err  <- probe.expectError.exit
+                _    <- signalError(e)
+                err2 <- probe.expectError.timeout(100.millis).exit
+              } yield assert(err)(succeeds(equalTo(e))) && assert(err2)(fails(anything))
+            }
+        )
       }
     )