@Override
        public void onNext(I obj) {
            requireNonNull(obj, "obj");

            askedUpstreamForElement = false;
            handlerProduced = false;
            try {
                // Call the handler so that it publishes something.
                if (isHttpDecoder && obj instanceof HttpHeaders) {
                    // HttpDecoder has dedicated APIs for headers and trailers.
                    final HttpDecoder<O> httpDecoder = (HttpDecoder<O>) decoder;
                    final HttpHeaders headers = (HttpHeaders) obj;
                    if (headers instanceof ResponseHeaders &&
                        ((ResponseHeaders) headers).status().isInformational()) {
                        httpDecoder.processInformationalHeaders((ResponseHeaders) headers,
                                                                DecodedStreamMessage.this);
                    } else if (!sawLeadingHeaders) {
                        sawLeadingHeaders = true;
                        httpDecoder.processHeaders((HttpHeaders) obj, DecodedStreamMessage.this);
                    } else {
                        httpDecoder.processTrailers((HttpHeaders) obj, DecodedStreamMessage.this);
                    }
                } else {
                    final ByteBuf byteBuf = decoder.toByteBuf(obj);
                    requireNonNull(byteBuf, "decoder.toByteBuf() returned null");
                    if (input.add(byteBuf)) {
                        decoder.process(input, DecodedStreamMessage.this);
                    }
                }

                if (handlerProduced) {
                    // Handler produced something.
                    if (!askedUpstreamForElement) {
                        // Ask the upstream for more elements after the produced elements are fully consumed and
                        // there are still demands left.
                        whenConsumed().handle((unused1, unused2) -> {
                            if (demand() > 0) {
                                askUpstreamForElement();
                            }
                            return null;
                        });
                    } else {
                        // No need to ask the upstream for more elements because:
                        // - The handler triggered Subscription.request(); or
                        // - Subscription.request() was called from another thread.
                    }
                } else {
                    // Handler didn't produce anything, which means it needs more elements from the upstream
                    // to produce something if there is a demand.
                    if (demand() > 0) {
                        askUpstreamForElement();
                    }
                }
            } catch (Throwable ex) {
                decoder.processOnError(ex);
                cancelAndCleanup();
                abort(ex);
                Exceptions.throwIfFatal(ex);
            }
        }
Returns a new {@link DecodedStreamMessage} with the specified {@link HttpDecoder},
{@link ByteBufAllocator} and {@code byteBufConverter}.
core/src/main/java/com/linecorp/armeria/internal/common/stream/DecodedStreamMessage.java