KQueueDomainUnsafe Class — netty Architecture

Source Code

transport-classes-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannel.java lines 127–188

    private final class KQueueDomainUnsafe extends KQueueStreamUnsafe {
        @Override
        void readReady(KQueueRecvByteAllocatorHandle allocHandle) {
            switch (config().getReadMode()) {
                case BYTES:
                    super.readReady(allocHandle);
                    break;
                case FILE_DESCRIPTORS:
                    readReadyFd();
                    break;
                default:
                    throw new Error("Unexpected read mode: " + config().getReadMode());
            }
        }

        private void readReadyFd() {
            if (socket.isInputShutdown()) {
                super.clearReadFilter0();
                return;
            }
            final ChannelConfig config = config();
            final KQueueRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();

            final ChannelPipeline pipeline = pipeline();
            allocHandle.reset(config);

            try {
                readLoop: do {
                    // lastBytesRead represents the fd. We use lastBytesRead because it must be set so that the
                    // KQueueRecvByteAllocatorHandle knows if it should try to read again or not when autoRead is
                    // enabled.
                    int recvFd = socket.recvFd();
                    switch(recvFd) {
                        case 0:
                            allocHandle.lastBytesRead(0);
                            break readLoop;
                        case -1:
                            allocHandle.lastBytesRead(-1);
                            close(voidPromise());
                            return;
                        default:
                            allocHandle.lastBytesRead(1);
                            allocHandle.incMessagesRead(1);
                            readPending = false;
                            pipeline.fireChannelRead(new FileDescriptor(recvFd));
                            break;
                    }
                } while (allocHandle.continueReading());

                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();
            } catch (Throwable t) {
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();
                pipeline.fireExceptionCaught(t);
            } finally {
                if (shouldStopReading(config)) {
                    clearReadFilter0();
                }
            }
        }
    }