Fix NetSocket where the end signal is notified before all messages have been dispatched.

Motivation:

The NetSocket implementation directly write the end sentinel message to the pending message queue, ignoging the potentially buffered messages in the connection.

Changes:

When we consider the message stream ends, we should instead model the last message as a connection message to ensure no reorder happens.

Author: vietj

vertx-core/src/main/java/io/vertx/core/net/impl/NetSocketImpl.java

@@ -180,7 +180,7 @@ public synchronized NetSocket handler(Handler<Buffer> dataHandler) {
     return this;
   }
 
-  private synchronized Handler<Object> messageHandler() {
+  private synchronized MessageHandler messageHandler() {
     return messageHandler;
   }
 
@@ -190,6 +190,10 @@ public synchronized NetSocketInternal messageHandler(Handler<Object> handler) {
       messageHandler = new DataMessageHandler();
     } else {
       messageHandler = new MessageHandler() {
+        @Override
+        public void handleEnd() {
+          // Noop
+        }
         @Override
         public void pause() {
           doPause();
@@ -390,16 +394,25 @@ public Future<Void> end() {
   protected void handleShutdown(Object reason, long timeout, TimeUnit unit, ChannelPromise promise) {
   }
 
+  private void handleEnded() {
+    read(InboundBuffer.END_SENTINEL);
+    endRead();
+  }
+
   @Override
   protected void handleClosed() {
-    pending.write(InboundBuffer.END_SENTINEL);
+    handleEnded();
     super.handleClosed();
   }
 
   @Override
   public void handleMessage(Object msg) {
-    Handler<Object> handler = messageHandler();
-    handler.handle(msg);
+    MessageHandler handler = messageHandler();
+    if (msg == InboundBuffer.END_SENTINEL) {
+      handler.handleEnd();
+    } else {
+      handler.handle(msg);
+    }
   }
 
   @Override
@@ -432,6 +445,7 @@ public NetSocketImpl shutdownHandler(@Nullable Handler<Void> handler) {
   interface MessageHandler extends Handler<Object> {
     void pause();
     void fetch(long amount);
+    void handleEnd();
   }
 
   private class DataMessageHandler implements MessageHandler {
@@ -446,6 +460,11 @@ public void handle(Object msg) {
       }
     }
 
+    @Override
+    public void handleEnd() {
+      pending.write(InboundBuffer.END_SENTINEL);
+    }
+
     @Override
     public void pause() {
       pending.pause();

vertx-core/src/main/java/io/vertx/core/net/impl/VertxConnection.java

@@ -255,6 +255,10 @@ final void read(Object msg) {
     }
   }
 
+  final void endRead() {
+    read = false;
+  }
+
   private void addPending(Object msg) {
     if (pending == null) {
       pending = new ArrayDeque<>();

vertx-core/src/test/java/io/vertx/tests/net/NetTest.java

@@ -131,6 +131,37 @@ protected void tearDown() throws Exception {
     super.tearDown();
   }
 
+  @Test
+  public void testEndHandlerCalledAfterAllEmissions() {
+    Buffer  buffer = TestUtils.randomBuffer(1024 * 1024);
+    server = vertx.createNetServer().connectHandler(so -> {
+      so.end(buffer);
+      so.close();
+    });
+    server.listen(1234).await();
+    NetClient client = vertx.createNetClient();
+    AtomicInteger received = new AtomicInteger();
+    AtomicInteger ended = new AtomicInteger();
+    client.connect(1234, "localhost").onComplete(ar -> {
+      if (ar.succeeded()) {
+        NetSocket socket = ar.result();
+        socket.handler(buf -> {
+          int amount = received.addAndGet(buf.length());
+          assertEquals(0, ended.get());
+          socket.pause();
+          vertx.setTimer(50, t -> {
+            socket.resume();
+          });
+        });
+        socket.endHandler(v -> {
+          assertEquals(0, ended.getAndIncrement());
+        });
+      }
+    });
+    assertWaitUntil(() -> received.get() == buffer.length());
+    assertWaitUntil(() -> ended.get() > 0);
+  }
+
   @Test
   public void testClientOptions() {
     NetClientOptions options = new NetClientOptions();