class 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()) {
                        // Suppress messages if future has already failed, or completed for other reason:
                        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);

}