import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
/**
 * response
 * 
 * @param exchange
 * @param statusCode
 * @param output
 *            auto release
 */
protected final void send(HttpServerExchange exchange, int statusCode, PooledByteBufferOutputStream output) {
	try {
		output.flip();

		StreamSinkChannel channel = getResponseChannel(exchange);
		Sender sender = exchange.getResponseSender();

		setStatusCode(exchange, statusCode);
		setResponseChannel(sender, channel);
		setPooledBuffers(sender, output.getPooledByteBuffers());

		sender.send(output.getByteBuffers());
	} catch (Throwable t) {
		UndertowLogger.REQUEST_IO_LOGGER.handleUnexpectedFailure(t);
	}
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
private void sendBadRequestAndClose(final StreamConnection connection, final Exception exception) {
    UndertowLogger.REQUEST_IO_LOGGER.failedToParseRequest(exception);
    connection.getSourceChannel().suspendReads();
    new StringWriteChannelListener(BAD_REQUEST) {
        @Override
        protected void writeDone(final StreamSinkChannel c) {
            super.writeDone(c);
            c.suspendWrites();
            IoUtils.safeClose(connection);
        }

        @Override
        protected void handleError(StreamSinkChannel channel, IOException e) {
            IoUtils.safeClose(connection);
        }
    }.setup(connection.getSinkChannel());
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
@Override
public long transferTo(final long count, final ByteBuffer throughBuffer, final StreamSinkChannel target) throws IOException {
    if (exchange.getResponseCode() == StatusCodes.EXPECTATION_FAILED) {
        //rejected
        return -1;
    }
    if (!sent) {
        sent = true;
        response = HttpContinue.createResponseSender(exchange);
    }
    if (response != null) {
        if (!response.send()) {
            return 0;
        }
        response = null;
    }
    return super.transferTo(count, throughBuffer, target);
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
/**
 * Get the response channel. The channel must be closed and fully flushed before the next response can be started.
 * In order to close the channel you must first call {@link org.xnio.channels.StreamSinkChannel#shutdownWrites()},
 * and then call {@link org.xnio.channels.StreamSinkChannel#flush()} until it returns true. Alternatively you can
 * call {@link #endExchange()}, which will close the channel as part of its cleanup.
 * <p/>
 * Closing a fixed-length response before the corresponding number of bytes has been written will cause the connection
 * to be reset and subsequent requests to fail; thus it is important to ensure that the proper content length is
 * delivered when one is specified.  The response channel may not be writable until after the response headers have
 * been sent.
 * <p/>
 * If this method is not called then an empty or default response body will be used, depending on the response code set.
 * <p/>
 * The returned channel will begin to write out headers when the first write request is initiated, or when
 * {@link org.xnio.channels.StreamSinkChannel#shutdownWrites()} is called on the channel with no content being written.
 * Once the channel is acquired, however, the response code and headers may not be modified.
 * <p/>
 *
 * @return the response channel, or {@code null} if another party already acquired the channel
 */
public StreamSinkChannel getResponseChannel() {
    if (responseChannel != null) {
        return null;
    }
    final ConduitWrapper<StreamSinkConduit>[] wrappers = responseWrappers;
    this.responseWrappers = null;
    final ConduitStreamSinkChannel sinkChannel = connection.getSinkChannel();
    if (sinkChannel == null) {
        return null;
    }
    if(wrappers != null) {
        final WrapperStreamSinkConduitFactory factory = new WrapperStreamSinkConduitFactory(wrappers, responseWrapperCount, this, sinkChannel.getConduit());
        sinkChannel.setConduit(factory.create());
    } else {
        sinkChannel.setConduit(connection.getSinkConduit(this, sinkChannel.getConduit()));
    }
    this.responseChannel = new WriteDispatchChannel(sinkChannel);
    this.startResponse();
    return responseChannel;
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
public long transferTo(final long count, final ByteBuffer throughBuffer, final StreamSinkChannel target) throws IOException {
    if (count == 0L) {
        return 0L;
    }
    long val = state;
    checkMaxSize(val);
    if (anyAreSet(val, FLAG_CLOSED | FLAG_FINISHED) || allAreClear(val, MASK_COUNT)) {
        if (allAreClear(val, FLAG_FINISHED)) {
            invokeFinishListener();
        }
        return -1;
    }
    long res = 0L;
    try {
        return res = next.transferTo(min(count, val & MASK_COUNT), throughBuffer, target);
    } catch (IOException | RuntimeException e) {
        IoUtils.safeClose(exchange.getConnection());
        throw e;
    } finally {
        exitRead(res == -1L ? val & MASK_COUNT : res + throughBuffer.remaining());
    }
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的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.channels.StreamSinkChannel; //导入依赖的package包/类
public void setup(final StreamSinkChannel channel) {
    try {
        int c;
        do {
            c = channel.write(buffer);
        } while (buffer.hasRemaining() && c > 0);
        if (buffer.hasRemaining()) {
            channel.getWriteSetter().set(this);
            channel.resumeWrites();
        } else {
            writeDone(channel);
        }
    } catch (IOException e) {
        handleError(channel, e);
    }
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
@Override
public void handleEvent(final StreamSinkChannel channel) {
    try {
        int c;
        do {
            c = channel.write(buffer);
        } while (buffer.hasRemaining() && c > 0);
        if (buffer.hasRemaining()) {
            channel.resumeWrites();
            return;
        } else {
            writeDone(channel);
        }
    } catch (IOException e) {
        handleError(channel, e);
    }
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的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.channels.StreamSinkChannel; //导入依赖的package包/类
private void initWriteListener() {
    writeListener = new ChannelListener<StreamSinkChannel>() {
        @Override
        public void handleEvent(final StreamSinkChannel streamSinkChannel) {
            try {
                long toWrite = Buffers.remaining(buffer);
                long written = 0;
                while (written < toWrite) {
                    long res = streamSinkChannel.write(buffer, 0, buffer.length);
                    written += res;
                    if (res == 0) {
                        return;
                    }
                }
                streamSinkChannel.suspendWrites();
                invokeOnComplete();
            } catch (IOException e) {
                streamSinkChannel.suspendWrites();
                invokeOnException(callback, e);
            }
        }
    };
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
public void send(Object obj) {
    if (httpServerExchange.getRequestMethod()
            .equals(HEAD)) {
        httpServerExchange.endExchange();
        return;
    }
    httpServerExchange.getResponseHeaders()
            .put(Headers.CONTENT_TYPE, JSON_CONTENT_TYPE);

    ObjectMapper objectMapper = LOCAL_MAPPER.get();
    if (objectMapper == null) {
        LOGGER.info("Creating new parser.........");
        objectMapper = JsonFactory.createUseJSONDates();
        LOCAL_MAPPER.set(objectMapper);
    }

    String jsonStr = objectMapper
            .toJson(obj);
    ByteBuffer byteBuffer = ByteBuffer.wrap(jsonStr.getBytes(StandardCharsets.UTF_8));
    StreamSinkChannel responseChannel = httpServerExchange.getResponseChannel();
    int written;
    try {
        do {
            written = responseChannel.write(byteBuffer);
        } while (byteBuffer.hasRemaining() && written > 0);
    } catch (IOException e) {
        LOGGER.error("Can not write response: " + e.getMessage(), e);
    }
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
private void sendResponse(final HttpServerExchange exchange, final AbstractResponse res, final long beginningTime)
		throws IOException {
	res.setDuration((int) (System.currentTimeMillis() - beginningTime));
	final String response = gson.toJson(res);

	if (res instanceof ErrorResponse) {
		exchange.setStatusCode(400); // bad request
	} else if (res instanceof AccessLimitedResponse) {
		exchange.setStatusCode(401); // api method not allowed
	} else if (res instanceof ExceptionResponse) {
		exchange.setStatusCode(500); // internal error
	}

	setupResponseHeaders(exchange);

	ByteBuffer responseBuf = ByteBuffer.wrap(response.getBytes(StandardCharsets.UTF_8));
	exchange.setResponseContentLength(responseBuf.array().length);
	StreamSinkChannel sinkChannel = exchange.getResponseChannel();
	sinkChannel.getWriteSetter().set(channel -> {
		if (responseBuf.remaining() > 0)
			try {
				sinkChannel.write(responseBuf);
				if (responseBuf.remaining() == 0) {
					exchange.endExchange();
				}
			} catch (IOException e) {
				log.error("Error writing response", e);
				exchange.endExchange();
				sinkChannel.getWriteSetter().set(null);
			}
		else {
			exchange.endExchange();
		}
	});
	sinkChannel.resumeWrites();
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
private void handleError(final StreamSinkChannel channel, final IOException e) {
    try {
        listener.onError(e);
    } finally {
        IoUtils.safeClose(channel);
    }
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
@Override
public long transferTo(long count, ByteBuffer throughBuffer, StreamSinkChannel streamSinkChannel) throws IOException {
    handleNewHeaders();
    long read = super.transferTo(count, throughBuffer, streamSinkChannel);
    updateFlowControlWindow((int) read + throughBuffer.remaining());
    return read;
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的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.channels.StreamSinkChannel; //导入依赖的package包/类
@Override
public StreamSinkChannel getRequestChannel() {
    return new DetachableStreamSinkChannel(requestChannel) {
        @Override
        protected boolean isFinished() {
            return anyAreSet(state, REQUEST_TERMINATED);
        }
    };
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
@Override
public StreamSinkChannel getRequestChannel() {
    return new DetachableStreamSinkChannel(clientConnection.getConnection().getSinkChannel()) {
        @Override
        protected boolean isFinished() {
            return anyAreSet(state, REQUEST_TERMINATED);
        }
    };
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
@Override
public long transferTo(long count, ByteBuffer throughBuffer, StreamSinkChannel target) throws IOException {
    try {
        return IoUtils.transfer(new ConduitReadableByteChannel(this), count, throughBuffer, target);
    } catch (IOException | RuntimeException e) {
        IoUtils.safeClose(exchange.getConnection());
        throw e;
    }
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
@Override
public long transferTo(long count, ByteBuffer throughBuffer, StreamSinkChannel target) throws IOException {
    Pooled<ByteBuffer>[] buffered = getAttachment(BUFFERED_REQUEST_DATA);
    if (buffered == null) {
        return super.transferTo(count, throughBuffer, target);
    }
    //make sure there is no garbage in throughBuffer
    throughBuffer.position(0);
    throughBuffer.limit(0);
    long copied = 0;
    for (int i = 0; i < buffered.length; ++i) {
        Pooled<ByteBuffer> pooled = buffered[i];
        if (pooled != null) {
            final ByteBuffer buf = pooled.getResource();
            if (buf.hasRemaining()) {
                int res = target.write(buf);

                if (!buf.hasRemaining()) {
                    pooled.free();
                    buffered[i] = null;
                }
                if (res == 0) {
                    return copied;
                } else {
                    copied += res;
                }
            } else {
                pooled.free();
                buffered[i] = null;
            }
        }
    }
    removeAttachment(BUFFERED_REQUEST_DATA);
    if (copied == 0) {
        return super.transferTo(count, throughBuffer, target);
    } else {
        return copied;
    }
}
 

import org.xnio.channels.StreamSinkChannel; //导入依赖的package包/类
public long transferTo(final long count, final ByteBuffer throughBuffer, final StreamSinkChannel target) throws IOException {
    long res = 0;
    try {
        return res = next.transferTo(count, throughBuffer, target);
    } finally {
        exitRead(res);
    }
}