import java.lang.System.Logger.Level;

import jdk.incubator.http.HttpResponse.BodySubscriber;

import java.util.Objects;

import java.util.concurrent.ConcurrentLinkedDeque;

import java.util.concurrent.Flow;

import jdk.incubator.http.internal.common.Demand;

import jdk.incubator.http.internal.common.FlowTube;

import jdk.incubator.http.internal.common.SequentialScheduler;

import static jdk.incubator.http.HttpClient.Version.HTTP_1_1;

 * Encapsulates one HTTP/1.1 request/response exchange.

    static final boolean DEBUG = Utils.DEBUG; // Revisit: temporary dev flag.

    final System.Logger  debug = Utils.getDebugLogger(this::dbgString, DEBUG);

    private static final System.Logger DEBUG_LOGGER =

            Utils.getDebugLogger("Http1Exchange"::toString, DEBUG);

    final HttpRequestImpl request; // main request

    private final Http1AsyncReceiver asyncReceiver;

    /** Records a possible cancellation raised before any operation

     * has been initiated, or an error received while sending the request. */

    private Throwable failed;

    private final List<CompletableFuture<?>> operations; // used for cancel

    /** Must be held when operating on any internal state or data. */

    private final Object lock = new Object();

    /** Holds the outgoing data, either the headers or a request body part. Or

     * an error from the request body publisher. At most there can be ~2 pieces

     * of outgoing data ( onComplete|onError can be invoked without demand ).*/

    final ConcurrentLinkedDeque<DataPair> outgoing = new ConcurrentLinkedDeque<>();

    /** The write publisher, responsible for writing the complete request ( both

     * headers and body ( if any ). */

    private final Http1Publisher writePublisher = new Http1Publisher();

    /** Completed when the header have been published, or there is an error */

    private volatile CompletableFuture<ExchangeImpl<T>> headersSentCF  = new MinimalFuture<>();

     /** Completed when the body has been published, or there is an error */

    private volatile CompletableFuture<ExchangeImpl<T>> bodySentCF = new MinimalFuture<>();

    /** The subscriber to the request's body published. Maybe null. */

    private volatile Http1BodySubscriber bodySubscriber;

    enum State { INITIAL,

                 HEADERS,

                 BODY,

                 ERROR,          // terminal state

                 COMPLETING,

                 COMPLETED }     // terminal state

    private State state = State.INITIAL;

    /** A carrier for either data or an error. Used to carry data, and communicate

     * errors from the request ( both headers and body ) to the exchange. */

    static class DataPair {

        Throwable throwable;

        List<ByteBuffer> data;

        DataPair(List<ByteBuffer> data, Throwable throwable){

            this.data = data;

            this.throwable = throwable;

        }

        @Override

        public String toString() {

            return "DataPair [data=" + data + ", throwable=" + throwable + "]";

        }

    }

    /** An abstract supertype for HTTP/1.1 body subscribers. There are two

     * concrete implementations: {@link Http1Request.StreamSubscriber}, and

     * {@link Http1Request.FixedContentSubscriber}, for receiving chunked and

     * fixed length bodies, respectively. */

    static abstract class Http1BodySubscriber implements Flow.Subscriber<ByteBuffer> {

        protected volatile Flow.Subscription subscription;

        protected volatile boolean complete;

        /** Final sentinel in the stream of request body. */

        static final List<ByteBuffer> COMPLETED = List.of(ByteBuffer.allocate(0));

        void request(long n) {

            DEBUG_LOGGER.log(Level.DEBUG, () ->

                "Http1BodySubscriber requesting " + n + ", from " + subscription);

            subscription.request(n);

        }

        static Http1BodySubscriber completeSubscriber() {

            return new Http1BodySubscriber() {

                @Override public void onSubscribe(Flow.Subscription subscription) { error(); }

                @Override public void onNext(ByteBuffer item) { error(); }

                @Override public void onError(Throwable throwable) { error(); }

                @Override public void onComplete() { error(); }

                private void error() {

                    throw new InternalError("should not reach here");

                }

            };

        }

    }

        return "HTTP/1.1 " + request.toString();

        operations.add(headersSentCF);

        operations.add(bodySentCF);

            InetSocketAddress addr = request.getAddress();

            this.connection = HttpConnection.getConnection(addr, client, request, HTTP_1_1);

        this.requestAction = new Http1Request(request, this);

        this.asyncReceiver = new Http1AsyncReceiver(executor, this);

        asyncReceiver.subscribe(new InitialErrorReceiver());

    /** An initial receiver that handles no data, but cancels the request if

     * it receives an error. Will be replaced when reading response body. */

    final class InitialErrorReceiver implements Http1AsyncReceiver.Http1AsyncDelegate {

        volatile AbstractSubscription s;

        @Override

        public boolean tryAsyncReceive(ByteBuffer ref) {

            return false;  // no data has been processed, leave it in the queue

        }

        @Override

        public void onReadError(Throwable ex) {

            cancelImpl(ex);

        }

        @Override

        public void onSubscribe(AbstractSubscription s) {

            this.s = s;

        }

        public AbstractSubscription subscription() {

            return s;

        }

    }

    @Override

    private void connectFlows(HttpConnection connection) {

        FlowTube tube =  connection.getConnectionFlow();

        debug.log(Level.DEBUG, "%s connecting flows", tube);

        // Connect the flow to our Http1TubeSubscriber:

        //   asyncReceiver.subscriber().

        tube.connectFlows(writePublisher,

                          asyncReceiver.subscriber());

    }

    CompletableFuture<ExchangeImpl<T>> sendHeadersAsync() {

        // create the response before sending the request headers, so that

        // the response can set the appropriate receivers.

        debug.log(Level.DEBUG, "Sending headers only");

        if (response == null) {

            response = new Http1Response<>(connection, this, asyncReceiver);

        debug.log(Level.DEBUG, "response created in advance");

        // If the first attempt to read something triggers EOF, or

        // IOException("channel reset by peer"), we're going to retry.

        // Instruct the asyncReceiver to throw ConnectionExpiredException

        // to force a retry.

        asyncReceiver.setRetryOnError(true);

        CompletableFuture<Void> connectCF;

        if (!connection.connected()) {

            debug.log(Level.DEBUG, "initiating connect async");

            connectCF = connection.connectAsync();

            synchronized (lock) {

                operations.add(connectCF);

            }

        } else {

            connectCF = new MinimalFuture<>();

            connectCF.complete(null);

        }

        return connectCF

                .thenCompose(unused -> {

                    CompletableFuture<Void> cf = new MinimalFuture<>();

                    try {

                        connectFlows(connection);

                        debug.log(Level.DEBUG, "requestAction.headers");

                        List<ByteBuffer> data = requestAction.headers();

                        synchronized (lock) {

                            state = State.HEADERS;

                        }

                        debug.log(Level.DEBUG, "setting outgoing with headers");

                        assert outgoing.isEmpty() : "Unexpected outgoing:" + outgoing;

                        appendToOutgoing(data);

                        cf.complete(null);

                        return cf;

                    } catch (Throwable t) {

                        debug.log(Level.DEBUG, "Failed to send headers: %s", t);

                        connection.close();

                        cf.completeExceptionally(t);

                        return cf;

                    } })

                .thenCompose(unused -> headersSentCF);

    @Override

    CompletableFuture<ExchangeImpl<T>> sendBodyAsync() {

        assert headersSentCF.isDone();

        try {

            bodySubscriber = requestAction.continueRequest();

            if (bodySubscriber == null) {

                bodySubscriber = Http1BodySubscriber.completeSubscriber();

                appendToOutgoing(Http1BodySubscriber.COMPLETED);

            } else {

                bodySubscriber.request(1);  // start

            }

        } catch (Throwable t) {

            connection.close();

            bodySentCF.completeExceptionally(t);

        }

        return bodySentCF;

    @Override

    CompletableFuture<Response> getResponseAsync(Executor executor) {

        CompletableFuture<Response> cf = response.readHeadersAsync(executor);

        Throwable cause;

        synchronized (lock) {

            operations.add(cf);

            cause = failed;

            failed = null;

        }

        if (cause != null) {

            Log.logTrace("Http1Exchange: request [{0}/timeout={1}ms]"

                            + "\n\tCompleting exceptionally with {2}\n",

                         request.uri(),

                         request.timeout().isPresent() ?

                            // calling duration.toMillis() can throw an exception.

                            // this is just debugging, we don't care if it overflows.

                            (request.timeout().get().getSeconds() * 1000

                             + request.timeout().get().getNano() / 1000000) : -1,

                         cause);

            boolean acknowledged = cf.completeExceptionally(cause);

            debug.log(Level.DEBUG,

                      () -> acknowledged

                            ? ("completed response with " + cause)

                            : ("response already completed, ignoring " + cause));

        }

        return cf;

        BodySubscriber<T> bs = handler.apply(response.responseCode(),

                                             response.responseHeaders());

        CompletableFuture<T> bodyCF = response.readBody(bs,

    CompletableFuture<Void> ignoreBody() {

        return response.ignoreBody(executor);

    }

    ByteBuffer drainLeftOverBytes() {

        synchronized (lock) {

            asyncReceiver.stop();

            return asyncReceiver.drain(Utils.EMPTY_BYTEBUFFER);

    void released() {

        Http1Response<T> resp = this.response;

        if (resp != null) resp.completed();

        asyncReceiver.clear();

    void completed() {

        Http1Response<T> resp = this.response;

        if (resp != null) resp.completed();

        cancelImpl(new IOException("Request cancelled"));

    void cancel(IOException cause) {

        cancelImpl(cause);

    }

    private void cancelImpl(Throwable cause) {

        LinkedList<CompletableFuture<?>> toComplete = null;

        synchronized (lock) {

            if (failed == null)

                failed = cause;

            if (requestAction != null && requestAction.finished()

                    && response != null && response.finished()) {

                return;

            }

            connection.close();   // TODO: ensure non-blocking if holding the lock

            writePublisher.writeScheduler.stop();

            if (operations.isEmpty()) {

                Log.logTrace("Http1Exchange: request [{0}/timeout={1}ms] no pending operation."

                                + "\n\tCan''t cancel yet with {2}",

                             request.uri(),

                             request.timeout().isPresent() ?

                                // calling duration.toMillis() can throw an exception.

                                // this is just debugging, we don't care if it overflows.

                                (request.timeout().get().getSeconds() * 1000

                                 + request.timeout().get().getNano() / 1000000) : -1,

                             cause);

            } else {

                for (CompletableFuture<?> cf : operations) {

                    if (!cf.isDone()) {

                        if (toComplete == null) toComplete = new LinkedList<>();

                        toComplete.add(cf);

                        count++;

                    }

                }

                operations.clear();

        if (toComplete != null) {

            // We might be in the selector thread in case of timeout, when

            // the SelectorManager calls purgeTimeoutsAndReturnNextDeadline()

            // There may or may not be other places that reach here

            // from the SelectorManager thread, so just make sure we

            // don't complete any CF from within the selector manager

            // thread.

            Executor exec = client.isSelectorThread()

                            ? executor

                            : this::runInline;

            while (!toComplete.isEmpty()) {

                CompletableFuture<?> cf = toComplete.poll();

                exec.execute(() -> {

                    if (cf.completeExceptionally(cause)) {

                        debug.log(Level.DEBUG, "completed cf with %s",

                                 (Object) cause);

                    }

                });

            }

        }

    }

    private void runInline(Runnable run) {

        assert !client.isSelectorThread();

        run.run();

    /** Returns true if this exchange was canceled. */

    boolean isCanceled() {

        synchronized (lock) {

            return failed != null;

        }

    }

    /** Returns the cause for which this exchange was canceled, if available. */

    Throwable getCancelCause() {

        synchronized (lock) {

            return failed;

        }

    }

    /** Convenience for {@link #appendToOutgoing(DataPair)}, with just a Throwable. */

    void appendToOutgoing(Throwable throwable) {

        appendToOutgoing(new DataPair(null, throwable));

    }

    /** Convenience for {@link #appendToOutgoing(DataPair)}, with just data. */

    void appendToOutgoing(List<ByteBuffer> item) {

        appendToOutgoing(new DataPair(item, null));

    }

    private void appendToOutgoing(DataPair dp) {

        debug.log(Level.DEBUG, "appending to outgoing " + dp);

        outgoing.add(dp);

        writePublisher.writeScheduler.runOrSchedule();

    }

    /** Tells whether, or not, there is any outgoing data that can be published,

     * or if there is an error. */

    private boolean hasOutgoing() {

        return !outgoing.isEmpty();

    // Invoked only by the publisher

    // ALL tasks should execute off the Selector-Manager thread

    /** Returns the next portion of the HTTP request, or the error. */

    private DataPair getOutgoing() {

        final Executor exec = client.theExecutor();

        final DataPair dp = outgoing.pollFirst();

        if (dp == null)  // publisher has not published anything yet

            return null;

        synchronized (lock) {

            if (dp.throwable != null) {

                state = State.ERROR;

                exec.execute(() -> {

                    connection.close();

                    headersSentCF.completeExceptionally(dp.throwable);

                    bodySentCF.completeExceptionally(dp.throwable);

                });

                return dp;

            }

            switch (state) {

                case HEADERS:

                    state = State.BODY;

                    // completeAsync, since dependent tasks should run in another thread

                    debug.log(Level.DEBUG, "initiating completion of headersSentCF");

                    headersSentCF.completeAsync(() -> this, exec);

                    break;

                case BODY:

                    if (dp.data == Http1BodySubscriber.COMPLETED) {

                        state = State.COMPLETING;

                        debug.log(Level.DEBUG, "initiating completion of bodySentCF");

                        bodySentCF.completeAsync(() -> this, exec);