Simplify HTTP/2 framework.

Motivation:

The introduction of the HTTP/2 multiplex implementation has introduced a level of indirection between the HTTP client implementation and the internal stream API. The internal stream API is implemented by the HTTP/2 framework client stream in order to adapt one to the other.

Changes:

Remove the un-necessary indirection and let Http2ClientStream implement the internal HttpClientStream API.

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java

@@ -426,7 +426,6 @@ private static class StreamImpl extends Stream implements HttpClientStream {
     private Handler<HttpResponseHead> headHandler;
     private Handler<Buffer> chunkHandler;
     private Handler<MultiMap> trailerHandler;
-    private Handler<Void> endHandler;
     private Handler<Void> drainHandler;
     private Handler<Void> continueHandler;
 
@@ -450,13 +449,9 @@ protected void handlePause() {
         @Override
         protected void handleMessage(Object item) {
           if (item instanceof MultiMap) {
-            Handler<MultiMap> handler1 = trailerHandler;
-            if (handler1 != null) {
-              context.dispatch((MultiMap) item, handler1);
-            }
-            Handler<Void> handler2 = endHandler;
-            if (handler2 != null) {
-              context.dispatch(null, handler2);
+            Handler<MultiMap> handler = trailerHandler;
+            if (handler != null) {
+              context.dispatch((MultiMap) item, handler);
             }
           } else {
             Buffer buffer = (Buffer) item;
@@ -500,24 +495,29 @@ public HttpClientStream setWriteQueueMaxSize(int maxSize) {
     }
 
     @Override
-    public boolean writeQueueFull() {
-      return conn.writeQueueFull();
+    public boolean isWritable() {
+      return !conn.writeQueueFull();
     }
 
     @Override
-    public HttpClientStream headHandler(Handler<HttpResponseHead> handler) {
+    public HttpClientStream headersHandler(Handler<HttpResponseHead> handler) {
       this.headHandler = handler;
       return this;
     }
 
+    @Override
+    public HttpClientStream resetHandler(Handler<Long> handler) {
+      return this;
+    }
+
     @Override
     public HttpClientStream closeHandler(Handler<Void> handler) {
       closeHandler = handler;
       return this;
     }
 
     @Override
-    public HttpClientStream priorityHandler(Handler<StreamPriority> handler) {
+    public HttpClientStream priorityChangeHandler(Handler<StreamPriority> handler) {
       // No op
       return this;
     }
@@ -529,7 +529,7 @@ public HttpClientStream pushHandler(Handler<Http2ClientPush> handler) {
     }
 
     @Override
-    public HttpClientStream unknownFrameHandler(Handler<HttpFrame> handler) {
+    public HttpClientStream customFrameHandler(Handler<HttpFrame> handler) {
       // No op
       return this;
     }
@@ -593,35 +593,30 @@ public HttpClientStream pause() {
       return this;
     }
 
-    @Override
-    public HttpClientStream resume() {
-      queue.fetch(Long.MAX_VALUE);
-      return this;
-    }
-
     @Override
     public HttpClientStream fetch(long amount) {
       queue.fetch(amount);
       return this;
     }
 
     @Override
-    public Future<Void> reset(Throwable cause) {
+    public Future<Void> writeReset(long code) {
       Promise<Void> promise = context.promise();
       EventLoop eventLoop = conn.context.nettyEventLoop();
       if (eventLoop.inEventLoop()) {
-        reset(cause, promise);
+        reset(code, promise);
       } else {
-        eventLoop.execute(() -> reset(cause, promise));
+        eventLoop.execute(() -> reset(code, promise));
       }
       return promise.future();
     }
 
-    private void reset(Throwable cause, Promise<Void> promise) {
+    private void reset(long code, Promise<Void> promise) {
       Boolean removed = conn.reset(this);
       if (removed == null) {
         promise.fail("Stream already reset");
       } else {
+        Throwable cause = new StreamResetException(code);
         if (removed) {
           context.execute(cause, this::handleClosed);
         } else {
@@ -674,7 +669,7 @@ void handleHead(HttpResponseHead response) {
     }
 
     @Override
-    public HttpClientStream handler(Handler<Buffer> handler) {
+    public HttpClientStream dataHandler(Handler<Buffer> handler) {
       chunkHandler = handler;
       return this;
     }
@@ -685,12 +680,6 @@ public HttpClientStream trailersHandler(Handler<MultiMap> handler) {
       return this;
     }
 
-    @Override
-    public HttpClientStream endHandler(Handler<Void> handler) {
-      endHandler = handler;
-      return this;
-    }
-
     void handleChunk(Buffer buff) {
       queue.write(buff);
     }