Fix replyStream interruption (#111)

devsisters · Mar 6, 2024 · 8333e6a · 8333e6a
1 parent 7390aa1
commit 8333e6a
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/entities/src/main/scala/com/devsisters/shardcake/internal/ReplyChannel.scala b/entities/src/main/scala/com/devsisters/shardcake/internal/ReplyChannel.scala
@@ -20,7 +20,8 @@ private[shardcake] object ReplyChannel {
     def replyStream(stream: ZStream[Any, Throwable, A]): UIO[Unit] =
       (stream
         .runForeachChunk(chunk => queue.offer(Take.chunk(chunk)))
-        .foldCauseZIO(cause => queue.offer(Take.failCause(cause)), _ => queue.offer(Take.end)) race await).fork.unit
+        .onExit(e => queue.offer(e.foldExit(Take.failCause, _ => Take.end)))
+        .ignore race await).fork.unit
     val output: ZStream[Any, Throwable, A]                         = ZStream.fromQueueWithShutdown(queue).flattenTake.onError(fail)
   }