import org.xnio.ChannelListeners; //导入依赖的package包/类
public void sendGoAway(int status, final ChannelExceptionHandler<SpdyStreamSinkChannel> exceptionHandler) {
    if(thisGoneAway) {
        return;
    }
    thisGoneAway = true;
    SpdyGoAwayStreamSinkChannel goAway = new SpdyGoAwayStreamSinkChannel(this, status, lastGoodStreamId);
    try {
        goAway.shutdownWrites();
        if (!goAway.flush()) {
            goAway.getWriteSetter().set(ChannelListeners.flushingChannelListener(null, exceptionHandler));
            goAway.resumeWrites();
        }
    } catch (IOException e) {
        exceptionHandler.handleException(goAway, e);
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
HttpClientConnection(final StreamConnection connection, final OptionMap options, final Pool<ByteBuffer> bufferPool) {
    this.options = options;
    this.connection = connection;
    this.pushBackStreamSourceConduit = new PushBackStreamSourceConduit(connection.getSourceChannel().getConduit());
    this.connection.getSourceChannel().setConduit(pushBackStreamSourceConduit);
    this.bufferPool = bufferPool;
    this.originalSinkConduit = connection.getSinkChannel().getConduit();

    connection.getCloseSetter().set(new ChannelListener<StreamConnection>() {

        public void handleEvent(StreamConnection channel) {
            HttpClientConnection.this.state |= CLOSED;
            ChannelListeners.invokeChannelListener(HttpClientConnection.this, closeSetter.get());
        }
    });
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
@Override
public void handleEvent(StreamConnection channel) {
    try {
        for (CloseListener l : closeListeners) {
            try {
                l.closed(AbstractServerConnection.this);
            } catch (Throwable e) {
                UndertowLogger.REQUEST_LOGGER.exceptionInvokingCloseListener(l, e);
            }
        }
        if (current != null) {
            current.endExchange();
        }
        ChannelListeners.invokeChannelListener(AbstractServerConnection.this, listener);
    } finally {
        if(extraBytes != null) {
            extraBytes.free();
            extraBytes = null;
        }
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
@SuppressWarnings({"unchecked", "rawtypes"})
@Override
public void handleEvent(final StreamSourceChannel channel) {
    final R receiver = AbstractFramedChannel.this.receiver;
    if ((isLastFrameReceived() || receivesSuspended) && receiver == null) {
        channel.suspendReads();
        return;
    } else {
        final ChannelListener listener = receiveSetter.get();
        if (listener != null) {
            WebSocketLogger.REQUEST_LOGGER.debugf("Invoking receive listener", receiver);
            ChannelListeners.invokeChannelListener(AbstractFramedChannel.this, listener);
        } else {
            channel.suspendReads();
        }
    }
    if (readData != null && channel.isOpen()) {
        ChannelListeners.invokeChannelListener(channel.getIoThread(), channel, this);
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
@Override
public void close() throws IOException {
    if(anyAreSet(state, STATE_CLOSED)) {
        return;
    }
    state |= STATE_CLOSED;
    if (allAreClear(state, STATE_DONE | STATE_LAST_FRAME)) {
        state |= STATE_STREAM_BROKEN;
        getFramedChannel().notifyClosed(this);
        channelForciblyClosed();
    }
    if (data != null) {
        data.free();
        data = null;
    }
    while (!pendingFrameData.isEmpty()) {
        pendingFrameData.poll().frameData.free();
    }
    ChannelListeners.invokeChannelListener(this, (ChannelListener<? super AbstractFramedStreamSourceChannel<C, R, S>>) closeSetter.get());
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
@Override
public void run() {
    handle = null;
    if (expireTime == -1) {
        return;
    }
    long current = System.currentTimeMillis();
    if (current  < expireTime) {
        //timeout has been bumped, re-schedule
        handle = connection.getIoThread().executeAfter(timeoutCommand, (expireTime - current) + FUZZ_FACTOR, TimeUnit.MILLISECONDS);
        return;
    }
    UndertowLogger.REQUEST_LOGGER.tracef("Timing out channel %s due to inactivity");
    IoUtils.safeClose(connection);
    if (connection.getSourceChannel().isReadResumed()) {
        ChannelListeners.invokeChannelListener(connection.getSourceChannel(), connection.getSourceChannel().getReadListener());
    }
    if (connection.getSinkChannel().isWriteResumed()) {
        ChannelListeners.invokeChannelListener(connection.getSinkChannel(), connection.getSinkChannel().getWriteListener());
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
/**
 * Send a Close frame without a payload
 */
public void sendClose() throws IOException {
    closeReason = "";
    closeCode = CloseMessage.NORMAL_CLOSURE;
    StreamSinkFrameChannel closeChannel = send(WebSocketFrameType.CLOSE, 0);
    closeChannel.shutdownWrites();
    if (!closeChannel.flush()) {
        closeChannel.getWriteSetter().set(ChannelListeners.flushingChannelListener(
                null, new ChannelExceptionHandler<StreamSinkChannel>() {
            @Override
            public void handleException(final StreamSinkChannel channel, final IOException exception) {
                IoUtils.safeClose(WebSocketChannel.this);
            }
        }
        ));
        closeChannel.resumeWrites();
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
protected void writeDone(final StreamSinkChannel channel) {
    try {
        channel.shutdownWrites();

        if (!channel.flush()) {
            channel.getWriteSetter().set(ChannelListeners.flushingChannelListener(new ChannelListener<StreamSinkChannel>() {
                @Override
                public void handleEvent(StreamSinkChannel o) {
                    IoUtils.safeClose(channel);
                }
            }, ChannelListeners.closingChannelExceptionHandler()));
            channel.resumeWrites();

        }
    } catch (IOException e) {
        handleError(channel, e);
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
public void start() {
    try {

        OptionMap.Builder serverOptionsBuilder = OptionMap.builder()
                .set(Options.TCP_NODELAY, true)
                .set(Options.REUSE_ADDRESSES, true);
        ChannelListener acceptListener = ChannelListeners.openListenerAdapter(openListener);
        if (httpAddress != null) {
            normalServer = worker.createStreamConnectionServer(httpAddress, acceptListener, serverOptionsBuilder.getMap());
            normalServer.resumeAccepts();
        }
        if (secureAddress != null) {
            if (sslClientAuthMode != null) {
                serverOptionsBuilder.set(SSL_CLIENT_AUTH_MODE, sslClientAuthMode);
            }
            OptionMap secureOptions = serverOptionsBuilder.getMap();
            XnioSsl xnioSsl = new UndertowXnioSsl(worker.getXnio(), secureOptions, sslContext);
            secureServer = xnioSsl.createSslConnectionServer(worker, secureAddress, acceptListener, secureOptions);
            secureServer.resumeAccepts();
        }
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
private AcceptingChannel<? extends StreamConnection> createServer(int sourcePort, int targetPort)
    throws IllegalArgumentException, IOException {
    OptionMap socketOptions = OptionMap.builder()
        .set(Options.WORKER_IO_THREADS, 16)
        .set(Options.TCP_NODELAY, true)
        .set(Options.REUSE_ADDRESSES, true)
        .getMap();

    ChannelListener<AcceptingChannel<StreamConnection>> acceptListener = ChannelListeners.openListenerAdapter(
        new PortForwardOpenListener(connection, portForwardURI.getPath(), targetPort, requestId, bufferPoolSlice,
            OptionMap.EMPTY));
    AcceptingChannel<? extends StreamConnection> server =
        xnioWorker.createStreamConnectionServer(new InetSocketAddress(portForwardBindAddress, sourcePort),
            acceptListener, socketOptions);
    server.resumeAccepts();
    return server;
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
@Override
protected void complete() throws IOException {
    super.complete();
    if(completionListener != null) {
        ChannelListeners.invokeChannelListener(this, completionListener);
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
@Override
protected void handleFlushComplete(boolean finalFrame) {
    super.handleFlushComplete(finalFrame);
    if (finalFrame) {
        if (completionListener != null) {
            ChannelListeners.invokeChannelListener(this, completionListener);
        }
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
public void sendPing(int id, final ChannelExceptionHandler<SpdyStreamSinkChannel> exceptionHandler) {
    SpdyPingStreamSinkChannel ping = new SpdyPingStreamSinkChannel(this, id);
    try {
        ping.shutdownWrites();
        if (!ping.flush()) {
            ping.getWriteSetter().set(ChannelListeners.flushingChannelListener(null, exceptionHandler));
            ping.resumeWrites();
        }
    } catch (IOException e) {
        exceptionHandler.handleException(ping, e);
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
public void sendUpdateWindowSize(int streamId, int delta) {
    SpdyWindowUpdateStreamSinkChannel windowUpdateStreamSinkChannel = new SpdyWindowUpdateStreamSinkChannel(this, streamId, delta);
    try {
        windowUpdateStreamSinkChannel.shutdownWrites();
        if (!windowUpdateStreamSinkChannel.flush()) {
            windowUpdateStreamSinkChannel.getWriteSetter().set(ChannelListeners.flushingChannelListener(null, new SpdyControlMessageExceptionHandler()));
            windowUpdateStreamSinkChannel.resumeWrites();
        }
    } catch (IOException e) {
        handleBrokenSinkChannel(e);
    }

}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
AjpClientConnection(final AjpClientChannel connection, final OptionMap options, final Pool<ByteBuffer> bufferPool) {
    this.options = options;
    this.connection = connection;
    this.bufferPool = bufferPool;

    connection.addCloseTask(new ChannelListener<AjpClientChannel>() {
        @Override
        public void handleEvent(AjpClientChannel channel) {
            ChannelListeners.invokeChannelListener(AjpClientConnection.this, closeSetter.get());
        }
    });
    connection.getReceiveSetter().set(new ClientReceiveListener());
    connection.resumeReceives();
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
private void handleFailedFlush(AjpClientRequestClientStreamSinkChannel sinkChannel) {
    sinkChannel.getWriteSetter().set(ChannelListeners.flushingChannelListener(null, new ChannelExceptionHandler<StreamSinkChannel>() {
        @Override
        public void handleException(StreamSinkChannel channel, IOException exception) {
            handleError(exception);
        }
    }));
    sinkChannel.resumeWrites();
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
public SpdyClientConnection(SpdyChannel spdyChannel) {
    this.spdyChannel = spdyChannel;
    spdyChannel.getReceiveSetter().set(new SpdyReceiveListener());
    spdyChannel.resumeReceives();
    spdyChannel.addCloseTask(new ChannelListener<SpdyChannel>() {
        @Override
        public void handleEvent(SpdyChannel channel) {
            ChannelListeners.invokeChannelListener(SpdyClientConnection.this, closeSetter.get());
        }
    });
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
@Override
public void close() throws IOException {
    if(fullyFlushed || anyAreSet(state, STATE_CLOSED)) {
        return;
    }
    try {
        state |= STATE_CLOSED;
        buffer.free();
        buffer = null;
        if (header != null && header.getByteBuffer() != null) {
            header.getByteBuffer().free();
        }
        if (trailer != null) {
            trailer.free();
        }
        if (anyAreSet(state, STATE_FIRST_DATA_WRITTEN)) {
            channelForciblyClosed();
        }
        //we need to wake up/invoke the write listener
        if (isWriteResumed()) {
            ChannelListeners.invokeChannelListener(getIoThread(), this, (ChannelListener) getWriteListener());
        }
        wakeupWrites();
    } finally {
        wakeupWaiters();
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
public void markBroken() {
    this.broken = true;
    try {
        wakeupWrites();
        wakeupWaiters();
        if (isWriteResumed()) {
            ChannelListener<? super S> writeListener = this.writeSetter.get();
            if (writeListener != null) {
                ChannelListeners.invokeChannelListener(getIoThread(), (S) this, writeListener);
            }
        }
        ChannelListener<? super S> closeListener = this.closeSetter.get();
        if (closeListener != null) {
            ChannelListeners.invokeChannelListener(getIoThread(), (S) this, closeListener);
        }
    } finally {
        if(header != null && header.getByteBuffer() != null) {
            header.getByteBuffer().free();
        }
        if(trailer != null) {
            trailer.free();
        }
        if(buffer != null) {
            buffer.free();
            buffer = null;
        }
    }
}
 

import org.xnio.ChannelListeners; //导入依赖的package包/类
/**
 * For this class there is no difference between a resume and a wakeup
 */
void resumeReadsInternal(boolean wakeup) {
    boolean alreadyResumed = anyAreSet(state, STATE_READS_RESUMED);
    state |= STATE_READS_RESUMED;
    if(!alreadyResumed || wakeup) {
        if (!anyAreSet(state, STATE_IN_LISTENER_LOOP)) {
            state |= STATE_IN_LISTENER_LOOP;
            getIoThread().execute(new Runnable() {

                @Override
                public void run() {
                    try {
                        boolean moreData;
                        do {
                            ChannelListener<? super R> listener = getReadListener();
                            if (listener == null || !isReadResumed()) {
                                return;
                            }
                            ChannelListeners.invokeChannelListener((R) AbstractFramedStreamSourceChannel.this, listener);
                            //if writes are shutdown or we become active then we stop looping
                            //we stop when writes are shutdown because we can't flush until we are active
                            //although we may be flushed as part of a batch
                            moreData = (frameDataRemaining > 0 &&  data != null) || !pendingFrameData.isEmpty();
                        } while (allAreSet(state, STATE_READS_RESUMED) && allAreClear(state, STATE_CLOSED) && moreData);
                    } finally {
                        state &= ~STATE_IN_LISTENER_LOOP;
                    }
                }
            });
        }
    }
}