AbstractKQueueUnsafe Class — netty Architecture

Source Code

transport-classes-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java lines 372–690

    @UnstableApi
    public abstract class AbstractKQueueUnsafe extends AbstractUnsafe implements KQueueIoHandle {
        boolean readPending;
        private KQueueRecvByteAllocatorHandle allocHandle;

        Channel channel() {
            return AbstractKQueueChannel.this;
        }

        @Override
        public int ident() {
            return fd().intValue();
        }

        @Override
        public void close() {
            close(voidPromise());
        }

        @Override
        public void handle(IoRegistration registration, IoEvent event) {
            KQueueIoEvent kqueueEvent = (KQueueIoEvent) event;
            final short filter = kqueueEvent.filter();
            final short flags = kqueueEvent.flags();
            final int fflags = kqueueEvent.fflags();
            final long data = kqueueEvent.data();

            // First check for EPOLLOUT as we may need to fail the connect ChannelPromise before try
            // to read from the file descriptor.
            if (filter == Native.EVFILT_WRITE) {
                writeReady();
            } else if (filter == Native.EVFILT_READ) {
                // Check READ before EOF to ensure all data is read before shutting down the input.
                KQueueRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
                readReady(allocHandle);
            } else if (filter == Native.EVFILT_SOCK && (fflags & Native.NOTE_RDHUP) != 0) {
                readEOF();
                return;
            }

            // Check if EV_EOF was set, this will notify us for connection-reset in which case
            // we may close the channel directly or try to read more data depending on the state of the
            // Channel and also depending on the AbstractKQueueChannel subtype.
            if ((flags & Native.EV_EOF) != 0) {
                readEOF();
            }
        }

        abstract void readReady(KQueueRecvByteAllocatorHandle allocHandle);

        final boolean shouldStopReading(ChannelConfig config) {
            // 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
            return !readPending && !config.isAutoRead();
        }

        final boolean failConnectPromise(Throwable cause) {
            if (connectPromise != null) {
                // SO_ERROR has been shown to return 0 on macOS if detect an error via read() and the write filter was
                // not set before calling connect. This means finishConnect will not detect any error and would
                // successfully complete the connectPromise and update the channel state to active (which is incorrect).
                ChannelPromise connectPromise = AbstractKQueueChannel.this.connectPromise;
                AbstractKQueueChannel.this.connectPromise = null;
                if (connectPromise.tryFailure((cause instanceof ConnectException) ? cause
                                : new ConnectException("failed to connect").initCause(cause))) {
                    closeIfClosed();
                    return true;
                }
            }
            return false;
        }

        private void writeReady() {
            if (connectPromise != null) {
                // pending connect which is now complete so handle it.
                finishConnect();
            } else if (!socket.isOutputShutdown()) {