/*

 * Copyright (c) 2016, 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.security.AccessControlContext;

import java.util.Optional;

import java.util.concurrent.CompletableFuture;

import jdk.incubator.http.HttpResponse.BodyHandler;

import jdk.incubator.http.HttpResponse.UntrustedBodyHandler;

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

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

/**

 * One PushGroup object is associated with the parent Stream of the pushed

 * Streams. This keeps track of all common state associated with the pushes.

 */

class PushGroup<U,T> {

    // the overall completion object, completed when all pushes are done.

    final CompletableFuture<Void> resultCF;

    final CompletableFuture<Void> noMorePushesCF;

    volatile Throwable error; // any exception that occurred during pushes

    // CF for main response

    final CompletableFuture<HttpResponse<T>> mainResponse;

    // user's subscriber object

    final HttpResponse.MultiSubscriber<U, T> multiSubscriber;

    final HttpResponse.BodyHandler<T> mainBodyHandler;

    private final AccessControlContext acc;

    int numberOfPushes;

    int remainingPushes;

    boolean noMorePushes = false;

    PushGroup(HttpResponse.MultiSubscriber<U, T> multiSubscriber,

              HttpRequestImpl req,

              AccessControlContext acc) {

        this(multiSubscriber, req, new MinimalFuture<>(), acc);

    }

    // Check mainBodyHandler before calling nested constructor.

    private PushGroup(HttpResponse.MultiSubscriber<U, T> multiSubscriber,

                      HttpRequestImpl req,

                      CompletableFuture<HttpResponse<T>> mainResponse,

                      AccessControlContext acc) {

        this(multiSubscriber,

             mainResponse,

             multiSubscriber.onRequest(req),

             acc);

    }

    // This private constructor is called after all parameters have been checked.

    private PushGroup(HttpResponse.MultiSubscriber<U, T> multiSubscriber,

                      CompletableFuture<HttpResponse<T>> mainResponse,

                      HttpResponse.BodyHandler<T> mainBodyHandler,

                      AccessControlContext acc) {

        assert mainResponse != null; // A new instance is created above

        assert mainBodyHandler != null; // should have been checked above

        this.resultCF = new MinimalFuture<>();

        this.noMorePushesCF = new MinimalFuture<>();

        this.multiSubscriber = multiSubscriber;

        this.mainResponse = mainResponse.thenApply(r -> {

            multiSubscriber.onResponse(r);

            return r;

        });

        this.mainBodyHandler = mainBodyHandler;

        if (acc != null) {

            // Restricts the file publisher with the senders ACC, if any

            if (mainBodyHandler instanceof UntrustedBodyHandler)

                ((UntrustedBodyHandler)this.mainBodyHandler).setAccessControlContext(acc);

        }

        this.acc = acc;

    }

    CompletableFuture<Void> groupResult() {

        return resultCF;

    }

    HttpResponse.MultiSubscriber<U, T> subscriber() {

        return multiSubscriber;

    }

    Optional<BodyHandler<T>> handlerForPushRequest(HttpRequest ppRequest) {

        Optional<BodyHandler<T>> bh = multiSubscriber.onPushPromise(ppRequest);

        if (acc != null && bh.isPresent()) {

            // Restricts the file publisher with the senders ACC, if any

            BodyHandler<T> x = bh.get();

            if (x instanceof UntrustedBodyHandler)

                ((UntrustedBodyHandler)x).setAccessControlContext(acc);

            bh = Optional.of(x);

        }

        return bh;

    }

    HttpResponse.BodyHandler<T> mainResponseHandler() {

        return mainBodyHandler;

    }

    synchronized void setMainResponse(CompletableFuture<HttpResponse<T>> r) {

        r.whenComplete((HttpResponse<T> response, Throwable t) -> {

            if (t != null)

                mainResponse.completeExceptionally(t);

            else

                mainResponse.complete(response);

        });

    }

    synchronized void addPush() {

        numberOfPushes++;

        remainingPushes++;

    }

    // This is called when the main body response completes because it means

    // no more PUSH_PROMISEs are possible

    synchronized void noMorePushes(boolean noMore) {

        noMorePushes = noMore;

        checkIfCompleted();

        noMorePushesCF.complete(null);

    }

    CompletableFuture<Void> pushesCF() {

        return noMorePushesCF;

    }

    synchronized boolean noMorePushes() {

        return noMorePushes;

    }

    synchronized void pushCompleted() {

        remainingPushes--;

        checkIfCompleted();

    }

    synchronized void checkIfCompleted() {

        if (Log.trace()) {

            Log.logTrace("PushGroup remainingPushes={0} error={1} noMorePushes={2}",

                         remainingPushes,

                         (error==null)?error:error.getClass().getSimpleName(),

                         noMorePushes);

        }

        if (remainingPushes == 0 && error == null && noMorePushes) {

            if (Log.trace()) {

                Log.logTrace("push completed");

            }

            resultCF.complete(null);

        }

    }

    synchronized void pushError(Throwable t) {

        if (t == null) {

            return;

        }

        this.error = t;

        resultCF.completeExceptionally(t);

    }

}