public final void onInput()

in httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/AbstractHttp1StreamDuplexer.java [268:358]
79 lines of code
23 McCabe index (conditional complexity)

    public final void onInput(final ByteBuffer src) throws HttpException, IOException {
        if (src != null) {
            final int n = src.remaining();
            inbuf.put(src);
            inTransportMetrics.incrementBytesTransferred(n);
        }

        if (connState.compareTo(ConnectionState.GRACEFUL_SHUTDOWN) >= 0 && !inbuf.hasData() && inputIdle()) {
            ioSession.clearEvent(SelectionKey.OP_READ);
            return;
        }

        boolean endOfStream = false;
        if (incomingMessage == null) {
            final int bytesRead = inbuf.fill(ioSession);
            if (bytesRead > 0) {
                inTransportMetrics.incrementBytesTransferred(bytesRead);
            }
            endOfStream = bytesRead == -1;
        }

        do {
            if (incomingMessage == null) {

                final IncomingMessage messageHead = parseMessageHead(endOfStream);
                if (messageHead != null) {
                    this.version = messageHead.getVersion();

                    updateInputMetrics(messageHead, connMetrics);
                    final ContentDecoder contentDecoder;
                    if (handleIncomingMessage(messageHead)) {
                        final long len = incomingContentStrategy.determineLength(messageHead);
                        contentDecoder = createContentDecoder(len, ioSession, inbuf, inTransportMetrics);
                        consumeHeader(messageHead, contentDecoder != null ? new IncomingEntityDetails(messageHead, len) : null);
                    } else {
                        consumeHeader(messageHead, null);
                        contentDecoder = null;
                    }
                    capacityWindow = new CapacityWindow(http1Config.getInitialWindowSize(), ioSession);
                    if (contentDecoder != null) {
                        incomingMessage = new Message<>(messageHead, contentDecoder);
                    } else {
                        inputEnd();
                        if (connState.compareTo(ConnectionState.ACTIVE) == 0) {
                            ioSession.setEvent(SelectionKey.OP_READ);
                        }
                    }
                } else {
                    break;
                }
            }

            if (incomingMessage != null) {
                final ContentDecoder contentDecoder = incomingMessage.getBody();

                // At present the consumer can be forced to consume data
                // over its declared capacity in order to avoid having
                // unprocessed message body content stuck in the session
                // input buffer
                final int bytesRead = contentDecoder.read(contentBuffer);
                if (bytesRead > 0) {
                    contentBuffer.flip();
                    consumeData(contentBuffer);
                    contentBuffer.clear();
                    final int capacity = capacityWindow.removeCapacity(bytesRead);
                    if (capacity <= 0) {
                        if (!contentDecoder.isCompleted()) {
                            updateCapacity(capacityWindow);
                        }
                    }
                }
                if (contentDecoder.isCompleted()) {
                    dataEnd(contentDecoder.getTrailers());
                    capacityWindow.close();
                    incomingMessage = null;
                    ioSession.setEvent(SelectionKey.OP_READ);
                    inputEnd();
                } else if (bytesRead == 0) {
                    break;
                }
            }
        } while (inbuf.hasData());

        if (endOfStream && !inbuf.hasData()) {
            if (inputIdle()) {
                requestShutdown(CloseMode.GRACEFUL);
            } else {
                shutdownSession(new ConnectionClosedException("Connection closed by peer"));
            }
        }
    }