Client push promise

src/main/java/io/vertx/core/http/HttpClientRequest.java

@@ -251,4 +251,28 @@ public interface HttpClientRequest extends WriteStream<Buffer>, ReadStream<HttpC
   @Fluent
   HttpClientRequest setTimeout(long timeoutMs);
 
+  /**
+   * Set a push promise handler for this request.<p/>
+   *
+   * The handler is called when the client receives a <i>push promise</i> from the server. The handler can be called
+   * multiple times, for each push promise.<p/>
+   *
+   * The handler is called with a <i>read-only</i> {@link HttpClientRequest}, the following methods can be called:<p/>
+   *
+   * <ul>
+   *   <li>{@link HttpClientRequest#method()}</li>
+   *   <li>{@link HttpClientRequest#uri()}</li>
+   *   <li>{@link HttpClientRequest#headers()}</li>
+   *   <li>{@link HttpClientRequest#getHost()}</li>
+   * </ul>
+   *
+   * In addition the handler should call the {@link HttpClientRequest#handler} method to set an handler for
+   * processing the response.<p/>
+   *
+   * @param handler the handler
+   * @return a reference to this, so the API can be used fluently
+   */
+  @Fluent
+  HttpClientRequest pushPromiseHandler(Handler<HttpClientRequest> handler);
+
 }

src/main/java/io/vertx/core/http/HttpServerResponse.java

@@ -398,7 +398,7 @@ default HttpServerResponse sendFile(String filename, long offset, Handler<AsyncR
    * @return a reference to this, so the API can be used fluently
    */
   @Fluent
-  HttpServerResponse promisePush(HttpMethod method, String path, Handler<AsyncResult<HttpServerResponse>> handler);
+  HttpServerResponse pushPromise(HttpMethod method, String path, Handler<AsyncResult<HttpServerResponse>> handler);
 
   /**
    * Reset this stream.

src/main/java/io/vertx/core/http/impl/ConnectionManager.java

@@ -175,7 +175,7 @@ public class ConnQueue {
       this.address = address;
 
       if (options.getProtocolVersion() == HttpVersion.HTTP_2) {
-        pool = new Http2Pool(this);
+        pool = new Http2Pool(this, client);
       } else {
         pool = new Http1xPool(this);
       }

src/main/java/io/vertx/core/http/impl/Http2Pool.java

@@ -47,6 +47,7 @@
 import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.net.NetSocket;
 
+import java.util.ArrayDeque;
 import java.util.Map;
 
 /**
@@ -55,9 +56,11 @@
 class Http2Pool extends ConnectionManager.Pool {
 
   private VertxClientHandler clientHandler;
+  private HttpClientImpl client;
 
-  public Http2Pool(ConnectionManager.ConnQueue queue) {
+  public Http2Pool(ConnectionManager.ConnQueue queue, HttpClientImpl client) {
     super(queue, 1);
+    this.client = client;
   }
 
   public boolean getConnection(HttpClientRequestImpl req, Handler<HttpClientStream> handler, ContextImpl context) {
@@ -112,7 +115,7 @@ class Http2ClientStream implements HttpClientStream {
     private boolean paused;
     private int numBytes;
 
-    public Http2ClientStream(HttpClientRequestImpl req,
+    public Http2ClientStream(HttpClientRequestBase req,
                              ContextImpl context,
                              ChannelHandlerContext handlerCtx,
                              Http2Connection conn,
@@ -177,6 +180,10 @@ private void handleEnd() {
       resp.handleEnd(new CaseInsensitiveHeaders());
     }
 
+    void handlePushPromise(HttpClientRequestBase promised) throws Http2Exception {
+      ((HttpClientRequestImpl)req).handlePush(promised);
+    }
+
     @Override
     public void writeHead(HttpMethod method, String uri, MultiMap headers, String hostHeader, boolean chunked) {
       throw new UnsupportedOperationException();
@@ -298,7 +305,7 @@ void handle(Handler<HttpClientStream> handler, HttpClientRequestImpl req) {
       handler.handle(stream);
     }
 
-    Http2ClientStream createStream(HttpClientRequestImpl req) {
+    Http2ClientStream createStream(HttpClientRequestBase req) {
       try {
         Http2ClientStream stream = new Http2ClientStream(req, context, handlerCtx, connection(), encoder());
         streams.put(stream.stream.id(), stream);
@@ -353,6 +360,15 @@ public void onPingAckRead(ChannelHandlerContext ctx, ByteBuf data) throws Http2E
 
     @Override
     public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId, Http2Headers headers, int padding) throws Http2Exception {
+      Http2ClientStream stream = streams.get(streamId);
+      HttpMethod method = UriUtils.toVertxMethod(headers.method().toString());
+      String uri = headers.path().toString();
+      String host = headers.authority() != null ? headers.authority().toString() : null;
+      MultiMap headersMap = new Http2HeadersAdaptor(headers);
+      HttpClientRequestPushPromise req = new HttpClientRequestPushPromise(client, method, uri, host, headersMap);
+      Http2ClientStream newStream = new Http2ClientStream(req, context, handlerCtx, stream.conn, stream.encoder);
+      streams.put(promisedStreamId, newStream);
+      stream.handlePushPromise(req);
     }
 
     @Override

src/main/java/io/vertx/core/http/impl/Http2ServerRequestImpl.java

@@ -285,11 +285,7 @@ public HttpVersion version() {
   public HttpMethod method() {
     if (method == null) {
       String sMethod = headers.method().toString();
-      try {
-        method = io.vertx.core.http.HttpMethod.valueOf(sMethod);
-      } catch (IllegalArgumentException e) {
-        method = HttpMethod.UNKNOWN;
-      }
+      method = UriUtils.toVertxMethod(sMethod);
     }
     return method;
   }

src/main/java/io/vertx/core/http/impl/Http2ServerResponseImpl.java

@@ -479,7 +479,7 @@ public void reset(long code) {
     ctx.flush();
   }
 
-  public HttpServerResponse promisePush(HttpMethod method, String path, Handler<AsyncResult<HttpServerResponse>> handler) {
+  public HttpServerResponse pushPromise(HttpMethod method, String path, Handler<AsyncResult<HttpServerResponse>> handler) {
     if (push) {
       throw new IllegalStateException("A push response cannot promise another push");
     }

src/main/java/io/vertx/core/http/impl/HttpClientRequestImpl.java

@@ -59,6 +59,7 @@ public class HttpClientRequestImpl extends HttpClientRequestBase implements Http
   private Handler<Void> continueHandler;
   private volatile HttpClientStream conn;
   private Handler<Void> drainHandler;
+  private Handler<HttpClientRequest> pushHandler;
   private boolean headWritten;
   private boolean completed;
   private ByteBuf pendingChunks;
@@ -333,6 +334,14 @@ public HttpClientRequest putHeader(CharSequence name, Iterable<CharSequence> val
     }
   }
 
+  @Override
+  public HttpClientRequest pushPromiseHandler(Handler<HttpClientRequest> handler) {
+    synchronized (getLock()) {
+      pushHandler = handler;
+    }
+    return this;
+  }
+
   void handleDrained() {
     synchronized (getLock()) {
       if (!completed && drainHandler != null) {
@@ -668,4 +677,12 @@ private void checkResponseHandler() {
       throw new IllegalStateException("You must set an handler for the HttpClientResponse before connecting");
     }
   }
+
+  void handlePush(HttpClientRequest pushedRequest) {
+    synchronized (getLock()) {
+      if (pushHandler != null) {
+        pushHandler.handle(pushedRequest);
+      }
+    }
+  }
 }

src/main/java/io/vertx/core/http/impl/HttpClientRequestPushPromise.java

@@ -0,0 +1,206 @@
+/*
+ * Copyright (c) 2011-2013 The original author or authors
+ *  ------------------------------------------------------
+ *  All rights reserved. This program and the accompanying materials
+ *  are made available under the terms of the Eclipse Public License v1.0
+ *  and Apache License v2.0 which accompanies this distribution.
+ *
+ *      The Eclipse Public License is available at
+ *      http://www.eclipse.org/legal/epl-v10.html
+ *
+ *      The Apache License v2.0 is available at
+ *      http://www.opensource.org/licenses/apache2.0.php
+ *
+ *  You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.vertx.core.http.impl;
+
+import io.vertx.codegen.annotations.Nullable;
+import io.vertx.core.Handler;
+import io.vertx.core.MultiMap;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClientRequest;
+import io.vertx.core.http.HttpClientResponse;
+import io.vertx.core.http.HttpMethod;
+
+/**
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ */
+class HttpClientRequestPushPromise extends HttpClientRequestBase {
+
+  private final HttpMethod method;
+  private final String uri;
+  private final String host;
+  private final MultiMap headers;
+  private Handler<HttpClientResponse> respHandler;
+
+  public HttpClientRequestPushPromise(HttpClientImpl client, HttpMethod method, String uri, String host, MultiMap headers) {
+    super(client);
+    this.method = method;
+    this.uri = uri;
+    this.host = host;
+    this.headers = headers;
+  }
+
+  @Override
+  protected Object getLock() {
+    return this; //
+  }
+
+  @Override
+  protected void doHandleResponse(HttpClientResponseImpl resp) {
+    synchronized (getLock()) {
+      if (respHandler != null) {
+        respHandler.handle(resp);
+      }
+    }
+  }
+
+  @Override
+  protected void checkComplete() {
+  }
+
+  @Override
+  public HttpClientRequest handler(Handler<HttpClientResponse> handler) {
+    synchronized (getLock()) {
+      respHandler = handler;
+      return this;
+    }
+  }
+
+  @Override
+  public boolean isChunked() {
+    return false;
+  }
+
+  @Override
+  public HttpMethod method() {
+    return method;
+  }
+
+  @Override
+  public String uri() {
+    return uri;
+  }
+
+  @Override
+  public String getHost() {
+    return host;
+  }
+
+  @Override
+  public MultiMap headers() {
+    return headers;
+  }
+
+  @Override
+  public HttpClientRequest write(Buffer data) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest setWriteQueueMaxSize(int maxSize) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest drainHandler(Handler<Void> handler) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest pause() {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest resume() {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest endHandler(Handler<Void> endHandler) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest setChunked(boolean chunked) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest setHost(String host) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putHeader(String name, String value) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putHeader(CharSequence name, CharSequence value) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putHeader(String name, Iterable<String> values) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest putHeader(CharSequence name, Iterable<CharSequence> values) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest write(String chunk) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest write(String chunk, String enc) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest continueHandler(@Nullable Handler<Void> handler) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest sendHead() {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public void end(String chunk) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public void end(String chunk, String enc) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public void end(Buffer chunk) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public HttpClientRequest pushPromiseHandler(Handler<HttpClientRequest> handler) {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public void end() {
+    throw new IllegalStateException();
+  }
+
+  @Override
+  public boolean writeQueueFull() {
+    throw new IllegalStateException();
+  }
+}

src/main/java/io/vertx/core/http/impl/HttpClientResponseImpl.java

@@ -20,6 +20,7 @@
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClientRequest;
 import io.vertx.core.http.HttpClientResponse;
 import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.http.HttpVersion;

src/main/java/io/vertx/core/http/impl/HttpServerResponseImpl.java

@@ -609,7 +609,7 @@ public void reset(long code) {
   }
 
   @Override
-  public HttpServerResponse promisePush(io.vertx.core.http.HttpMethod method, String path, Handler<AsyncResult<HttpServerResponse>> handler) {
+  public HttpServerResponse pushPromise(io.vertx.core.http.HttpMethod method, String path, Handler<AsyncResult<HttpServerResponse>> handler) {
     handler.handle(Future.failedFuture("Push promise is only supported with HTTP2"));
     return this;
   }