class NettyToStyxRequestDecoder extends MessageToMessageDecoder<HttpObject> {

    @Override
    protected void decode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out) throws Exception {
        if (msg.getDecoderResult().isFailure()) {
            String formattedHttpObject = httpMessageFormatter.formatNettyMessage(msg);
            throw new BadRequestException("Error while decoding request: " + formattedHttpObject,
                    httpMessageFormatter.wrap(msg.getDecoderResult().cause()));
        }

        try {
            if (msg instanceof HttpRequest) {
                ctx.channel().config().setAutoRead(false);
                ctx.channel().read();

                HttpRequest nettyRequest = (HttpRequest) msg;

                this.producer = Optional.of(createProducer(ctx, nettyRequest.uri()));
                Publisher<Buffer> contentPublisher = new FlowControllingPublisher(queueDrainingExecutor, this.producer.get());

                LiveHttpRequest styxRequest = toStyxRequest(nettyRequest, contentPublisher);
                out.add(styxRequest);

            }
            if (msg instanceof HttpContent) {
                assert this.producer.isPresent();

                ByteBuf content = ((ByteBufHolder) msg).content();
                if (content.isReadable()) {
                    ByteBuf byteBuf = retain(content);
                    queueDrainingExecutor.execute(() -> producer.get().newChunk(byteBuf));
                }
                if (msg instanceof LastHttpContent) {
                    queueDrainingExecutor.execute(() -> producer.get().lastHttpContent());
                }
            }
        } catch (BadRequestException ex) {
            throw ex;
        } catch (Exception ex) {
            throw new BadRequestException(ex.getMessage() + " in " + msg, ex);
        }
    }

    private  NettyToStyxRequestDecoder(Builder builder);

}