Delay the end handler notification when an stream end handler is call…

…ed - see #2852
eclipse-vertx · Feb 25, 2019 · 24434e9 · 24434e9
1 parent 90455f8
commit 24434e9
Show file tree

Hide file tree

Showing 17 changed files with 229 additions and 76 deletions.
diff --git a/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java b/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java
@@ -69,6 +69,7 @@ public class AsyncFileImpl implements AsyncFile {
   private int lwm = maxWrites / 2;
   private int readBufferSize = DEFAULT_READ_BUFFER_SIZE;
   private InboundBuffer<Buffer> queue;
+  private Handler<Buffer> handler;
   private Handler<Void> endHandler;
   private long readPos;
   private long readLength = Long.MAX_VALUE;
@@ -102,7 +103,13 @@ public class AsyncFileImpl implements AsyncFile {
     }
     this.context = context;
     this.queue = new InboundBuffer<>(context, 0);
-
+    queue.handler(buff -> {
+      if (buff.length() > 0) {
+        handleBuffer(buff);
+      } else {
+        handleEnd();
+      }
+    });
     queue.drainHandler(v -> {
       doRead();
     });
@@ -235,7 +242,7 @@ public synchronized AsyncFile handler(Handler<Buffer> handler) {
     if (closed) {
       return this;
     }
-    queue.handler(handler);
+    this.handler = handler;
     if (handler != null) {
       doRead();
     } else {
@@ -346,24 +353,27 @@ private synchronized void doRead(ByteBuffer bb) {
     doRead(buff, 0, bb, readPos, ar -> {
       if (ar.succeeded()) {
         Buffer buffer = ar.result();
-        if (buffer.length() == 0) {
+        readPos += buffer.length();
-          // Empty buffer represents end of file
+        readLength -= buffer.length();
-          handleEnd();
+        // Empty buffer represents end of file
-        } else {
+        if (queue.write(buffer) && buffer.length() > 0) {
-          readPos += buffer.length();
+          doRead(bb);
-          readLength -= buffer.length();
-          if (queue.write(buffer)) {
-            doRead(bb);
-          }
         }
       } else {
         handleException(ar.cause());
       }
     });
   }
 
+  private synchronized void handleBuffer(Buffer buff) {
+    if (handler != null) {
+      checkContext();
+      handler.handle(buff);
+    }
+  }
+
   private synchronized void handleEnd() {
-    queue.handler(null);
+    handler = null;
     if (endHandler != null) {
       checkContext();
       endHandler.handle(null);

diff --git a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java
@@ -348,12 +348,9 @@ public void doPause() {
 
     @Override
     public void doFetch(long amount) {
-      queue.fetch(amount);
+      if (!queue.fetch(amount)) {
-    }
+        response.handleEnd(trailers);
-
+      }
-    @Override
-    public void doResume() {
-      queue.resume();
     }
 
     @Override
@@ -489,7 +486,7 @@ private boolean endResponse(LastHttpContent trailer) {
           }
         }
         trailers = new HeadersAdaptor(trailer.trailingHeaders());
-        if (queue.isEmpty()) {
+        if (queue.isEmpty() && !queue.isPaused()) {
           response.handleEnd(trailers);
         }
         responseEnded = true;

diff --git a/src/main/java/io/vertx/core/http/impl/Http2ServerRequestImpl.java b/src/main/java/io/vertx/core/http/impl/Http2ServerRequestImpl.java
@@ -225,19 +225,22 @@ public HttpServerRequest handler(Handler<Buffer> handler) {
 
   @Override
   public HttpServerRequest pause() {
-    doPause();
+    synchronized (conn) {
+      checkEnded();
+      doPause();
+    }
     return this;
   }
 
   @Override
   public HttpServerRequest resume() {
-    doResume();
+    return fetch(Long.MAX_VALUE);
-    return this;
   }
 
   @Override
   public HttpServerRequest fetch(long amount) {
     synchronized (conn) {
+      checkEnded();
       doFetch(amount);
     }
     return this;

diff --git a/src/main/java/io/vertx/core/http/impl/Http2UpgradedClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http2UpgradedClientConnection.java
@@ -215,11 +215,6 @@ public void doPause() {
       stream.doPause();
     }
 
-    @Override
-    public void doResume() {
-      stream.doResume();
-    }
-
     @Override
     public void doFetch(long amount) {
       stream.doFetch(amount);

diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientResponseImpl.java b/src/main/java/io/vertx/core/http/impl/HttpClientResponseImpl.java
@@ -132,26 +132,41 @@ public List<String> cookies() {
     }
   }
 
+  private void checkEnded() {
+    if (trailers != null) {
+      throw new IllegalStateException();
+    }
+  }
+
   @Override
-  public HttpClientResponse handler(Handler<Buffer> dataHandler) {
+  public HttpClientResponse handler(Handler<Buffer> handle) {
     synchronized (conn) {
-      this.dataHandler = dataHandler;
+      if (handle != null) {
+        checkEnded();
+      }
+      dataHandler = handle;
       return this;
     }
   }
 
   @Override
-  public HttpClientResponse endHandler(Handler<Void> endHandler) {
+  public HttpClientResponse endHandler(Handler<Void> handler) {
     synchronized (conn) {
-      this.endHandler = endHandler;
+      if (handler != null) {
+        checkEnded();
+      }
+      endHandler = handler;
       return this;
     }
   }
 
   @Override
-  public HttpClientResponse exceptionHandler(Handler<Throwable> exceptionHandler) {
+  public HttpClientResponse exceptionHandler(Handler<Throwable> handler) {
     synchronized (conn) {
-      this.exceptionHandler = exceptionHandler;
+      if (handler != null) {
+        checkEnded();
+      }
+      exceptionHandler = handler;
       return this;
     }
   }
@@ -164,8 +179,7 @@ public HttpClientResponse pause() {
 
   @Override
   public HttpClientResponse resume() {
-    stream.doResume();
+    return fetch(Long.MAX_VALUE);
-    return this;
   }
 
   @Override
@@ -175,16 +189,24 @@ public HttpClientResponse fetch(long amount) {
   }
 
   @Override
-  public HttpClientResponse bodyHandler(final Handler<Buffer> bodyHandler) {
+  public HttpClientResponse bodyHandler(final Handler<Buffer> handler) {
-    BodyHandler handler = new BodyHandler();
+    if (handler != null) {
-    handler(handler);
+      BodyHandler bodyHandler = new BodyHandler();
-    endHandler(v -> handler.notifyHandler(bodyHandler));
+      handler(bodyHandler);
+      endHandler(v -> bodyHandler.notifyHandler(handler));
+    } else {
+      handler(null);
+      endHandler(null);
+    }
     return this;
   }
 
   @Override
   public HttpClientResponse customFrameHandler(Handler<HttpFrame> handler) {
     synchronized (conn) {
+      if (endHandler != null) {
+        checkEnded();
+      }
       customFrameHandler = handler;
       return this;
     }
@@ -217,16 +239,19 @@ void handleChunk(Buffer data) {
   }
 
   void handleEnd(MultiMap trailers) {
+    Handler<Void> handler;
     synchronized (conn) {
       stream.reportBytesRead(bytesRead);
       bytesRead = 0;
       this.trailers = trailers;
-      if (endHandler != null) {
+      handler = endHandler;
-        try {
+      endHandler = null;
-          endHandler.handle(null);
+    }
-        } catch (Throwable t) {
+    if (handler != null) {
-          handleException(t);
+      try {
-        }
+        handler.handle(null);
+      } catch (Throwable t) {
+        handleException(t);
       }
     }
   }
@@ -277,6 +302,9 @@ void notifyHandler(Handler<Buffer> bodyHandler) {
   @Override
   public HttpClientResponse streamPriorityHandler(Handler<StreamPriority> handler) {
     synchronized (conn) {
+      if (handler != null) {
+        checkEnded();
+      }
       priorityHandler = handler;
     }
     return this;

diff --git a/src/main/java/io/vertx/core/http/impl/HttpClientStream.java b/src/main/java/io/vertx/core/http/impl/HttpClientStream.java
@@ -49,7 +49,6 @@ interface HttpClientStream {
   void doSetWriteQueueMaxSize(int size);
   boolean isNotWritable();
   void doPause();
-  void doResume();
   void doFetch(long amount);
 
   void reset(long code);

diff --git a/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java b/src/main/java/io/vertx/core/http/impl/HttpServerRequestImpl.java
@@ -161,7 +161,7 @@ void handlePipelined() {
     boolean end = ended;
     ended = false;
     handleBegin();
-    if (pending != null && pending.size() > 0) {
+    if (pending != null && pending.isPaused()) {
       pending.resume();
     }
     if (end) {
@@ -340,7 +340,13 @@ public HttpServerRequest fetch(long amount) {
   public HttpServerRequest resume() {
     synchronized (conn) {
       if (!isEnded()) {
-        pendingQueue().resume();
+        if (ended) {
+          if (!pending.resume()) {
+            doEnd();
+          }
+        } else if (pending != null) {
+          pending.resume();
+        }
       }
       return this;
     }
@@ -478,7 +484,7 @@ public SocketAddress localAddress() {
   @Override
   public boolean isEnded() {
     synchronized (conn) {
-      return ended && (pending == null || pending.isEmpty());
+      return ended && (pending == null || (!pending.isPaused() && pending.isEmpty()));
     }
   }
 

diff --git a/src/main/java/io/vertx/core/http/impl/VertxHttp2NetSocket.java b/src/main/java/io/vertx/core/http/impl/VertxHttp2NetSocket.java
@@ -151,8 +151,7 @@ public NetSocket pause() {
 
   @Override
   public NetSocket resume() {
-    doResume();
+    return fetch(Long.MAX_VALUE);
-    return this;
   }
 
   @Override

diff --git a/src/main/java/io/vertx/core/http/impl/VertxHttp2Stream.java b/src/main/java/io/vertx/core/http/impl/VertxHttp2Stream.java
@@ -36,7 +36,7 @@ abstract class VertxHttp2Stream<C extends Http2ConnectionBase> {
   protected final ChannelHandlerContext handlerContext;
   protected final Http2Stream stream;
 
-  private InboundBuffer<Buffer> pending;
+  private final InboundBuffer<Buffer> pending;
   private int pendingBytes;
   private MultiMap trailers;
   private boolean writable;
@@ -95,7 +95,7 @@ void onEnd() {
   void onEnd(MultiMap map) {
     synchronized (conn) {
       trailers = map;
-      if (pending.isEmpty()) {
+      if (pending.isEmpty() && !pending.isPaused()) {
         handleEnd(trailers);
       }
     }
@@ -109,12 +109,12 @@ public void doPause() {
     pending.pause();
   }
 
-  public void doResume() {
-    pending.resume();
-  }
-
   public void doFetch(long amount) {
-    pending.fetch(amount);
+    if (!pending.fetch(amount)) {
+      if (trailers != null) {
+        handleEnd(trailers);
+      }
+    }
   }
 
   boolean isNotWritable() {

diff --git a/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java b/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java
@@ -21,6 +21,7 @@
 import io.vertx.core.http.WebSocketFrame;
 import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
 import io.vertx.core.http.impl.ws.WebSocketFrameInternal;
+import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.streams.impl.InboundBuffer;
@@ -444,7 +445,7 @@ void handleClosed() {
     Handler<Void> endHandler;
     Handler<Void> closeHandler;
     synchronized (conn) {
-      endHandler = this.endHandler;
+      endHandler = pending.isPaused() ? null : this.endHandler;
       closeHandler = this.closeHandler;
       closed = true;
       binaryHandlerRegistration = null;
@@ -548,8 +549,17 @@ public S pause() {
 
   @Override
   public S resume() {
-    if (!isClosed()) {
+    synchronized (this) {
-      pending.resume();
+      if (isClosed()) {
+        Handler<Void> handler = endHandler;
+        endHandler = null;
+        if (handler != null) {
+          ContextInternal ctx = conn.getContext();
+          ctx.runOnContext(v -> handler.handle(null));
+        }
+      } else {
+        pending.resume();
+      }
     }
     return (S) this;
   }