/*

 * Copyright (c) 2015, 2017, Oracle and/or its affiliates. All rights reserved.

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.  Oracle designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Oracle in the LICENSE file that accompanied this code.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

package jdk.incubator.http;

import java.io.IOException;

import java.time.Duration;

import java.util.List;

import java.security.AccessControlContext;

import java.security.AccessController;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.CompletionException;

import java.util.concurrent.ExecutionException;

import java.util.function.BiFunction;

import java.util.concurrent.Executor;

import java.util.function.UnaryOperator;

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

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

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

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

import static jdk.incubator.http.internal.common.Pair.pair;

/**

 * Encapsulates multiple Exchanges belonging to one HttpRequestImpl.

 * - manages filters

 * - retries due to filters.

 * - I/O errors and most other exceptions get returned directly to user

 *

 * Creates a new Exchange for each request/response interaction

 */

class MultiExchange<U,T> {

    private final HttpRequest userRequest; // the user request

    private final HttpRequestImpl request; // a copy of the user request

    final AccessControlContext acc;

    final HttpClientImpl client;

    final HttpResponse.BodyHandler<T> responseHandler;

    final ExecutorWrapper execWrapper;

    final Executor executor;

    final HttpResponse.MultiProcessor<U,T> multiResponseHandler;

    HttpRequestImpl currentreq; // used for async only

    Exchange<T> exchange; // the current exchange

    Exchange<T> previous;

    int attempts;

    // Maximum number of times a request will be retried/redirected

    // for any reason

    static final int DEFAULT_MAX_ATTEMPTS = 5;

    static final int max_attempts = Utils.getIntegerNetProperty(

            "jdk.httpclient.redirects.retrylimit", DEFAULT_MAX_ATTEMPTS

    );

    private final List<HeaderFilter> filters;

    TimedEvent timedEvent;

    volatile boolean cancelled;

    final PushGroup<U,T> pushGroup;

    /**

     * Filter fields. These are attached as required by filters

     * and only used by the filter implementations. This could be

     * generalised into Objects that are passed explicitly to the filters

     * (one per MultiExchange object, and one per Exchange object possibly)

     */

    volatile AuthenticationFilter.AuthInfo serverauth, proxyauth;

    // RedirectHandler

    volatile int numberOfRedirects = 0;

    /**

     * MultiExchange with one final response.

     */

    MultiExchange(HttpRequest req,

                  HttpClientImpl client,

                  HttpResponse.BodyHandler<T> responseHandler) {

        this.previous = null;

        this.userRequest = req;

        this.request = new HttpRequestImpl(req);

        this.currentreq = request;

        this.attempts = 0;

        this.client = client;

        this.filters = client.filterChain();

        if (System.getSecurityManager() != null) {

            this.acc = AccessController.getContext();

        } else {

            this.acc = null;

        }

        this.execWrapper = new ExecutorWrapper(client.executor(), acc);

        this.executor = execWrapper.executor();

        this.responseHandler = responseHandler;

        this.exchange = new Exchange<>(request, this);

        this.multiResponseHandler = null;

        this.pushGroup = null;

    }

    /**

     * MultiExchange with multiple responses (HTTP/2 server pushes).

     */

    MultiExchange(HttpRequest req,

                  HttpClientImpl client,

                  HttpResponse.MultiProcessor<U, T> multiResponseHandler) {

        this.previous = null;

        this.userRequest = req;

        this.request = new HttpRequestImpl(req);

        this.currentreq = request;

        this.attempts = 0;

        this.client = client;

        this.filters = client.filterChain();

        if (System.getSecurityManager() != null) {

            this.acc = AccessController.getContext();

        } else {

            this.acc = null;

        }

        this.execWrapper = new ExecutorWrapper(client.executor(), acc);

        this.executor = execWrapper.executor();

        this.multiResponseHandler = multiResponseHandler;

        this.pushGroup = new PushGroup<>(multiResponseHandler, request);

        this.exchange = new Exchange<>(request, this);

        this.responseHandler = pushGroup.mainResponseHandler();

    }

    public HttpResponseImpl<T> response() throws IOException, InterruptedException {

        HttpRequestImpl r = request;

        if (r.duration() != null) {

            timedEvent = new TimedEvent(r.duration());

            client.registerTimer(timedEvent);

        }

        while (attempts < max_attempts) {

            try {

                attempts++;

                Exchange<T> currExchange = getExchange();

                requestFilters(r);

                Response response = currExchange.response();

                HttpRequestImpl newreq = responseFilters(response);

                if (newreq == null) {

                    if (attempts > 1) {

                        Log.logError("Succeeded on attempt: " + attempts);

                    }

                    T body = currExchange.readBody(responseHandler);

                    cancelTimer();

                    return new HttpResponseImpl<>(userRequest, response, body, currExchange);

                }

                //response.body(HttpResponse.ignoreBody());

                setExchange(new Exchange<>(newreq, this, acc));

                r = newreq;

            } catch (IOException e) {

                if (cancelled) {

                    throw new HttpTimeoutException("Request timed out");

                }

                throw e;

            }

        }

        cancelTimer();

        throw new IOException("Retry limit exceeded");

    }

    CompletableFuture<Void> multiCompletionCF() {

        return pushGroup.groupResult();

    }

    private synchronized Exchange<T> getExchange() {

        return exchange;

    }

    HttpClientImpl client() {

        return client;

    }

    HttpClient.Redirect followRedirects() {

        return client.followRedirects();

    }

    HttpClient.Version version() {

        return request.version().orElse(client.version());

    }

    private synchronized void setExchange(Exchange<T> exchange) {

        this.exchange = exchange;

    }

    private void cancelTimer() {

        if (timedEvent != null) {

            client.cancelTimer(timedEvent);

        }

    }

    private void requestFilters(HttpRequestImpl r) throws IOException {

        Log.logTrace("Applying request filters");

        for (HeaderFilter filter : filters) {

            Log.logTrace("Applying {0}", filter);

            filter.request(r, this);

        }

        Log.logTrace("All filters applied");

    }

    private HttpRequestImpl responseFilters(Response response) throws IOException

    {

        Log.logTrace("Applying response filters");

        for (HeaderFilter filter : filters) {

            Log.logTrace("Applying {0}", filter);

            HttpRequestImpl newreq = filter.response(response);

            if (newreq != null) {

                Log.logTrace("New request: stopping filters");

                return newreq;

            }

        }

        Log.logTrace("All filters applied");

        return null;

    }

    public void cancel() {

        cancelled = true;

        getExchange().cancel();

    }

    public void cancel(IOException cause) {

        cancelled = true;

        getExchange().cancel(cause);

    }

    public CompletableFuture<HttpResponseImpl<T>> responseAsync() {

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

        CompletableFuture<HttpResponseImpl<T>> cf = responseAsync0(start);

        start.completeAsync( () -> null, executor); // trigger execution

        return cf;

    }

    private CompletableFuture<HttpResponseImpl<T>> responseAsync0(CompletableFuture<Void> start) {

        return start.thenCompose( v -> responseAsyncImpl())

            .thenCompose((Response r) -> {

                Exchange<T> exch = getExchange();

                return exch.readBodyAsync(responseHandler)

                        .thenApply((T body) ->  new HttpResponseImpl<>(userRequest, r, body, exch));

            });

    }

    CompletableFuture<U> multiResponseAsync() {

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

        CompletableFuture<HttpResponseImpl<T>> cf = responseAsync0(start);

        CompletableFuture<HttpResponse<T>> mainResponse =

                cf.thenApply((HttpResponseImpl<T> b) -> {

                      multiResponseHandler.onResponse(b);

                      return (HttpResponse<T>)b;

                   });

        pushGroup.setMainResponse(mainResponse);

        // set up house-keeping related to multi-response

        mainResponse.thenAccept((r) -> {

            // All push promises received by now.

            pushGroup.noMorePushes(true);

        });

        CompletableFuture<U> res = multiResponseHandler.completion(pushGroup.groupResult(), pushGroup.pushesCF());

        start.completeAsync( () -> null, executor); // trigger execution

        return res;

    }

    private CompletableFuture<Response> responseAsyncImpl() {

        CompletableFuture<Response> cf;

        if (++attempts > max_attempts) {

            cf = MinimalFuture.failedFuture(new IOException("Too many retries"));

        } else {

            if (currentreq.duration() != null) {

                timedEvent = new TimedEvent(currentreq.duration());

                client.registerTimer(timedEvent);

            }

            try {

                // 1. Apply request filters

                requestFilters(currentreq);

            } catch (IOException e) {

                return MinimalFuture.failedFuture(e);

            }

            Exchange<T> exch = getExchange();

            // 2. get response

            cf = exch.responseAsync()

                .thenCompose((Response response) -> {

                    HttpRequestImpl newrequest = null;

                    try {

                        // 3. Apply response filters

                        newrequest = responseFilters(response);

                    } catch (IOException e) {

                        return MinimalFuture.failedFuture(e);

                    }

                    // 4. Check filter result and repeat or continue

                    if (newrequest == null) {

                        if (attempts > 1) {

                            Log.logError("Succeeded on attempt: " + attempts);

                        }

                        return MinimalFuture.completedFuture(response);

                    } else {

                        currentreq = newrequest;

                        setExchange(new Exchange<>(currentreq, this, acc));

                        //reads body off previous, and then waits for next response

                        return responseAsyncImpl();

                    }

                })

            // 5. Handle errors and cancel any timer set

            .handle((response, ex) -> {

                cancelTimer();

                if (ex == null) {

                    assert response != null;

                    return MinimalFuture.completedFuture(response);

                }

                // all exceptions thrown are handled here

                CompletableFuture<Response> error = getExceptionalCF(ex);

                if (error == null) {

                    return responseAsyncImpl();

                } else {

                    return error;

                }

            })

            .thenCompose(UnaryOperator.identity());

        }

        return cf;

    }

    /**

     * Take a Throwable and return a suitable CompletableFuture that is

     * completed exceptionally.

     */

    private CompletableFuture<Response> getExceptionalCF(Throwable t) {

        if ((t instanceof CompletionException) || (t instanceof ExecutionException)) {

            if (t.getCause() != null) {

                t = t.getCause();

            }

        }

        if (cancelled && t instanceof IOException) {

            t = new HttpTimeoutException("request timed out");

        }

        return MinimalFuture.failedFuture(t);

    }

    class TimedEvent extends TimeoutEvent {

        TimedEvent(Duration duration) {

            super(duration);

        }

        @Override

        public void handle() {

            cancel(new HttpTimeoutException("request timed out"));

        }

    }

}