io.netty.channel.ChannelConfig#getMaxMessagesPerRead ( )源码实例Demo

@Override
protected void doRead() {
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    boolean closed = false;
    final int maxMessagesPerRead = config.getMaxMessagesPerRead();

    Throwable exception = null;
    int localRead = 0;
    int totalRead = 0;
    try {
        for (;;) {
            // Perform a read.
            localRead = doReadMessages(readBuf);
            if (localRead == 0) {
                break;
            }
            if (localRead < 0) {
                closed = true;
                break;
            }

            // Notify with the received messages and clear the buffer.
            int size = readBuf.size();
            for (int i = 0; i < size; i ++) {
                pipeline.fireChannelRead(readBuf.get(i));
            }
            readBuf.clear();

            // Do not read beyond maxMessagesPerRead.
            // Do not continue reading if autoRead has been turned off.
            totalRead += localRead;
            if (totalRead >= maxMessagesPerRead || !config.isAutoRead()) {
                break;
            }
        }
    } catch (Throwable t) {
        exception = t;
    }

    pipeline.fireChannelReadComplete();

    if (exception != null) {
        if (exception instanceof IOException) {
            closed = true;
        }

        pipeline().fireExceptionCaught(exception);
    }

    if (closed) {
        if (isOpen()) {
            unsafe().close(unsafe().voidPromise());
        }
    } else if (localRead == 0 && isActive()) {
        // If the read amount was 0 and the channel is still active we need to trigger a new read()
        // as otherwise we will never try to read again and the user will never know.
        // Just call read() is ok here as it will be submitted to the EventLoop as a task and so we are
        // able to process the rest of the tasks in the queue first.
        //
        // See https://github.com/netty/netty/issues/2404
        read();
    }
}
 

@Override
public final void read() {
    final ChannelConfig config = config();
    if (!config.isAutoRead() && !isReadPending()) {
        // ChannelConfig.setAutoRead(false) was called in the meantime
        removeReadOp();
        return;
    }

    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
    final int maxMessagesPerRead = config.getMaxMessagesPerRead();
    RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
    if (allocHandle == null) {
        this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
    }

    ByteBuf byteBuf = null;
    int messages = 0;
    boolean close = false;
    try {
        int totalReadAmount = 0;
        boolean readPendingReset = false;
        do {
            byteBuf = allocHandle.allocate(allocator);
            int writable = byteBuf.writableBytes();
            int localReadAmount = doReadBytes(byteBuf);
            if (localReadAmount <= 0) {
                // not was read release the buffer
                byteBuf.release();
                byteBuf = null;
                close = localReadAmount < 0;
                break;
            }
            if (!readPendingReset) {
                readPendingReset = true;
                setReadPending(false);
            }
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;

            if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
                // Avoid overflow.
                totalReadAmount = Integer.MAX_VALUE;
                break;
            }

            totalReadAmount += localReadAmount;

            // stop reading
            if (!config.isAutoRead()) {
                break;
            }

            if (localReadAmount < writable) {
                // Read less than what the buffer can hold,
                // which might mean we drained the recv buffer completely.
                break;
            }
        } while (++ messages < maxMessagesPerRead);

        pipeline.fireChannelReadComplete();
        allocHandle.record(totalReadAmount);

        if (close) {
            closeOnRead(pipeline);
            close = false;
        }
    } catch (Throwable t) {
        handleReadException(pipeline, byteBuf, t, close);
    } finally {
        // Check if there is a readPending which was not processed yet.
        // This could be for two reasons:
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
        //
        // See https://github.com/netty/netty/issues/2254
        if (!config.isAutoRead() && !isReadPending()) {
            removeReadOp();
        }
    }
}
 

@Override
void epollInReady() {
    final ChannelConfig config = config();
    boolean edgeTriggered = isFlagSet(Native.EPOLLET);

    if (!readPending && !edgeTriggered && !config.isAutoRead()) {
        // ChannelConfig.setAutoRead(false) was called in the meantime
        clearEpollIn0();
        return;
    }

    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
    RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
    if (allocHandle == null) {
        this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
    }

    ByteBuf byteBuf = null;
    boolean close = false;
    try {
        // if edgeTriggered is used we need to read all messages as we are not notified again otherwise.
        final int maxMessagesPerRead = edgeTriggered
                ? Integer.MAX_VALUE : config.getMaxMessagesPerRead();
        int messages = 0;
        int totalReadAmount = 0;
        do {
            // we use a direct buffer here as the native implementations only be able
            // to handle direct buffers.
            byteBuf = allocHandle.allocate(allocator);
            int writable = byteBuf.writableBytes();
            int localReadAmount = doReadBytes(byteBuf);
            if (localReadAmount <= 0) {
                // not was read release the buffer
                byteBuf.release();
                close = localReadAmount < 0;
                break;
            }
            readPending = false;
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;

            if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
                allocHandle.record(totalReadAmount);

                // Avoid overflow.
                totalReadAmount = localReadAmount;
            } else {
                totalReadAmount += localReadAmount;
            }

            if (localReadAmount < writable) {
                // Read less than what the buffer can hold,
                // which might mean we drained the recv buffer completely.
                break;
            }
            if (!edgeTriggered && !config.isAutoRead()) {
                // This is not using EPOLLET so we can stop reading
                // ASAP as we will get notified again later with
                // pending data
                break;
            }
        } while (++ messages < maxMessagesPerRead);

        pipeline.fireChannelReadComplete();
        allocHandle.record(totalReadAmount);

        if (close) {
            closeOnRead(pipeline);
            close = false;
        }
    } catch (Throwable t) {
        boolean closed = handleReadException(pipeline, byteBuf, t, close);
        if (!closed) {
            // trigger a read again as there may be something left to read and because of epoll ET we
            // will not get notified again until we read everything from the socket
            eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    epollInReady();
                }
            });
        }
    } finally {
        // Check if there is a readPending which was not processed yet.
        // This could be for two reasons:
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
        //
        // See https://github.com/netty/netty/issues/2254
        if (!readPending && !config.isAutoRead()) {
            clearEpollIn0();
        }
    }
}
 

@Override
public void handleEvent(ConduitStreamSourceChannel channel) {
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
    final int maxMessagesPerRead = config.getMaxMessagesPerRead();
    RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
    if (allocHandle == null) {
        this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
    }

    ByteBuf byteBuf = null;
    int messages = 0;
    boolean close = false;
    try {
        int byteBufCapacity = allocHandle.guess();
        int totalReadAmount = 0;
        do {
            byteBuf = allocator.ioBuffer(byteBufCapacity);
            int writable = byteBuf.writableBytes();
            int localReadAmount = byteBuf.writeBytes(channel, byteBuf.writableBytes());
            if (localReadAmount <= 0) {
                // not was read release the buffer
                byteBuf.release();
                close = localReadAmount < 0;
                break;
            }
            ((AbstractXnioUnsafe) unsafe()).readPending = false;
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;

            if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
                // Avoid overflow.
                totalReadAmount = Integer.MAX_VALUE;
                break;
            }

            totalReadAmount += localReadAmount;

            // stop reading
            if (!config.isAutoRead()) {
                break;
            }

            if (localReadAmount < writable) {
                // Read less than what the buffer can hold,
                // which might mean we drained the recv buffer completely.
                break;
            }
        } while (++ messages < maxMessagesPerRead);

        pipeline.fireChannelReadComplete();
        allocHandle.record(totalReadAmount);

        if (close) {
            closeOnRead();
            close = false;
        }
    } catch (Throwable t) {
        handleReadException(pipeline, byteBuf, t, close);
    } finally {
        // Check if there is a readPending which was not processed yet.
        // This could be for two reasons:
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
        //
        // See https://github.com/netty/netty/issues/2254
        if (!config.isAutoRead() && !((AbstractXnioUnsafe) unsafe()).readPending) {
            removeReadOp(channel);
        }
    }
}