@Test public void unsubscribesFromContentWhenCancelled() throws Exception { CaptureHttpResponseWriteEventsHandler writeEventsCollector = new CaptureHttpResponseWriteEventsHandler(); AtomicBoolean unsubscribed = new AtomicBoolean(false); EmbeddedChannel ch = new EmbeddedChannel( new CaptureChannelArgumentsHandler(channelArgs), writeEventsCollector, new SimpleChannelInboundHandler<LiveHttpResponse>() { @Override protected void channelRead0(ChannelHandlerContext ctx, LiveHttpResponse response) throws Exception { HttpResponseWriter writer = new HttpResponseWriter(ctx); CompletableFuture<Void> future = writer.write(response); writeAck(channelArgs); assertThat(future.isDone(), is(false)); future.cancel(false); assertThat(unsubscribed.get(), is(true)); assertThat(future.isDone(), is(true)); channelRead.set(true); } } ); ch.writeInbound(response(OK).body(new ByteStream(contentObservable.doOnCancel(() -> unsubscribed.set(true)))).build()); assertThat(channelRead.get(), is(true)); }
public CompletableFuture<Void> write(LiveHttpResponse response) { CompletableFuture<Void> future = new CompletableFuture<>(); try { writeHeaders(response).addListener((ChannelFutureListener) writeOp -> { if (writeOp.isSuccess()) { writeOpsAcked.incrementAndGet(); } else { LOGGER.warn("Unable to send response headers. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } }); response.body().subscribe(new BaseSubscriber<Buffer>() { @Override public void hookOnSubscribe(Subscription subscription) { future.handle((ignore, cause) -> { if (future.isCompletedExceptionally() && cause instanceof CancellationException) { subscription.cancel(); } return null; }); subscription.request(1); } @Override public void hookOnComplete() { if (!future.isDone()) { nettyWriteAndFlush(EMPTY_LAST_CONTENT).addListener((ChannelFutureListener) this::onWriteEmptyLastChunkOutcome); contentCompleted.set(true); completeIfAllSent(future); } } @Override public void hookOnError(Throwable cause) { LOGGER.warn("Content observable error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), cause }); future.completeExceptionally(cause); } @Override public void hookOnNext(Buffer buffer) { ByteBuf byteBuf = Buffers.toByteBuf(buffer); if (future.isDone()) { byteBuf.release(); } else { long bufSize = (long) byteBuf.readableBytes(); contentBytesWritten.addAndGet(bufSize); nettyWriteAndFlush(new DefaultHttpContent(byteBuf)) .addListener(it -> onWriteOutcome((ChannelFuture) it, bufSize)); } } private void onWriteOutcome(ChannelFuture writeOp, long bufSize) { if (writeOp.isSuccess()) { contentBytesAcked.addAndGet(bufSize); writeOpsAcked.incrementAndGet(); request(1); completeIfAllSent(future); } else if (!future.isDone()) { cancel(); LOGGER.warn("Write error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes), Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), response, writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } } private void onWriteEmptyLastChunkOutcome(ChannelFuture writeOp) { writeOpsAcked.incrementAndGet(); completeIfAllSent(future); cancel(); } }); return future; } catch (Throwable cause) { LOGGER.warn("Failed to convert response headers. response={}, Cause={}", new Object[]{response, cause}); Flux.from(response.body().drop()).subscribe(); future.completeExceptionally(cause); return future; } }
HttpResponseWriter { public CompletableFuture<Void> write(LiveHttpResponse response) { CompletableFuture<Void> future = new CompletableFuture<>(); try { writeHeaders(response).addListener((ChannelFutureListener) writeOp -> { if (writeOp.isSuccess()) { writeOpsAcked.incrementAndGet(); } else { LOGGER.warn("Unable to send response headers. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } }); response.body().subscribe(new BaseSubscriber<Buffer>() { @Override public void hookOnSubscribe(Subscription subscription) { future.handle((ignore, cause) -> { if (future.isCompletedExceptionally() && cause instanceof CancellationException) { subscription.cancel(); } return null; }); subscription.request(1); } @Override public void hookOnComplete() { if (!future.isDone()) { nettyWriteAndFlush(EMPTY_LAST_CONTENT).addListener((ChannelFutureListener) this::onWriteEmptyLastChunkOutcome); contentCompleted.set(true); completeIfAllSent(future); } } @Override public void hookOnError(Throwable cause) { LOGGER.warn("Content observable error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), cause }); future.completeExceptionally(cause); } @Override public void hookOnNext(Buffer buffer) { ByteBuf byteBuf = Buffers.toByteBuf(buffer); if (future.isDone()) { byteBuf.release(); } else { long bufSize = (long) byteBuf.readableBytes(); contentBytesWritten.addAndGet(bufSize); nettyWriteAndFlush(new DefaultHttpContent(byteBuf)) .addListener(it -> onWriteOutcome((ChannelFuture) it, bufSize)); } } private void onWriteOutcome(ChannelFuture writeOp, long bufSize) { if (writeOp.isSuccess()) { contentBytesAcked.addAndGet(bufSize); writeOpsAcked.incrementAndGet(); request(1); completeIfAllSent(future); } else if (!future.isDone()) { cancel(); LOGGER.warn("Write error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes), Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), response, writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } } private void onWriteEmptyLastChunkOutcome(ChannelFuture writeOp) { writeOpsAcked.incrementAndGet(); completeIfAllSent(future); cancel(); } }); return future; } catch (Throwable cause) { LOGGER.warn("Failed to convert response headers. response={}, Cause={}", new Object[]{response, cause}); Flux.from(response.body().drop()).subscribe(); future.completeExceptionally(cause); return future; } } }
HttpResponseWriter { public CompletableFuture<Void> write(LiveHttpResponse response) { CompletableFuture<Void> future = new CompletableFuture<>(); try { writeHeaders(response).addListener((ChannelFutureListener) writeOp -> { if (writeOp.isSuccess()) { writeOpsAcked.incrementAndGet(); } else { LOGGER.warn("Unable to send response headers. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } }); response.body().subscribe(new BaseSubscriber<Buffer>() { @Override public void hookOnSubscribe(Subscription subscription) { future.handle((ignore, cause) -> { if (future.isCompletedExceptionally() && cause instanceof CancellationException) { subscription.cancel(); } return null; }); subscription.request(1); } @Override public void hookOnComplete() { if (!future.isDone()) { nettyWriteAndFlush(EMPTY_LAST_CONTENT).addListener((ChannelFutureListener) this::onWriteEmptyLastChunkOutcome); contentCompleted.set(true); completeIfAllSent(future); } } @Override public void hookOnError(Throwable cause) { LOGGER.warn("Content observable error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), cause }); future.completeExceptionally(cause); } @Override public void hookOnNext(Buffer buffer) { ByteBuf byteBuf = Buffers.toByteBuf(buffer); if (future.isDone()) { byteBuf.release(); } else { long bufSize = (long) byteBuf.readableBytes(); contentBytesWritten.addAndGet(bufSize); nettyWriteAndFlush(new DefaultHttpContent(byteBuf)) .addListener(it -> onWriteOutcome((ChannelFuture) it, bufSize)); } } private void onWriteOutcome(ChannelFuture writeOp, long bufSize) { if (writeOp.isSuccess()) { contentBytesAcked.addAndGet(bufSize); writeOpsAcked.incrementAndGet(); request(1); completeIfAllSent(future); } else if (!future.isDone()) { cancel(); LOGGER.warn("Write error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes), Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), response, writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } } private void onWriteEmptyLastChunkOutcome(ChannelFuture writeOp) { writeOpsAcked.incrementAndGet(); completeIfAllSent(future); cancel(); } }); return future; } catch (Throwable cause) { LOGGER.warn("Failed to convert response headers. response={}, Cause={}", new Object[]{response, cause}); Flux.from(response.body().drop()).subscribe(); future.completeExceptionally(cause); return future; } } HttpResponseWriter(ChannelHandlerContext ctx);   HttpResponseWriter(ChannelHandlerContext ctx, ResponseTranslator responseTranslator); }
HttpResponseWriter { public CompletableFuture<Void> write(LiveHttpResponse response) { CompletableFuture<Void> future = new CompletableFuture<>(); try { writeHeaders(response).addListener((ChannelFutureListener) writeOp -> { if (writeOp.isSuccess()) { writeOpsAcked.incrementAndGet(); } else { LOGGER.warn("Unable to send response headers. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } }); response.body().subscribe(new BaseSubscriber<Buffer>() { @Override public void hookOnSubscribe(Subscription subscription) { future.handle((ignore, cause) -> { if (future.isCompletedExceptionally() && cause instanceof CancellationException) { subscription.cancel(); } return null; }); subscription.request(1); } @Override public void hookOnComplete() { if (!future.isDone()) { nettyWriteAndFlush(EMPTY_LAST_CONTENT).addListener((ChannelFutureListener) this::onWriteEmptyLastChunkOutcome); contentCompleted.set(true); completeIfAllSent(future); } } @Override public void hookOnError(Throwable cause) { LOGGER.warn("Content observable error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), cause }); future.completeExceptionally(cause); } @Override public void hookOnNext(Buffer buffer) { ByteBuf byteBuf = Buffers.toByteBuf(buffer); if (future.isDone()) { byteBuf.release(); } else { long bufSize = (long) byteBuf.readableBytes(); contentBytesWritten.addAndGet(bufSize); nettyWriteAndFlush(new DefaultHttpContent(byteBuf)) .addListener(it -> onWriteOutcome((ChannelFuture) it, bufSize)); } } private void onWriteOutcome(ChannelFuture writeOp, long bufSize) { if (writeOp.isSuccess()) { contentBytesAcked.addAndGet(bufSize); writeOpsAcked.incrementAndGet(); request(1); completeIfAllSent(future); } else if (!future.isDone()) { cancel(); LOGGER.warn("Write error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes), Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), response, writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } } private void onWriteEmptyLastChunkOutcome(ChannelFuture writeOp) { writeOpsAcked.incrementAndGet(); completeIfAllSent(future); cancel(); } }); return future; } catch (Throwable cause) { LOGGER.warn("Failed to convert response headers. response={}, Cause={}", new Object[]{response, cause}); Flux.from(response.body().drop()).subscribe(); future.completeExceptionally(cause); return future; } } HttpResponseWriter(ChannelHandlerContext ctx);   HttpResponseWriter(ChannelHandlerContext ctx, ResponseTranslator responseTranslator); CompletableFuture<Void> write(LiveHttpResponse response); }
HttpResponseWriter { public CompletableFuture<Void> write(LiveHttpResponse response) { CompletableFuture<Void> future = new CompletableFuture<>(); try { writeHeaders(response).addListener((ChannelFutureListener) writeOp -> { if (writeOp.isSuccess()) { writeOpsAcked.incrementAndGet(); } else { LOGGER.warn("Unable to send response headers. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } }); response.body().subscribe(new BaseSubscriber<Buffer>() { @Override public void hookOnSubscribe(Subscription subscription) { future.handle((ignore, cause) -> { if (future.isCompletedExceptionally() && cause instanceof CancellationException) { subscription.cancel(); } return null; }); subscription.request(1); } @Override public void hookOnComplete() { if (!future.isDone()) { nettyWriteAndFlush(EMPTY_LAST_CONTENT).addListener((ChannelFutureListener) this::onWriteEmptyLastChunkOutcome); contentCompleted.set(true); completeIfAllSent(future); } } @Override public void hookOnError(Throwable cause) { LOGGER.warn("Content observable error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes). Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), cause }); future.completeExceptionally(cause); } @Override public void hookOnNext(Buffer buffer) { ByteBuf byteBuf = Buffers.toByteBuf(buffer); if (future.isDone()) { byteBuf.release(); } else { long bufSize = (long) byteBuf.readableBytes(); contentBytesWritten.addAndGet(bufSize); nettyWriteAndFlush(new DefaultHttpContent(byteBuf)) .addListener(it -> onWriteOutcome((ChannelFuture) it, bufSize)); } } private void onWriteOutcome(ChannelFuture writeOp, long bufSize) { if (writeOp.isSuccess()) { contentBytesAcked.addAndGet(bufSize); writeOpsAcked.incrementAndGet(); request(1); completeIfAllSent(future); } else if (!future.isDone()) { cancel(); LOGGER.warn("Write error. Written content bytes {}/{} (ackd/sent). Write events {}/{} (ackd/writes), Exception={}", new Object[]{ contentBytesAcked.get(), contentBytesWritten.get(), writeOpsAcked.get(), writeOps.get(), response, writeOp.cause()}); future.completeExceptionally(writeOp.cause()); } } private void onWriteEmptyLastChunkOutcome(ChannelFuture writeOp) { writeOpsAcked.incrementAndGet(); completeIfAllSent(future); cancel(); } }); return future; } catch (Throwable cause) { LOGGER.warn("Failed to convert response headers. response={}, Cause={}", new Object[]{response, cause}); Flux.from(response.body().drop()).subscribe(); future.completeExceptionally(cause); return future; } } HttpResponseWriter(ChannelHandlerContext ctx);   HttpResponseWriter(ChannelHandlerContext ctx, ResponseTranslator responseTranslator); CompletableFuture<Void> write(LiveHttpResponse response);  }