EpollStreamUnsafe Class — netty Architecture

Source Code

transport-classes-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java lines 713–843

    class EpollStreamUnsafe extends AbstractEpollUnsafe {
        // Overridden here just to be able to access this method from AbstractEpollStreamChannel
        @Override
        protected Executor prepareToClose() {
            return super.prepareToClose();
        }

        private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause,
                                         boolean allDataRead, EpollRecvByteAllocatorHandle allocHandle) {
            if (byteBuf != null) {
                if (byteBuf.isReadable()) {
                    readPending = false;
                    pipeline.fireChannelRead(byteBuf);
                } else {
                    byteBuf.release();
                }
            }
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete();
            pipeline.fireExceptionCaught(cause);

            // If oom will close the read event, release connection.
            // See https://github.com/netty/netty/issues/10434
            if (allDataRead ||
                    cause instanceof OutOfMemoryError ||
                    cause instanceof LeakPresenceDetector.AllocationProhibitedException ||
                    cause instanceof IOException) {
                shutdownInput(true);
            }
        }

        @Override
        EpollRecvByteAllocatorHandle newEpollHandle(RecvByteBufAllocator.ExtendedHandle handle) {
            return new EpollRecvByteAllocatorStreamingHandle(handle);
        }

        @Override
        void epollInReady() {
            final ChannelConfig config = config();
            if (shouldBreakEpollInReady(config)) {
                clearEpollIn0();
                return;
            }
            final EpollRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
            final ChannelPipeline pipeline = pipeline();
            final ByteBufAllocator allocator = config.getAllocator();
            allocHandle.reset(config);

            ByteBuf byteBuf = null;
            boolean allDataRead = false;
            Queue<SpliceInTask> sQueue = null;
            try {
                do {
                    if (sQueue != null || (sQueue = spliceQueue) != null) {
                        SpliceInTask spliceTask = sQueue.peek();
                        if (spliceTask != null) {
                            boolean spliceInResult = spliceTask.spliceIn(allocHandle);

                            if (allocHandle.isReceivedRdHup()) {
                                shutdownInput(false);
                            }
                            if (spliceInResult) {
                                // We need to check if it is still active as if not we removed all SpliceTasks in
                                // doClose(...)
                                if (isActive()) {
                                    sQueue.remove();
                                }
                                continue;
                            } else {
                                break;
                            }
                        }
                    }

                    // we use a direct buffer here as the native implementations only be able
                    // to handle direct buffers.
                    byteBuf = allocHandle.allocate(allocator);
                    allocHandle.lastBytesRead(doReadBytes(byteBuf));
                    if (allocHandle.lastBytesRead() <= 0) {
                        // nothing was read, release the buffer.
                        byteBuf.release();