/*
 * 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.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Flow;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

import jdk.incubator.http.internal.common.*;
import jdk.incubator.http.internal.frame.*;
import jdk.incubator.http.internal.hpack.DecodingCallback;
import static java.util.stream.Collectors.toList;

/**
 * Http/2 Stream handling.
 *
 * REQUESTS
 *
 * sendHeadersOnly() -- assembles HEADERS frame and puts on connection outbound Q
 *
 * sendRequest() -- sendHeadersOnly() + sendBody()
 *
 * sendBody() -- in calling thread: obeys all flow control (so may block)
 *               obtains data from request body processor and places on connection
 *               outbound Q.
 *
 * sendBodyAsync() -- calls sendBody() in an executor thread.
 *
 * sendHeadersAsync() -- calls sendHeadersOnly() which does not block
 *
 * sendRequestAsync() -- calls sendRequest() in an executor thread
 *
 * RESPONSES
 *
 * Multiple responses can be received per request. Responses are queued up on
 * a LinkedList of CF<HttpResponse> and the the first one on the list is completed
 * with the next response
 *
 * getResponseAsync() -- queries list of response CFs and returns first one
 *               if one exists. Otherwise, creates one and adds it to list
 *               and returns it. Completion is achieved through the
 *               incoming() upcall from connection reader thread.
 *
 * getResponse() -- calls getResponseAsync() and waits for CF to complete
 *
 * responseBody() -- in calling thread: blocks for incoming DATA frames on
 *               stream inputQ. Obeys remote and local flow control so may block.
 *               Calls user response body processor with data buffers.
 *
 * responseBodyAsync() -- calls responseBody() in an executor thread.
 *
 * incoming() -- entry point called from connection reader thread. Frames are
 *               either handled immediately without blocking or for data frames
 *               placed on the stream's inputQ which is consumed by the stream's
 *               reader thread.
 *
 * PushedStream sub class
 * ======================
 * Sending side methods are not used because the request comes from a PUSH_PROMISE
 * frame sent by the server. When a PUSH_PROMISE is received the PushedStream
 * is created. PushedStream does not use responseCF list as there can be only
 * one response. The CF is created when the object created and when the response
 * HEADERS frame is received the object is completed.
 */
class Stream<T> extends ExchangeImpl<T> {

    final AsyncDataReadQueue inputQ = new AsyncDataReadQueue();

    /**
     * This stream's identifier. Assigned lazily by the HTTP2Connection before
     * the stream's first frame is sent.
     */
    protected volatile int streamid;

    long responseContentLen = -1;
    long responseBytesProcessed = 0;
    long requestContentLen;

    final Http2Connection connection;
    HttpClientImpl client;
    final HttpRequestImpl request;
    final DecodingCallback rspHeadersConsumer;
    HttpHeadersImpl responseHeaders;
    final HttpHeadersImpl requestHeaders;
    final HttpHeadersImpl requestPseudoHeaders;
    HttpResponse.BodyProcessor<T> responseProcessor;
    final HttpRequest.BodyProcessor requestProcessor;
    volatile int responseCode;
    volatile Response response;
    volatile CompletableFuture<Response> responseCF;
    final AbstractPushPublisher<List<ByteBuffer>> publisher;
    final CompletableFuture<Void> requestBodyCF = new MinimalFuture<>();

    /** True if END_STREAM has been seen in a frame received on this stream. */
    private volatile boolean remotelyClosed;
    private volatile boolean closed;
    private volatile boolean endStreamSent;

    // state flags
    boolean requestSent, responseReceived, responseHeadersReceived;

    /**
     * A reference to this Stream's connection Send Window controller. The
     * stream MUST acquire the appropriate amount of Send Window before
     * sending any data. Will be null for PushStreams, as they cannot send data.
     */
    private final WindowController windowController;
    private final WindowUpdateSender windowUpdater;

    @Override
    HttpConnection connection() {
        return connection.connection;
    }

    @Override
    CompletableFuture<T> readBodyAsync(HttpResponse.BodyHandler<T> handler,
                                       boolean returnConnectionToPool,
                                       Executor executor)
    {
        Log.logTrace("Reading body on stream {0}", streamid);
        responseProcessor = handler.apply(responseCode, responseHeaders);
        publisher.subscribe(responseProcessor);
        CompletableFuture<T> cf = receiveData(executor);

        PushGroup<?,?> pg = exchange.getPushGroup();
        if (pg != null) {
            // if an error occurs make sure it is recorded in the PushGroup
            cf = cf.whenComplete((t,e) -> pg.pushError(e));
        }
        return cf;
    }

    @Override
    T readBody(HttpResponse.BodyHandler<T> handler, boolean returnConnectionToPool)
        throws IOException
    {
        CompletableFuture<T> cf = readBodyAsync(handler,
                                                returnConnectionToPool,
                                                null);
        try {
            return cf.join();
        } catch (CompletionException e) {
            throw Utils.getIOException(e);
        }
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("streamid: ")
                .append(streamid);
        return sb.toString();
    }

    private boolean receiveDataFrame(Http2Frame frame) throws IOException, InterruptedException {
        if (frame instanceof ResetFrame) {
            handleReset((ResetFrame) frame);
            return true;
        } else if (!(frame instanceof DataFrame)) {
            assert false;
            return true;
        }
        DataFrame df = (DataFrame) frame;
        // RFC 7540 6.1:
        // The entire DATA frame payload is included in flow control,
        // including the Pad Length and Padding fields if present
        int len = df.payloadLength();
        ByteBufferReference[] buffers = df.getData();
        List<ByteBuffer> dsts = Arrays.stream(buffers)
                .map(ByteBufferReference::get)
                .filter(ByteBuffer::hasRemaining)
                .collect(toList());
        if (!dsts.isEmpty()) {
            publisher.acceptData(Optional.of(dsts));
        }
        connection.windowUpdater.update(len);
        if (df.getFlag(DataFrame.END_STREAM)) {
            setEndStreamReceived();
            publisher.acceptData(Optional.empty());
            return false;
        }
        // Don't send window update on a stream which is
        // closed or half closed.
        windowUpdater.update(len);
        return true;
    }

    // pushes entire response body into response processor
    // blocking when required by local or remote flow control
    CompletableFuture<T> receiveData(Executor executor) {
        CompletableFuture<T> cf = responseProcessor
                .getBody()
                .toCompletableFuture();
        Consumer<Throwable> onError = e -> {
            Log.logTrace("receiveData: {0}", e.toString());
            e.printStackTrace();
            cf.completeExceptionally(e);
            publisher.acceptError(e);
        };
        if (executor == null) {
            inputQ.blockingReceive(this::receiveDataFrame, onError);
        } else {
            inputQ.asyncReceive(executor, this::receiveDataFrame, onError);
        }
        return cf;
    }

    @Override
    void sendBody() throws IOException {
        try {
            sendBodyImpl().join();
        } catch (CompletionException e) {
            throw Utils.getIOException(e);
        }
    }

    CompletableFuture<ExchangeImpl<T>> sendBodyAsync() {
        return sendBodyImpl().thenApply( v -> this);
    }

    @SuppressWarnings("unchecked")
    Stream(HttpClientImpl client,
           Http2Connection connection,
           Exchange<T> e,
           WindowController windowController)
    {
        super(e);
        this.client = client;
        this.connection = connection;
        this.windowController = windowController;
        this.request = e.request();
        this.requestProcessor = request.requestProcessor;
        responseHeaders = new HttpHeadersImpl();
        requestHeaders = new HttpHeadersImpl();
        rspHeadersConsumer = (name, value) -> {
            responseHeaders.addHeader(name.toString(), value.toString());
            if (Log.headers() && Log.trace()) {
                Log.logTrace("RECEIVED HEADER (streamid={0}): {1}: {2}",
                             streamid, name, value);
            }
        };
        this.requestPseudoHeaders = new HttpHeadersImpl();
        // NEW
        this.publisher = new BlockingPushPublisher<>();
        this.windowUpdater = new StreamWindowUpdateSender(connection);
    }

    /**
     * Entry point from Http2Connection reader thread.
     *
     * Data frames will be removed by response body thread.
     */
    void incoming(Http2Frame frame) throws IOException {
        if ((frame instanceof HeaderFrame)) {
            HeaderFrame hframe = (HeaderFrame)frame;
            if (hframe.endHeaders()) {
                Log.logTrace("handling response (streamid={0})", streamid);
                handleResponse();
                if (hframe.getFlag(HeaderFrame.END_STREAM)) {
                    inputQ.put(new DataFrame(streamid, DataFrame.END_STREAM, new ByteBufferReference[0]));
                }
            }
        } else if (frame instanceof DataFrame) {
            inputQ.put(frame);
        } else {
            otherFrame(frame);
        }
    }

    void otherFrame(Http2Frame frame) throws IOException {
        switch (frame.type()) {
            case WindowUpdateFrame.TYPE:
                incoming_windowUpdate((WindowUpdateFrame) frame);
                break;
            case ResetFrame.TYPE:
                incoming_reset((ResetFrame) frame);
                break;
            case PriorityFrame.TYPE:
                incoming_priority((PriorityFrame) frame);
                break;
            default:
                String msg = "Unexpected frame: " + frame.toString();
                throw new IOException(msg);
        }
    }

    // The Hpack decoder decodes into one of these consumers of name,value pairs

    DecodingCallback rspHeadersConsumer() {
        return rspHeadersConsumer;
    }

    protected void handleResponse() throws IOException {
        synchronized(this) {
            responseHeadersReceived = true;
        }
        HttpConnection c = connection.connection; // TODO: improve
        responseCode = (int)responseHeaders
                .firstValueAsLong(":status")
                .orElseThrow(() -> new IOException("no statuscode in response"));

        response = new Response(
                request, exchange, responseHeaders,
                responseCode, HttpClient.Version.HTTP_2);

        this.responseContentLen = responseHeaders
                .firstValueAsLong("content-length")
                .orElse(-1L);

        if (Log.headers()) {
            StringBuilder sb = new StringBuilder("RESPONSE HEADERS:\n");
            Log.dumpHeaders(sb, "    ", responseHeaders);
            Log.logHeaders(sb.toString());
        }

        completeResponse(response);
    }

    void incoming_reset(ResetFrame frame) throws IOException {
        Log.logTrace("Received RST_STREAM on stream {0}", streamid);
        if (endStreamReceived()) {
            Log.logTrace("Ignoring RST_STREAM frame received on remotely closed stream {0}", streamid);
        } else if (closed) {
            Log.logTrace("Ignoring RST_STREAM frame received on closed stream {0}", streamid);
        } else {
            boolean pushedToQueue = false;
            synchronized(this) {
                // if the response headers are not yet
                // received, or the inputQueue is closed, handle reset directly.
                // Otherwise, put it in the input queue in order to read all
                // pending data frames first. Indeed, a server may send
                // RST_STREAM after sending END_STREAM, in which case we should
                // ignore it. However, we won't know if we have received END_STREAM
                // or not until all pending data frames are read.
                // Because the inputQ will not be read until the response
                // headers are received, and because response headers won't be
                // sent if the server sent RST_STREAM, then we must handle
                // reset here directly unless responseHeadersReceived is true.
                pushedToQueue = !closed && responseHeadersReceived && inputQ.tryPut(frame);
            }
            if (!pushedToQueue) {
                // RST_STREAM was not pushed to the queue: handle it.
                try {
                    handleReset(frame);
                } catch (IOException io) {
                    completeResponseExceptionally(io);
                }
            } else {
                // RST_STREAM was pushed to the queue. It will be handled by
                // asyncReceive after all pending data frames have been
                // processed.
                Log.logTrace("RST_STREAM pushed in queue for stream {0}", streamid);
            }
        }
    }

    void handleReset(ResetFrame frame) throws IOException {
        Log.logTrace("Handling RST_STREAM on stream {0}", streamid);
        if (!closed) {
            close();
            int error = frame.getErrorCode();
            throw new IOException(ErrorFrame.stringForCode(error));
        } else {
            Log.logTrace("Ignoring RST_STREAM frame received on closed stream {0}", streamid);
        }
    }

    void incoming_priority(PriorityFrame frame) {
        // TODO: implement priority
        throw new UnsupportedOperationException("Not implemented");
    }

    private void incoming_windowUpdate(WindowUpdateFrame frame)
        throws IOException
    {
        int amount = frame.getUpdate();
        if (amount <= 0) {
            Log.logTrace("Resetting stream: {0} %d, Window Update amount: %d\n",
                         streamid, streamid, amount);
            connection.resetStream(streamid, ResetFrame.FLOW_CONTROL_ERROR);
        } else {
            assert streamid != 0;
            boolean success = windowController.increaseStreamWindow(amount, streamid);
            if (!success) {  // overflow
                connection.resetStream(streamid, ResetFrame.FLOW_CONTROL_ERROR);
            }
        }
    }

    void incoming_pushPromise(HttpRequestImpl pushReq,
                              PushedStream<?,T> pushStream)
        throws IOException
    {
        if (Log.requests()) {
            Log.logRequest("PUSH_PROMISE: " + pushReq.toString());
        }
        PushGroup<?,T> pushGroup = exchange.getPushGroup();
        if (pushGroup == null || pushGroup.noMorePushes()) {
            cancelImpl(new IllegalStateException("unexpected push promise"
                + " on stream " + streamid));
        }

        HttpResponse.MultiProcessor<?,T> proc = pushGroup.processor();

        CompletableFuture<HttpResponse<T>> cf = pushStream.responseCF();

        Optional<HttpResponse.BodyHandler<T>> bpOpt = proc.onRequest(
                pushReq);

        if (!bpOpt.isPresent()) {
            IOException ex = new IOException("Stream "
                 + streamid + " cancelled by user");
            if (Log.trace()) {
                Log.logTrace("No body processor for {0}: {1}", pushReq,
                            ex.getMessage());
            }
            pushStream.cancelImpl(ex);
            cf.completeExceptionally(ex);
            return;
        }

        pushGroup.addPush();
        pushStream.requestSent();
        pushStream.setPushHandler(bpOpt.get());
        // setup housekeeping for when the push is received
        // TODO: deal with ignoring of CF anti-pattern
        cf.whenComplete((HttpResponse<T> resp, Throwable t) -> {
            if (Log.trace()) {
                Log.logTrace("Push completed on stream {0} for {1}{2}",
                             pushStream.streamid, resp,
                             ((t==null) ? "": " with exception " + t));
            }
            if (t != null) {
                pushGroup.pushError(t);
                proc.onError(pushReq, t);
            } else {
                proc.onResponse(resp);
            }
            pushGroup.pushCompleted();
        });

    }

    private OutgoingHeaders<Stream<T>> headerFrame(long contentLength) {
        HttpHeadersImpl h = request.getSystemHeaders();
        if (contentLength > 0) {
            h.setHeader("content-length", Long.toString(contentLength));
        }
        setPseudoHeaderFields();
        OutgoingHeaders<Stream<T>> f = new OutgoingHeaders<>(h, request.getUserHeaders(), this);
        if (contentLength == 0) {
            f.setFlag(HeadersFrame.END_STREAM);
            endStreamSent = true;
        }
        return f;
    }

    private void setPseudoHeaderFields() {
        HttpHeadersImpl hdrs = requestPseudoHeaders;
        String method = request.method();
        hdrs.setHeader(":method", method);
        URI uri = request.uri();
        hdrs.setHeader(":scheme", uri.getScheme());
        // TODO: userinfo deprecated. Needs to be removed
        hdrs.setHeader(":authority", uri.getAuthority());
        // TODO: ensure header names beginning with : not in user headers
        String query = uri.getQuery();
        String path = uri.getPath();
        if (path == null || path.isEmpty()) {
            if (method.equalsIgnoreCase("OPTIONS")) {
                path = "*";
            } else {
                path = "/";
            }
        }
        if (query != null) {
            path += "?" + query;
        }
        hdrs.setHeader(":path", path);
    }

    HttpHeadersImpl getRequestPseudoHeaders() {
        return requestPseudoHeaders;
    }

    @Override
    Response getResponse() throws IOException {
        try {
            if (request.duration() != null) {
                Log.logTrace("Waiting for response (streamid={0}, timeout={1}ms)",
                             streamid,
                             request.duration().toMillis());
                return getResponseAsync(null).get(
                        request.duration().toMillis(), TimeUnit.MILLISECONDS);
            } else {
                Log.logTrace("Waiting for response (streamid={0})", streamid);
                return getResponseAsync(null).join();
            }
        } catch (TimeoutException e) {
            Log.logTrace("Response timeout (streamid={0})", streamid);
            throw new HttpTimeoutException("Response timed out");
        } catch (InterruptedException | ExecutionException | CompletionException e) {
            Throwable t = e.getCause();
            Log.logTrace("Response failed (streamid={0}): {1}", streamid, t);
            if (t instanceof IOException) {
                throw (IOException)t;
            }
            throw new IOException(e);
        } finally {
            Log.logTrace("Got response or failed (streamid={0})", streamid);
        }
    }

    /** Sets endStreamReceived. Should be called only once. */
    void setEndStreamReceived() {
        assert remotelyClosed == false: "Unexpected endStream already set";
        remotelyClosed = true;
        responseReceived();
    }

    /** Tells whether, or not, the END_STREAM Flag has been seen in any frame
     *  received on this stream. */
    private boolean endStreamReceived() {
        return remotelyClosed;
    }

    @Override
    void sendHeadersOnly() throws IOException, InterruptedException {
        if (Log.requests() && request != null) {
            Log.logRequest(request.toString());
        }
        requestContentLen = requestProcessor.contentLength();
        OutgoingHeaders<Stream<T>> f = headerFrame(requestContentLen);
        connection.sendFrame(f);
    }

    void registerStream(int id) {
        this.streamid = id;
        connection.putStream(this, streamid);
    }

    class RequestSubscriber implements Flow.Subscriber<ByteBuffer> {
        // can be < 0 if the actual length is not known.
        private volatile long remainingContentLength;
        private volatile Subscription subscription;

        RequestSubscriber(long contentLen) {
            this.remainingContentLength = contentLen;
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            if (this.subscription != null) {
                throw new IllegalStateException();
            }
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override
        public void onNext(ByteBuffer item) {
            if (requestBodyCF.isDone()) {
                throw new IllegalStateException();
            }

            try {
                while (item.hasRemaining()) {
                    assert !endStreamSent : "internal error, send data after END_STREAM flag";
                    DataFrame df = getDataFrame(item);
                    if (remainingContentLength > 0) {
                        remainingContentLength -= df.getDataLength();
                        assert remainingContentLength >= 0;
                        if (remainingContentLength == 0) {
                            df.setFlag(DataFrame.END_STREAM);
                            endStreamSent = true;
                        }
                    }
                    connection.sendDataFrame(df);
                }
                subscription.request(1);
            } catch (InterruptedException ex) {
                subscription.cancel();
                requestBodyCF.completeExceptionally(ex);
            }
        }

        @Override
        public void onError(Throwable throwable) {
            if (requestBodyCF.isDone()) {
                return;
            }
            subscription.cancel();
            requestBodyCF.completeExceptionally(throwable);
        }

        @Override
        public void onComplete() {
            assert endStreamSent || remainingContentLength < 0;
            try {
                if (!endStreamSent) {
                    endStreamSent = true;
                    connection.sendDataFrame(getEmptyEndStreamDataFrame());
                }
                requestBodyCF.complete(null);
            } catch (InterruptedException ex) {
                requestBodyCF.completeExceptionally(ex);
            }
        }
    }

    DataFrame getDataFrame(ByteBuffer buffer) throws InterruptedException {
        int requestAmount = Math.min(connection.getMaxSendFrameSize(), buffer.remaining());
        // blocks waiting for stream send window, if exhausted
        int actualAmount = windowController.tryAcquire(requestAmount, streamid);
        ByteBuffer outBuf = Utils.slice(buffer,  actualAmount);
        DataFrame df = new DataFrame(streamid, 0 , ByteBufferReference.of(outBuf));
        return df;
    }

    private DataFrame getEmptyEndStreamDataFrame() throws InterruptedException {
        return new DataFrame(streamid, DataFrame.END_STREAM, new ByteBufferReference[0]);
    }

    /**
     * A List of responses relating to this stream. Normally there is only
     * one response, but intermediate responses like 100 are allowed
     * and must be passed up to higher level before continuing. Deals with races
     * such as if responses are returned before the CFs get created by
     * getResponseAsync()
     */

    final List<CompletableFuture<Response>> response_cfs = new ArrayList<>(5);

    @Override
    CompletableFuture<Response> getResponseAsync(Executor executor) {
        CompletableFuture<Response> cf = null;
        // The code below deals with race condition that can be caused when
        // completeResponse() is being called before getResponseAsync()
        synchronized (response_cfs) {
            if (!response_cfs.isEmpty()) {
                // This CompletableFuture was created by completeResponse().
                // it will be already completed.
                cf = response_cfs.remove(0);
                // if we find a cf here it should be already completed.
                // finding a non completed cf should not happen. just assert it.
                assert cf.isDone() : "Removing uncompleted response: could cause code to hang!";
            } else {
                // getResponseAsync() is called first. Create a CompletableFuture
                // that will be completed by completeResponse() when
                // completeResponse() is called.
                cf = new MinimalFuture<>();
                response_cfs.add(cf);
            }
        }
        if (executor != null && !cf.isDone()) {
            // protect from executing later chain of CompletableFuture operations from SelectorManager thread
            cf = cf.thenApplyAsync(r -> r, executor);
        }
        Log.logTrace("Response future (stream={0}) is: {1}", streamid, cf);
        PushGroup<?,?> pg = exchange.getPushGroup();
        if (pg != null) {
            // if an error occurs make sure it is recorded in the PushGroup
            cf = cf.whenComplete((t,e) -> pg.pushError(e));
        }
        return cf;
    }

    /**
     * Completes the first uncompleted CF on list, and removes it. If there is no
     * uncompleted CF then creates one (completes it) and adds to list
     */
    void completeResponse(Response resp) {
        synchronized (response_cfs) {
            CompletableFuture<Response> cf;
            int cfs_len = response_cfs.size();
            for (int i=0; i<cfs_len; i++) {
                cf = response_cfs.get(i);
                if (!cf.isDone()) {
                    Log.logTrace("Completing response (streamid={0}): {1}",
                                 streamid, cf);
                    cf.complete(resp);
                    response_cfs.remove(cf);
                    return;
                } // else we found the previous response: just leave it alone.
            }
            cf = MinimalFuture.completedFuture(resp);
            Log.logTrace("Created completed future (streamid={0}): {1}",
                         streamid, cf);
            response_cfs.add(cf);
        }
    }

    // methods to update state and remove stream when finished

    synchronized void requestSent() {
        requestSent = true;
        if (responseReceived) {
            close();
        }
    }

    final synchronized boolean isResponseReceived() {
        return responseReceived;
    }

    synchronized void responseReceived() {
        responseReceived = true;
        if (requestSent) {
            close();
        }
    }

    /**
     * same as above but for errors
     */
    void completeResponseExceptionally(Throwable t) {
        synchronized (response_cfs) {
            // use index to avoid ConcurrentModificationException
            // caused by removing the CF from within the loop.
            for (int i = 0; i < response_cfs.size(); i++) {
                CompletableFuture<Response> cf = response_cfs.get(i);
                if (!cf.isDone()) {
                    cf.completeExceptionally(t);
                    response_cfs.remove(i);
                    return;
                }
            }
            response_cfs.add(MinimalFuture.failedFuture(t));
        }
    }

    CompletableFuture<Void> sendBodyImpl() {
        RequestSubscriber subscriber = new RequestSubscriber(requestContentLen);
        requestProcessor.subscribe(subscriber);
        requestBodyCF.whenComplete((v,t) -> requestSent());
        return requestBodyCF;
    }

    @Override
    void cancel() {
        cancel(new IOException("Stream " + streamid + " cancelled"));
    }

    @Override
    void cancel(IOException cause) {
        cancelImpl(cause);
    }

    // This method sends a RST_STREAM frame
    void cancelImpl(Throwable e) {
        if (Log.trace()) {
            Log.logTrace("cancelling stream {0}: {1}\n", streamid, e);
        }
        boolean closing;
        if (closing = !closed) { // assigning closing to !closed
            synchronized (this) {
                if (closing = !closed) { // assigning closing to !closed
                    closed=true;
                }
            }
        }
        if (closing) { // true if the stream has not been closed yet
            inputQ.close();
        }
        completeResponseExceptionally(e);
        try {
            // will send a RST_STREAM frame
            if (streamid != 0) {
                connection.resetStream(streamid, ResetFrame.CANCEL);
            }
        } catch (IOException ex) {
            Log.logError(ex);
        }
    }

    // This method doesn't send any frame
    void close() {
        if (closed) return;
        synchronized(this) {
            if (closed) return;
            closed = true;
        }
        Log.logTrace("Closing stream {0}", streamid);
        inputQ.close();
        connection.closeStream(streamid);
        Log.logTrace("Stream {0} closed", streamid);
    }

    static class PushedStream<U,T> extends Stream<T> {
        final PushGroup<U,T> pushGroup;
        private final Stream<T> parent;      // used by server push streams
        // push streams need the response CF allocated up front as it is
        // given directly to user via the multi handler callback function.
        final CompletableFuture<Response> pushCF;
        final CompletableFuture<HttpResponse<T>> responseCF;
        final HttpRequestImpl pushReq;
        HttpResponse.BodyHandler<T> pushHandler;

        PushedStream(PushGroup<U,T> pushGroup, HttpClientImpl client,
                Http2Connection connection, Stream<T> parent,
                Exchange<T> pushReq) {
            // ## no request body possible, null window controller
            super(client, connection, pushReq, null);
            this.pushGroup = pushGroup;
            this.pushReq = pushReq.request();
            this.pushCF = new MinimalFuture<>();
            this.responseCF = new MinimalFuture<>();
            this.parent = parent;
        }

        CompletableFuture<HttpResponse<T>> responseCF() {
            return responseCF;
        }

        synchronized void setPushHandler(HttpResponse.BodyHandler<T> pushHandler) {
            this.pushHandler = pushHandler;
        }

        synchronized HttpResponse.BodyHandler<T> getPushHandler() {
            // ignored parameters to function can be used as BodyHandler
            return this.pushHandler;
        }

        // Following methods call the super class but in case of
        // error record it in the PushGroup. The error method is called
        // with a null value when no error occurred (is a no-op)
        @Override
        CompletableFuture<ExchangeImpl<T>> sendBodyAsync() {
            return super.sendBodyAsync()
                        .whenComplete((ExchangeImpl<T> v, Throwable t) -> pushGroup.pushError(t));
        }

        @Override
        CompletableFuture<ExchangeImpl<T>> sendHeadersAsync() {
            return super.sendHeadersAsync()
                        .whenComplete((ExchangeImpl<T> ex, Throwable t) -> pushGroup.pushError(t));
        }

        @Override
        CompletableFuture<Response> getResponseAsync(Executor executor) {
            CompletableFuture<Response> cf = pushCF.whenComplete((v, t) -> pushGroup.pushError(t));
            if(executor!=null && !cf.isDone()) {
                cf  = cf.thenApplyAsync( r -> r, executor);
            }
            return cf;
        }

        @Override
        CompletableFuture<T> readBodyAsync(
                HttpResponse.BodyHandler<T> handler,
                boolean returnConnectionToPool,
                Executor executor)
        {
            return super.readBodyAsync(handler, returnConnectionToPool, executor)
                        .whenComplete((v, t) -> pushGroup.pushError(t));
        }

        @Override
        void completeResponse(Response r) {
            HttpResponseImpl.logResponse(r);
            pushCF.complete(r); // not strictly required for push API
            // start reading the body using the obtained BodyProcessor
            CompletableFuture<Void> start = new MinimalFuture<>();
            start.thenCompose( v -> readBodyAsync(getPushHandler(), false, getExchange().executor()))
                .whenComplete((T body, Throwable t) -> {
                    if (t != null) {
                        responseCF.completeExceptionally(t);
                    } else {
                        HttpResponseImpl<T> response = new HttpResponseImpl<>(r.request, r, body, getExchange());
                        responseCF.complete(response);
                    }
                });
            start.completeAsync(() -> null, getExchange().executor());
        }

        @Override
        void completeResponseExceptionally(Throwable t) {
            pushCF.completeExceptionally(t);
        }

        @Override
        synchronized void responseReceived() {
            super.responseReceived();
        }

        // create and return the PushResponseImpl
        @Override
        protected void handleResponse() {
            HttpConnection c = connection.connection; // TODO: improve
            responseCode = (int)responseHeaders
                .firstValueAsLong(":status")
                .orElse(-1);

            if (responseCode == -1) {
                completeResponseExceptionally(new IOException("No status code"));
            }

            this.response = new Response(
                pushReq, exchange, responseHeaders,
                responseCode, HttpClient.Version.HTTP_2);

            this.responseContentLen = responseHeaders
                .firstValueAsLong("content-length")
                .orElse(-1L);

            if (Log.headers()) {
                StringBuilder sb = new StringBuilder("RESPONSE HEADERS");
                sb.append(" (streamid=").append(streamid).append("): ");
                Log.dumpHeaders(sb, "    ", responseHeaders);
                Log.logHeaders(sb.toString());
            }

            // different implementations for normal streams and pushed streams
            completeResponse(response);
        }
    }

    final class StreamWindowUpdateSender extends WindowUpdateSender {

        StreamWindowUpdateSender(Http2Connection connection) {
            super(connection);
        }

        @Override
        int getStreamId() {
            return streamid;
        }
    }

}