/**
 * 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;
}
 

@Override
public ConduitStreamSinkChannel getSinkChannel() {

    ConduitStreamSinkChannel sinkChannel = new ConduitStreamSinkChannel(
            Configurable.EMPTY,
            new BufferedStreamSinkConduit(
                    new NullStreamSinkConduit(worker.getIoThread()),
                    new PooledAdaptor(bufferPool.allocate())
            )
    );
    sinkChannel.setCloseListener(conduitStreamSinkChannel -> {
        for (CloseListener l : closeListeners) {
            try {
                l.closed(InVMConnection.this);
            } catch (Throwable e) {
                UndertowLogger.REQUEST_LOGGER.exceptionInvokingCloseListener(l, e);
            }
        }
    });
    return sinkChannel;
}
 

@Override
public ConduitStreamSinkChannel getSinkChannel() {

    ConduitStreamSinkChannel sinkChannel = new ConduitStreamSinkChannel(
            Configurable.EMPTY,
            new BufferedStreamSinkConduit(
                    new NullStreamSinkConduit(worker.getIoThread()),
                    new PooledAdaptor(bufferPool.allocate())
            )
    );
    sinkChannel.setCloseListener(conduitStreamSinkChannel -> {
        for (CloseListener l : closeListeners) {
            try {
                l.closed(InVMConnection.this);
            } catch (Throwable e) {
                UndertowLogger.REQUEST_LOGGER.exceptionInvokingCloseListener(l, e);
            }
        }
    });
    return sinkChannel;
}
 

@Override
public void writeReady() {
    if(anyAreSet(state, FLAG_READ_REQUIRES_WRITE)) {
        if(anyAreSet(state, FLAG_READS_RESUMED)) {
            readReadyHandler.readReady();
        } else {
            try {
                doHandshake();
            } catch (IOException e) {
                UndertowLogger.REQUEST_LOGGER.ioException(e);
                IoUtils.safeClose(delegate);
            } catch (Throwable t) {
                UndertowLogger.REQUEST_LOGGER.handleUnexpectedFailure(t);
                IoUtils.safeClose(delegate);
            }
        }
    }
    if (anyAreSet(state, FLAG_WRITES_RESUMED)) {
        if(delegateHandler == null) {
                final ChannelListener<? super ConduitStreamSinkChannel> writeListener = connection.getSinkChannel().getWriteListener();
                if (writeListener == null) {
                    suspendWrites();
                } else {
                    ChannelListeners.invokeChannelListener(connection.getSinkChannel(), writeListener);
                }
        } else {
            delegateHandler.writeReady();
        }
    }
    if(!anyAreSet(state, FLAG_WRITES_RESUMED | FLAG_READ_REQUIRES_WRITE)) {
        delegate.getSinkChannel().suspendWrites();
    }
}
 

public Http2ServerConnection(Http2Channel channel, Http2StreamSourceChannel requestChannel, OptionMap undertowOptions, int bufferSize, HttpHandler rootHandler) {
    this.channel = channel;
    this.requestChannel = requestChannel;
    this.undertowOptions = undertowOptions;
    this.bufferSize = bufferSize;
    this.rootHandler = rootHandler;
    responseChannel = requestChannel.getResponseChannel();
    originalSinkConduit = new StreamSinkChannelWrappingConduit(responseChannel);
    originalSourceConduit = new StreamSourceChannelWrappingConduit(requestChannel);
    this.conduitStreamSinkChannel = new ConduitStreamSinkChannel(responseChannel, originalSinkConduit);
    this.conduitStreamSourceChannel = new ConduitStreamSourceChannel(channel, originalSourceConduit);
}
 

/**
 * Channel that is used when the request is already half closed
 * @param channel
 * @param undertowOptions
 * @param bufferSize
 * @param rootHandler
 */
public Http2ServerConnection(Http2Channel channel, Http2DataStreamSinkChannel sinkChannel, OptionMap undertowOptions, int bufferSize, HttpHandler rootHandler) {
    this.channel = channel;
    this.rootHandler = rootHandler;
    this.requestChannel = null;
    this.undertowOptions = undertowOptions;
    this.bufferSize = bufferSize;
    responseChannel = sinkChannel;
    originalSinkConduit = new StreamSinkChannelWrappingConduit(responseChannel);
    originalSourceConduit = new StreamSourceChannelWrappingConduit(requestChannel);
    this.conduitStreamSinkChannel = new ConduitStreamSinkChannel(responseChannel, originalSinkConduit);
    this.conduitStreamSourceChannel = new ConduitStreamSourceChannel(Configurable.EMPTY, new EmptyStreamSourceConduit(getIoThread()));
}
 

@Override
public ChannelListener.Setter<? extends StreamSinkChannel> getWriteSetter() {
    if (writeSetter == null) {
        writeSetter = new ChannelListener.SimpleSetter<>();
        if (!isFinished()) {
            if(delegate instanceof ConduitStreamSinkChannel) {
                ((ConduitStreamSinkChannel) delegate).setWriteListener(new SetterDelegatingListener((ChannelListener.SimpleSetter)writeSetter, this));
            } else {
                delegate.getWriteSetter().set(new SetterDelegatingListener((ChannelListener.SimpleSetter)writeSetter, this));
            }
        }
    }
    return writeSetter;
}
 

public void responseDone() {
    if(delegate instanceof ConduitStreamSinkChannel) {
        ((ConduitStreamSinkChannel) delegate).setCloseListener(null);
        ((ConduitStreamSinkChannel) delegate).setWriteListener(null);
    } else {
        delegate.getCloseSetter().set(null);
        delegate.getWriteSetter().set(null);
    }
    if (delegate.isWriteResumed()) {
        delegate.suspendWrites();
    }
}
 

private void setOpWrite() {
    ConduitStreamSinkChannel sink = connection().getSinkChannel();
    if (!sink.isWriteResumed()) {
        ChannelListener<ConduitStreamSinkChannel> writeListener = this.writeListener;
        if (writeListener == null) {
            writeListener = this.writeListener = new WriteListener();
        }
        sink.getWriteSetter().set(writeListener);
        sink.resumeWrites();
    }
}
 

@Override
public ConduitStreamSinkChannel getSinkChannel() {
    return null;
}
 

private void initiateRequest(HttpClientExchange httpClientExchange) {
    this.requestCount++;
    currentRequest = httpClientExchange;
    pendingResponse = new HttpResponseBuilder();
    ClientRequest request = httpClientExchange.getRequest();

    String connectionString = request.getRequestHeaders().getFirst(Headers.CONNECTION);
    if (connectionString != null) {
        if (Headers.CLOSE.equalToString(connectionString)) {
            state |= CLOSE_REQ;
        } else if (Headers.UPGRADE.equalToString(connectionString)) {
            state |= UPGRADE_REQUESTED;
        }
    } else if (request.getProtocol() != Protocols.HTTP_1_1) {
        state |= CLOSE_REQ;
    }
    if (request.getRequestHeaders().contains(Headers.UPGRADE)) {
        state |= UPGRADE_REQUESTED;
    }
    if(request.getMethod().equals(Methods.CONNECT)) {
        //we treat CONNECT like upgrade requests
        state |= UPGRADE_REQUESTED;
    }

    //setup the client request conduits
    final ConduitStreamSourceChannel sourceChannel = connection.getSourceChannel();
    sourceChannel.setReadListener(clientReadListener);
    sourceChannel.resumeReads();

    ConduitStreamSinkChannel sinkChannel = connection.getSinkChannel();
    StreamSinkConduit conduit = originalSinkConduit;
    HttpRequestConduit httpRequestConduit = new HttpRequestConduit(conduit, bufferPool, request);
    httpClientExchange.setRequestConduit(httpRequestConduit);
    conduit = httpRequestConduit;

    String fixedLengthString = request.getRequestHeaders().getFirst(Headers.CONTENT_LENGTH);
    String transferEncodingString = request.getRequestHeaders().getLast(Headers.TRANSFER_ENCODING);

    boolean hasContent = true;

    if (fixedLengthString != null) {
        try {
            long length = Long.parseLong(fixedLengthString);
            conduit = new ClientFixedLengthStreamSinkConduit(conduit, length, false, false, currentRequest);
            hasContent = length != 0;
        } catch (NumberFormatException e) {
            handleError(e);
            return;
        }
    } else if (transferEncodingString != null) {
        if (!transferEncodingString.toLowerCase(Locale.ENGLISH).contains(Headers.CHUNKED.toString())) {
            handleError(UndertowClientMessages.MESSAGES.unknownTransferEncoding(transferEncodingString));
            return;
        }
        conduit = new ChunkedStreamSinkConduit(conduit, httpClientExchange.getConnection().getBufferPool(), false, false, httpClientExchange.getRequest().getRequestHeaders(), requestFinishListener, httpClientExchange);
    } else {
        conduit = new ClientFixedLengthStreamSinkConduit(conduit, 0, false, false, currentRequest);
        hasContent = false;
    }
    sinkChannel.setConduit(conduit);

    httpClientExchange.invokeReadReadyCallback();
    if (!hasContent) {
        //if there is no content we flush the response channel.
        //otherwise it is up to the user
        try {
            sinkChannel.shutdownWrites();
            if (!sinkChannel.flush()) {
                sinkChannel.setWriteListener(ChannelListeners.flushingChannelListener(null, new ChannelExceptionHandler<ConduitStreamSinkChannel>() {
                    @Override
                    public void handleException(ConduitStreamSinkChannel channel, IOException exception) {
                        handleError(exception);
                    }
                }));
                sinkChannel.resumeWrites();
            }
        } catch (Throwable t) {
            handleError(t);
        }
    }
}
 

@Override
protected ConduitStreamSinkChannel getSinkChannel() {
    return channel.getSinkChannel();
}
 

public static ConduitStreamSinkChannel getConduitSinkChannel(HttpServerExchange exchange) {
    return exchange.getConnection().getSinkChannel();
}
 

@Override
protected ConduitStreamSinkChannel getSinkChannel() {
    return conduitStreamSinkChannel;
}
 

/**
 * Gets a pre-encoded resource.
 * <p>
 * TODO: blocking / non-blocking semantics
 *
 * @param resource
 * @param exchange
 * @return
 * @throws IOException
 */
public ContentEncodedResource getResource(final Resource resource, final HttpServerExchange exchange) throws IOException {
    final String path = resource.getPath();
    Path file = resource.getFilePath();
    if (file == null) {
        return null;
    }
    if (minResourceSize > 0 && resource.getContentLength() < minResourceSize ||
            maxResourceSize > 0 && resource.getContentLength() > maxResourceSize ||
            !(encodingAllowed == null || encodingAllowed.resolve(exchange))) {
        return null;
    }
    AllowedContentEncodings encodings = contentEncodingRepository.getContentEncodings(exchange);
    if (encodings == null || encodings.isNoEncodingsAllowed()) {
        return null;
    }
    EncodingMapping encoding = encodings.getEncoding();
    if (encoding == null || encoding.getName().equals(ContentEncodingRepository.IDENTITY)) {
        return null;
    }
    String newPath = path + ".undertow.encoding." + encoding.getName();
    Resource preCompressed = encoded.getResource(newPath);
    if (preCompressed != null) {
        return new ContentEncodedResource(preCompressed, encoding.getName());
    }
    final LockKey key = new LockKey(path, encoding.getName());
    if (fileLocks.putIfAbsent(key, this) != null) {
        //another thread is already compressing
        //we don't do anything fancy here, just return and serve non-compressed content
        return null;
    }
    FileChannel targetFileChannel = null;
    FileChannel sourceFileChannel = null;
    try {
        //double check, the compressing thread could have finished just before we acquired the lock
        preCompressed = encoded.getResource(newPath);
        if (preCompressed != null) {
            return new ContentEncodedResource(preCompressed, encoding.getName());
        }

        final Path finalTarget = encodedResourcesRoot.resolve(newPath);
        final Path tempTarget = encodedResourcesRoot.resolve(newPath);

        //horrible hack to work around XNIO issue
        OutputStream tmp = Files.newOutputStream(tempTarget);
        try {
            tmp.close();
        } finally {
            IoUtils.safeClose(tmp);
        }

        targetFileChannel = FileChannel.open(tempTarget, StandardOpenOption.READ, StandardOpenOption.WRITE);
        sourceFileChannel = FileChannel.open(file, StandardOpenOption.READ);

        StreamSinkConduit conduit = encoding.getEncoding().getResponseWrapper().wrap(new ImmediateConduitFactory<StreamSinkConduit>(new FileConduitTarget(targetFileChannel, exchange)), exchange);
        final ConduitStreamSinkChannel targetChannel = new ConduitStreamSinkChannel(null, conduit);
        long transferred = sourceFileChannel.transferTo(0, resource.getContentLength(), targetChannel);
        targetChannel.shutdownWrites();
        org.xnio.channels.Channels.flushBlocking(targetChannel);
        if (transferred != resource.getContentLength()) {
            UndertowLogger.REQUEST_LOGGER.failedToWritePreCachedFile();
        }
        Files.move(tempTarget, finalTarget);
        encoded.invalidate(newPath);
        final Resource encodedResource = encoded.getResource(newPath);
        return new ContentEncodedResource(encodedResource, encoding.getName());
    } finally {
        IoUtils.safeClose(targetFileChannel);
        IoUtils.safeClose(sourceFileChannel);
        fileLocks.remove(key);
    }
}
 

WriteDispatchChannel(final ConduitStreamSinkChannel delegate) {
    super(delegate);
}
 

@Override
public void handleEvent(ConduitStreamSinkChannel channel) {
    ((AbstractXnioUnsafe)unsafe()).forceFlush();
}
 

protected abstract ConduitStreamSinkChannel getSinkChannel();