src/java.net.http/share/classes/jdk/internal/net/http/PlainTunnelingConnection.java

*** 24,38 ****
   */
  
  package jdk.internal.net.http;
  
  import java.io.IOException;
- import java.lang.System.Logger.Level;
  import java.net.InetSocketAddress;
  import java.nio.ByteBuffer;
  import java.nio.channels.SocketChannel;
  import java.util.concurrent.CompletableFuture;
  import java.util.function.Function;
  import java.net.http.HttpHeaders;
  import jdk.internal.net.http.common.FlowTube;
  import jdk.internal.net.http.common.MinimalFuture;
  import static java.net.http.HttpResponse.BodyHandlers.discarding;
--- 24,40 ----
   */
  
  package jdk.internal.net.http;
  
  import java.io.IOException;
  import java.net.InetSocketAddress;
+ import java.net.http.HttpTimeoutException;
  import java.nio.ByteBuffer;
  import java.nio.channels.SocketChannel;
+ import java.time.Duration;
  import java.util.concurrent.CompletableFuture;
+ import java.util.concurrent.CompletionException;
  import java.util.function.Function;
  import java.net.http.HttpHeaders;
  import jdk.internal.net.http.common.FlowTube;
  import jdk.internal.net.http.common.MinimalFuture;
  import static java.net.http.HttpResponse.BodyHandlers.discarding;
*** 58,78 ****
          this.proxyHeaders = proxyHeaders;
          delegate = new PlainHttpConnection(proxy, client);
      }
  
      @Override
!     public CompletableFuture<Void> connectAsync() {
          if (debug.on()) debug.log("Connecting plain connection");
!         return delegate.connectAsync()
              .thenCompose((Void v) -> {
                  if (debug.on()) debug.log("sending HTTP/1.1 CONNECT");
                  HttpClientImpl client = client();
                  assert client != null;
                  HttpRequestImpl req = new HttpRequestImpl("CONNECT", address, proxyHeaders);
                  MultiExchange<Void> mulEx = new MultiExchange<>(null, req,
                          client, discarding(), null, null);
!                 Exchange<Void> connectExchange = new Exchange<>(req, mulEx);
  
                  return connectExchange
                          .responseAsyncImpl(delegate)
                          .thenCompose((Response resp) -> {
                              CompletableFuture<Void> cf = new MinimalFuture<>();
--- 60,81 ----
          this.proxyHeaders = proxyHeaders;
          delegate = new PlainHttpConnection(proxy, client);
      }
  
      @Override
!     public CompletableFuture<Void> connectAsync(Exchange<?> exchange) {
          if (debug.on()) debug.log("Connecting plain connection");
!         return delegate.connectAsync(exchange)
!             .thenCompose(unused -> delegate.finishConnect())
              .thenCompose((Void v) -> {
                  if (debug.on()) debug.log("sending HTTP/1.1 CONNECT");
                  HttpClientImpl client = client();
                  assert client != null;
                  HttpRequestImpl req = new HttpRequestImpl("CONNECT", address, proxyHeaders);
                  MultiExchange<Void> mulEx = new MultiExchange<>(null, req,
                          client, discarding(), null, null);
!                 Exchange<Void> connectExchange = mulEx.getExchange();
  
                  return connectExchange
                          .responseAsyncImpl(delegate)
                          .thenCompose((Response resp) -> {
                              CompletableFuture<Void> cf = new MinimalFuture<>();
*** 94,109 ****
                              } else {
                                  // get the initial/remaining bytes
                                  ByteBuffer b = ((Http1Exchange<?>)connectExchange.exchImpl).drainLeftOverBytes();
                                  int remaining = b.remaining();
                                  assert remaining == 0: "Unexpected remaining: " + remaining;
-                                 connected = true;
                                  cf.complete(null);
                              }
                              return cf;
                          });
!             });
      }
  
      @Override
      boolean isTunnel() { return true; }
  
--- 97,134 ----
                              } else {
                                  // get the initial/remaining bytes
                                  ByteBuffer b = ((Http1Exchange<?>)connectExchange.exchImpl).drainLeftOverBytes();
                                  int remaining = b.remaining();
                                  assert remaining == 0: "Unexpected remaining: " + remaining;
                                  cf.complete(null);
                              }
                              return cf;
+                         })
+                         .handle((result, ex) -> {
+                             if (ex == null) {
+                                 return MinimalFuture.completedFuture(result);
+                             } else {
+                                 if (debug.on())
+                                     debug.log("tunnel failed with \"%s\"", ex.toString());
+                                 Throwable t = ex;
+                                 if (t instanceof CompletionException)
+                                     t = t.getCause();
+                                 if (t instanceof HttpTimeoutException) {
+                                     String msg = "proxy tunneling CONNECT request timed out";
+                                     t = new HttpTimeoutException(msg);
+                                     t.initCause(ex);
+                                 }
+                                 return MinimalFuture.<Void>failedFuture(t);
+                             }
+                         })
+                         .thenCompose(Function.identity());
              });
!     }
! 
!     public CompletableFuture<Void> finishConnect() {
!         connected = true;
!         return MinimalFuture.completedFuture(null);
      }
  
      @Override
      boolean isTunnel() { return true; }