--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/Stream.java Tue Sep 12 19:03:39 2017 +0200
@@ -0,0 +1,963 @@
+/*
+ * 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.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;
+
+/**
+ * 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<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();
+ for (ByteBufferReference b : buffers) {
+ ByteBuffer buf = b.get();
+ if (buf.hasRemaining()) {
+ publisher.acceptData(Optional.of(buf));
+ }
+ }
+ 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;
+ }
+ }
+
+}