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

package jdk.internal.net.http;

import java.util.Objects;

import java.net.http.HttpRequest;

import java.net.http.HttpResponse;

import java.net.http.HttpResponse.BodyHandler;

import java.net.http.HttpResponse.PushPromiseHandler;

import java.util.concurrent.Executor;

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

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

class PushGroup<T> {

    private final HttpRequest initiatingRequest;

    // user's subscriber object

    final PushPromiseHandler<T> pushPromiseHandler;

    private final Executor executor;

    PushGroup(PushPromiseHandler<T> pushPromiseHandler,

              HttpRequestImpl initiatingRequest,

              Executor executor) {

        this(pushPromiseHandler, initiatingRequest, new MinimalFuture<>(), executor);

    private PushGroup(HttpResponse.PushPromiseHandler<T> pushPromiseHandler,

                      HttpRequestImpl initiatingRequest,

                      Executor executor) {

        this.noMorePushesCF = new MinimalFuture<>();

        this.pushPromiseHandler = pushPromiseHandler;

        this.initiatingRequest = initiatingRequest;

        this.executor = executor;

    interface Acceptor<T> {

        BodyHandler<T> bodyHandler();

        CompletableFuture<HttpResponse<T>> cf();

        boolean accepted();

    private static class AcceptorImpl<T> implements Acceptor<T> {

        private final Executor executor;

        private volatile HttpResponse.BodyHandler<T> bodyHandler;

        private volatile CompletableFuture<HttpResponse<T>> cf;

        AcceptorImpl(Executor executor) {

            this.executor = executor;

        }

        CompletableFuture<HttpResponse<T>> accept(BodyHandler<T> bodyHandler) {

            Objects.requireNonNull(bodyHandler);

            if (this.bodyHandler != null)

                throw new IllegalStateException("non-null bodyHandler");

            this.bodyHandler = bodyHandler;

            cf = new MinimalFuture<>();

            return cf.whenCompleteAsync((r,t) -> {}, executor);

        }

        @Override public BodyHandler<T> bodyHandler() { return bodyHandler; }

        @Override public CompletableFuture<HttpResponse<T>> cf() { return cf; }

        @Override public boolean accepted() { return cf != null; }

    Acceptor<T> acceptPushRequest(HttpRequest pushRequest) {

        AcceptorImpl<T> acceptor = new AcceptorImpl<>(executor);

        try {

            pushPromiseHandler.applyPushPromise(initiatingRequest, pushRequest, acceptor::accept);

        } catch (Throwable t) {

            if (acceptor.accepted()) {

                CompletableFuture<?> cf = acceptor.cf();

                cf.completeExceptionally(t);

            }

            throw t;

        synchronized (this) {

            if (acceptor.accepted()) {

                numberOfPushes++;

                remainingPushes++;

            }

            return acceptor;

        }

    synchronized CompletableFuture<Void> pushesCF() {