onDataRead() — netty Function Reference

Source Code

codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionDecoder.java lines 251–333

        @Override
        public int onDataRead(final ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
                              boolean endOfStream) throws Http2Exception {
            Http2Stream stream = connection.stream(streamId);
            Http2LocalFlowController flowController = flowController();
            int readable = data.readableBytes();
            int bytesToReturn = readable + padding;

            final boolean shouldIgnore;
            try {
                shouldIgnore = shouldIgnoreHeadersOrDataFrame(ctx, streamId, stream, endOfStream, "DATA");
            } catch (Http2Exception e) {
                // Ignoring this frame. We still need to count the frame towards the connection flow control
                // window, but we immediately mark all bytes as consumed.
                flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
                flowController.consumeBytes(stream, bytesToReturn);
                throw e;
            } catch (Throwable t) {
                throw connectionError(INTERNAL_ERROR, t, "Unhandled error on data stream id %d", streamId);
            }

            if (shouldIgnore) {
                // Ignoring this frame. We still need to count the frame towards the connection flow control
                // window, but we immediately mark all bytes as consumed.
                flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
                flowController.consumeBytes(stream, bytesToReturn);

                // Verify that the stream may have existed after we apply flow control.
                verifyStreamMayHaveExisted(streamId, endOfStream, "DATA");

                // All bytes have been consumed.
                return bytesToReturn;
            }
            Http2Exception error = null;
            switch (stream.state()) {
                case OPEN:
                case HALF_CLOSED_LOCAL:
                    break;
                case HALF_CLOSED_REMOTE:
                case CLOSED:
                    error = streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
                        stream.id(), stream.state());
                    break;
                default:
                    error = streamError(stream.id(), PROTOCOL_ERROR,
                        "Stream %d in unexpected state: %s", stream.id(), stream.state());
                    break;
            }

            int unconsumedBytes = unconsumedBytes(stream);
            try {
                flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
                // Update the unconsumed bytes after flow control is applied.
                unconsumedBytes = unconsumedBytes(stream);

                // If the stream is in an invalid state to receive the frame, throw the error.
                if (error != null) {
                    throw error;
                }

                verifyContentLength(stream, readable, endOfStream);

                // Call back the application and retrieve the number of bytes that have been
                // immediately processed.
                bytesToReturn = listener.onDataRead(ctx, streamId, data, padding, endOfStream);

                if (endOfStream) {
                    lifecycleManager.closeStreamRemote(stream, ctx.newSucceededFuture());
                }

                return bytesToReturn;
            } catch (Http2Exception | RuntimeException e) {
                // If an exception happened during delivery, the listener may have returned part
                // of the bytes before the error occurred. If that's the case, subtract that from
                // the total processed bytes so that we don't return too many bytes.
                int delta = unconsumedBytes - unconsumedBytes(stream);
                bytesToReturn -= delta;
                throw e;
            } finally {
                // If appropriate, return the processed bytes to the flow controller.
                flowController.consumeBytes(stream, bytesToReturn);