protected void unbufferedWrite(final ChannelHandlerContext ctx, final Outgoing out) {

        if (out.message instanceof FullHttpMessage) {
            // Forward as is
            ctx.writeAndFlush(out.message, out.promise);
            out.promise.addListener((ChannelFutureListener) channelFuture -> executeInEventLoop(ctx, () -> {
                sentOutMessage(ctx);
                outgoing.remove();
                flushNext(ctx);
            }));

        } else if (out.message instanceof StreamedHttpMessage) {

            StreamedHttpMessage streamed = (StreamedHttpMessage) out.message;
            HandlerSubscriber<HttpContent> subscriber = new HandlerSubscriber<HttpContent>(ctx.executor()) {
                @Override
                protected void error(Throwable error) {
                    try {

                        if (LOG.isErrorEnabled()) {
                            LOG.error("Error occurred writing stream response: " + error.getMessage(), error);
                        }
                        ctx.writeAndFlush(new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR))
                           .addListener(ChannelFutureListener.CLOSE);
                    } finally {
                        ctx.read();
                    }
                }

                @Override
                protected void complete() {
                    executeInEventLoop(ctx, () -> completeBody(ctx));
                }
            };

            sendLastHttpContent = true;

            // DON'T pass the promise through, create a new promise instead.
            ctx.writeAndFlush(out.message);

            ctx.pipeline().addAfter(ctx.name(), ctx.name() + "-body-subscriber", subscriber);
            subscribeSubscriberToStream(streamed, subscriber);
        }

    }
protected void unbufferedWrite ( final ChannelHandlerContext ctx , final Outgoing out ) { if ( out . message instanceof FullHttpMessage ) { ctx . writeAndFlush ( out . message , out . promise ) ; out . promise . addListener ( ( ChannelFutureListener ) channelFuture -> executeInEventLoop ( ctx , ( ) -> { sentOutMessage ( ctx ) ; outgoing . remove ( ) ; flushNext ( ctx ) ; } ) ) ; } else if ( out . message instanceof StreamedHttpMessage ) { StreamedHttpMessage streamed = ( StreamedHttpMessage ) out . message ; HandlerSubscriber < HttpContent > subscriber = new HandlerSubscriber < HttpContent > ( ctx . executor ( ) ) { @ Override protected void error ( Throwable error ) { try { if ( LOG . isErrorEnabled ( ) ) { LOG . error ( "Error occurred writing stream response: " + error . getMessage ( ) , error ) ; } ctx . writeAndFlush ( new DefaultHttpResponse ( HttpVersion . HTTP_1_1 , HttpResponseStatus . INTERNAL_SERVER_ERROR ) ) . addListener ( ChannelFutureListener . CLOSE ) ; } finally { ctx . read ( ) ; } } @ Override protected void complete ( ) { executeInEventLoop ( ctx , ( ) -> completeBody ( ctx ) ) ; } } ; sendLastHttpContent = true ; ctx . writeAndFlush ( out . message ) ; ctx . pipeline ( ) . addAfter ( ctx . name ( ) , ctx . name ( ) + "-body-subscriber" , subscriber ) ; subscribeSubscriberToStream ( streamed , subscriber ) ; } }
@param ctx The channel handler context
@param out The output stream