Stream write result

vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1ServerConnection.java

@@ -38,7 +38,7 @@
 import io.vertx.core.internal.http.QueryParamDecoder;
 import io.vertx.core.net.NetSocket;
 import io.vertx.core.net.ServerSSLOptions;
-import io.vertx.core.net.impl.MessageWrite;
+import io.vertx.core.net.impl.WritePromise;
 import io.vertx.core.net.impl.tcp.NetSocketImpl;
 import io.vertx.core.internal.tls.SslContextManager;
 import io.vertx.core.net.impl.VertxHandler;
@@ -263,20 +263,18 @@ private void onEnd() {
     }
   }
 
-  void write(VertxHttpObject msg, Promise<Void> promise) {
-    writeToChannel(new MessageWrite() {
+  Future<Void> write(VertxHttpObject msg) {
+    WritePromise write = new WritePromise(context) {
       @Override
       public void write() {
-        Http1ServerConnection.this.unsafeWrite(msg, false, promise);
+        Http1ServerConnection.this.unsafeWrite(msg, false, this);
         if (msg.isEnded()) {
           responseComplete();
         }
       }
-      @Override
-      public void cancel(Throwable cause) {
-        promise.fail(cause);
-      }
-    });
+    };
+    writeToChannel(write);
+    return write;
   }
 
   void responseComplete() {
@@ -487,25 +485,25 @@ protected void handleWriteQueueDrained() {
     }
   }
 
-  void write100Continue(Promise<Void> promise) {
-    write(new VertxFullHttpResponse(
+  Future<Void> write100Continue() {
+    return write(new VertxFullHttpResponse(
       false,
       HTTP_1_1,
       CONTINUE,
       Unpooled.buffer(0),
       DefaultHttpHeadersFactory.headersFactory().newHeaders(),
       DefaultHttpHeadersFactory.trailersFactory().newHeaders(),
-      false), promise);
+      false));
   }
 
-  void write103EarlyHints(HttpHeaders headers, Promise<Void> promise) {
-    write(new VertxFullHttpResponse(false,
+  Future<Void> write103EarlyHints(HttpHeaders headers) {
+    return write(new VertxFullHttpResponse(false,
       HTTP_1_1,
       HttpResponseStatus.EARLY_HINTS,
       Unpooled.buffer(0),
       headers,
       EmptyHttpHeaders.INSTANCE,
-      false), promise);
+      false));
   }
 
   protected void handleClosed() {

vertx-core/src/main/java/io/vertx/core/http/impl/http1/Http1ServerResponse.java

@@ -312,7 +312,7 @@ public HttpServerResponse endHandler(@Nullable Handler<Void> handler) {
   @Override
   public Future<Void> writeHead() {
     checkThread();
-    PromiseInternal<Void> promise = context.promise();
+    Future<Void> f;
     synchronized (conn) {
       if (headWritten) {
         throw new IllegalStateException();
@@ -323,44 +323,35 @@ public Future<Void> writeHead() {
       VertxHttpObject msg;
       prepareHeaders(-1);
       msg = new VertxHttpResponse(head, version, status, headers);
-      conn.write(msg, promise);
+      f = conn.write(msg);
     }
-    return promise.future();
+    return f;
   }
 
   @Override
   public Future<Void> write(Buffer chunk) {
-    PromiseInternal<Void> promise = context.promise();
-    write(((BufferInternal)chunk).getByteBuf(), promise);
-    return promise.future();
+    return write(((BufferInternal)chunk).getByteBuf());
   }
 
   @Override
   public Future<Void> write(String chunk, String enc) {
-    PromiseInternal<Void> promise = context.promise();
-    write(BufferInternal.buffer(chunk, enc).getByteBuf(), promise);
-    return promise.future();
+    return write(BufferInternal.buffer(chunk, enc).getByteBuf());
   }
 
   @Override
   public Future<Void> write(String chunk) {
-    PromiseInternal<Void> promise = context.promise();
-    write(BufferInternal.buffer(chunk).getByteBuf(), promise);
-    return promise.future();
+    return write(BufferInternal.buffer(chunk).getByteBuf());
   }
 
   @Override
   public Future<Void> writeContinue() {
     checkThread();
-    Promise<Void> promise = context.promise();
-    conn.write100Continue(promise);
-    return promise.future();
+    return conn.write100Continue();
   }
 
   @Override
   public Future<Void> writeEarlyHints(MultiMap headers) {
     checkThread();
-    PromiseInternal<Void> promise = context.promise();
     Http1xHeaders headersMultiMap;
     if (headers instanceof Http1xHeaders) {
       headersMultiMap = (Http1xHeaders) headers;
@@ -371,8 +362,7 @@ public Future<Void> writeEarlyHints(MultiMap headers) {
     synchronized (conn) {
       checkHeadWritten();
     }
-    conn.write103EarlyHints(headersMultiMap, promise);
-    return promise.future();
+    return conn.write103EarlyHints(headersMultiMap);
   }
 
   @Override
@@ -387,12 +377,6 @@ public Future<Void> end(String chunk, String enc) {
 
   @Override
   public Future<Void> end(Buffer chunk) {
-    PromiseInternal<Void> promise = context.promise();
-    end(chunk, promise);
-    return promise.future();
-  }
-
-  private void end(Buffer chunk, PromiseInternal<Void> listener) {
     checkThread();
     synchronized (conn) {
       if (written) {
@@ -410,7 +394,7 @@ private void end(Buffer chunk, PromiseInternal<Void> listener) {
       } else {
         msg = new VertxLastHttpContent(data, trailingHeaders);
       }
-      conn.write(msg, listener);
+      Future<Void> result = conn.write(msg);
       if (bodyEndHandler != null) {
         bodyEndHandler.handle(null);
       }
@@ -420,6 +404,7 @@ private void end(Buffer chunk, PromiseInternal<Void> listener) {
       if (!keepAlive) {
         closed = true; // ?????
       }
+      return result;
     }
   }
 
@@ -514,7 +499,7 @@ private Future<Void> sendFileInternal(long offset, long length, long size, Rando
         prepareHeaders(actualLength);
         bytesWritten = actualLength;
         written = true;
-        conn.write(new VertxAssembledHttpResponse(head, version, status, headers), null);
+        conn.write(new VertxAssembledHttpResponse(head, version, status, headers));
         FileChannel toSend = fileChannel == null ? file.getChannel() : fileChannel;
         ChannelFuture channelFuture = conn.sendFile(toSend, actualOffset, actualLength);
         PromiseInternal<Void> promise = context.promise();
@@ -541,7 +526,8 @@ private Future<Void> sendFileInternal(long offset, long length, long size, Rando
             }
 
             // write an empty last content to let the http encoder know the response is complete
-            conn.write(new VertxLastHttpContent(Unpooled.buffer(0), DefaultHttpHeadersFactory.trailersFactory().newHeaders()), promise);
+            Future<Void> f = conn.write(new VertxLastHttpContent(Unpooled.buffer(0), DefaultHttpHeadersFactory.trailersFactory().newHeaders()));
+            f.onComplete(promise);
           } else {
             promise.fail(future.cause());
           }
@@ -718,7 +704,7 @@ private void reportResponseBegin() {
     }
   }
 
-  private Http1ServerResponse write(ByteBuf chunk, PromiseInternal<Void> promise) {
+  private Future<Void> write(ByteBuf chunk) {
     checkThread();
     synchronized (conn) {
       if (written) {
@@ -736,8 +722,7 @@ private Http1ServerResponse write(ByteBuf chunk, PromiseInternal<Void> promise)
       } else {
         msg = new VertxHttpContent(chunk);
       }
-      conn.write(msg, promise);
-      return this;
+      return conn.write(msg);
     }
   }
 
@@ -758,8 +743,7 @@ Future<NetSocket> netSocket(HttpMethod requestMethod, MultiMap requestHeaders) {
         }
         status = requestMethod == HttpMethod.CONNECT ? HttpResponseStatus.OK : HttpResponseStatus.SWITCHING_PROTOCOLS;
         prepareHeaders(-1);
-        PromiseInternal<Void> upgradePromise = context.promise();
-        conn.write(new VertxAssembledHttpResponse(head, version, status, headers), upgradePromise);
+        conn.write(new VertxAssembledHttpResponse(head, version, status, headers));
         written = true;
         Promise<NetSocket> promise = context.promise();
         netSocket = promise.future();

vertx-core/src/main/java/io/vertx/core/http/impl/http2/DefaultHttp2ClientStream.java

@@ -35,7 +35,7 @@
 import io.vertx.core.internal.PromiseInternal;
 import io.vertx.core.internal.buffer.BufferInternal;
 import io.vertx.core.net.HostAndPort;
-import io.vertx.core.net.impl.MessageWrite;
+import io.vertx.core.net.impl.WritePromise;
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.core.spi.metrics.TransportMetrics;
 import io.vertx.core.spi.tracing.VertxTracer;
@@ -117,7 +117,7 @@ public Future<Void> writeHead(HttpRequestHead request, boolean chunked, Buffer b
     priority(priority);
     scheme = request.scheme;
     authority = request.authority;
-    write(new HeadersWrite(request, buf != null ? ((BufferInternal)buf).getByteBuf() : null, end, promise));
+    write(new HeadersWrite(context, request, buf != null ? ((BufferInternal)buf).getByteBuf() : null, end, promise));
     return promise.future();
   }
 
@@ -128,17 +128,18 @@ public HttpClientStream setWriteQueueMaxSize(int maxSize) {
 
   void writeHeaders(HttpRequestHead request, ByteBuf buf, boolean end, StreamPriority priority, Promise<Void> promise) {
     priority(priority);
-    write(new HeadersWrite(request, buf, end, promise));
+    write(new HeadersWrite(context, request, buf, end, promise));
   }
 
-  private class HeadersWrite implements MessageWrite {
+  private class HeadersWrite extends WritePromise {
 
     private final HttpRequestHead request;
     private final ByteBuf buf;
     private final boolean end;
     private final Promise<Void> promise;
 
-    public HeadersWrite(HttpRequestHead request, ByteBuf buf, boolean end, Promise<Void> promise) {
+    public HeadersWrite(ContextInternal context,  HttpRequestHead request, ByteBuf buf, boolean end, Promise<Void> promise) {
+      super(context);
       this.request = request;
       this.buf = buf;
       this.end = end;

vertx-core/src/main/java/io/vertx/core/http/impl/http2/DefaultHttp2ServerStream.java

@@ -164,16 +164,14 @@ void handleHeader(HttpHeaders map) {
   }
 
   public final Future<Void> writeHead(HttpResponseHead head, Buffer chunk, boolean end) {
-    Promise<Void> promise = context.promise();
     HttpResponseHeaders headers = (HttpResponseHeaders)head.headers();
     headers.status(head.statusCode);
     if (chunk != null) {
-      writeHeaders(headers, false, false, null);
-      writeData(((BufferInternal)chunk).getByteBuf(), end, promise);
+      writeHeaders(headers, false, false);
+      return writeData(((BufferInternal)chunk).getByteBuf(), end);
     } else {
-      writeHeaders(headers, end, true, promise);
+      return writeHeaders(headers, end, true);
     }
-    return promise.future();
   }
 
   public void routed(String route) {