channelRead() — netty Function Reference

Source Code

codec-base/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java lines 285–341

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object input) throws Exception {
        if (decodeState == STATE_INIT) {
            do {
                if (input instanceof ByteBuf) {
                    selfFiredChannelRead = true;
                    CodecOutputList out = CodecOutputList.newInstance();
                    try {
                        first = cumulation == null;
                        cumulation = cumulator.cumulate(ctx.alloc(),
                                first ? EMPTY_BUFFER : cumulation, (ByteBuf) input);
                        callDecode(ctx, cumulation, out);
                    } catch (DecoderException e) {
                        throw e;
                    } catch (Exception e) {
                        throw new DecoderException(e);
                    } finally {
                        try {
                            if (cumulation != null && !cumulation.isReadable()) {
                                numReads = 0;
                                try {
                                    cumulation.release();
                                } catch (IllegalReferenceCountException e) {
                                    //noinspection ThrowFromFinallyBlock
                                    throw new IllegalReferenceCountException(
                                            getClass().getSimpleName() +
                                                    "#decode() might have released its input buffer, " +
                                                    "or passed it down the pipeline without a retain() call, " +
                                                    "which is not allowed.", e);
                                }
                                cumulation = null;
                            } else if (++numReads >= discardAfterReads) {
                                // We did enough reads already try to discard some bytes, so we not risk to see a OOME.
                                // See https://github.com/netty/netty/issues/4275
                                numReads = 0;
                                discardSomeReadBytes();
                            }

                            int size = out.size();
                            firedChannelRead |= out.insertSinceRecycled();
                            fireChannelRead(ctx, out, size);
                        } finally {
                            out.recycle();
                        }
                    }
                } else {
                    ctx.fireChannelRead(input);
                }
            } while (inputMessages != null && (input = inputMessages.poll()) != null);
        } else {
            // Reentrant call. Bail out here and let original call process our message.
            if (inputMessages == null) {
                inputMessages = new ArrayDeque<>(2);
            }
            inputMessages.offer(input);
        }
    }