diff --git a/grpc-zio/src/main/scala/proteus/client/ZioClientBackend.scala b/grpc-zio/src/main/scala/proteus/client/ZioClientBackend.scala index 331a7fe..4935873 100644 --- a/grpc-zio/src/main/scala/proteus/client/ZioClientBackend.scala +++ b/grpc-zio/src/main/scala/proteus/client/ZioClientBackend.scala @@ -158,8 +158,9 @@ class ZioClientBackend(channel: Channel, runtime: Runtime[Any], prefetchN: Int) // Sender surfaces failures via the response promise so promise.await can return them // even if the server hasn't closed; on success the server completes the promise via onClose. - val send: UIO[Unit] = sendAll.catchAll { e => - ZIO.succeed(call.cancel("Error sending requests", e)) <* listener.promise.fail(e) + val send: IO[StatusException, Unit] = sendAll.catchAllCause { cause => + if (cause.isInterruptedOnly) ZIO.refailCause(cause) + else ZIO.succeed(call.cancel("Error sending requests", cause.squash)) <* listener.promise.failCause(cause) } ZIO diff --git a/grpc-zio/src/test/scala/proteus/ZioBackendSpec.scala b/grpc-zio/src/test/scala/proteus/ZioBackendSpec.scala index 3972767..80cc537 100644 --- a/grpc-zio/src/test/scala/proteus/ZioBackendSpec.scala +++ b/grpc-zio/src/test/scala/proteus/ZioBackendSpec.scala @@ -288,6 +288,23 @@ object ZioBackendSpec extends ZIOSpecDefault { channel.shutdown().awaitTermination(5, TimeUnit.SECONDS) }.ignore) }, + test("client streaming should not hang when the request stream dies with a defect") { + val port = 7019 + val server = NettyServerBuilder.forPort(port).addService(streamingServerService).build().start() + val channel = NettyChannelBuilder.forAddress("localhost", port).usePlaintext().build() + val clientBackend = ZioClientBackend(channel) + + val client = clientBackend.client(clientStreamingRpc, streamingService) + val requestStream = ZStream(StreamRequest(1)).map[StreamRequest](_ => throw new RuntimeException("stream boom")) + + client(requestStream).exit + .timeout(Duration.fromSeconds(10)) + .flatMap(result => assertTrue(result.exists(_.isFailure))) + .ensuring(ZIO.attempt { + server.shutdown().awaitTermination(5, TimeUnit.SECONDS) + channel.shutdown().awaitTermination(5, TimeUnit.SECONDS) + }.ignore) + } @@ TestAspect.withLiveClock, test("bidi streaming should fail with StatusException, not die, when call.start throws") { val port = 7018 val server = NettyServerBuilder.forPort(port).addService(streamingServerService).build().start()