Fix NetSocket (and QuicStream) 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/SocketBase.java

@@ -270,14 +270,17 @@ protected void handleShutdown(Duration timeout, ChannelPromise promise) {
     }
   }
 
-  protected void handleEnd() {
-    pending.write(InboundBuffer.END_SENTINEL);
+  protected void handleEnded() {
+    read(InboundBuffer.END_SENTINEL);
+    endRead();
   }
 
   @Override
   protected void handleMessage(Object msg) {
     MessageHandler handler = messageHandler();
-    if (handler.accept(msg)) {
+    if (msg == InboundBuffer.END_SENTINEL) {
+      pending.write(InboundBuffer.END_SENTINEL);
+    } else if (handler.accept(msg)) {
       pending.write(handler.transform(msg));
     } else {
       if (msg instanceof ReferenceCounted) {

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

@@ -308,6 +308,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/main/java/io/vertx/core/net/impl/quic/QuicStreamImpl.java

@@ -10,7 +10,6 @@
  */
 package io.vertx.core.net.impl.quic;
 
-import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.socket.ChannelInputShutdownEvent;
@@ -149,7 +148,7 @@ protected void handleIdle(IdleStateEvent event) {
   @Override
   protected void handleEvent(Object event) {
     if (event == ChannelInputShutdownEvent.INSTANCE) {
-      handleEnd();
+      handleEnded();
     } else {
       super.handleEvent(event);
     }

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

@@ -180,7 +180,7 @@ public String applicationLayerProtocol() {
 
   @Override
   protected void handleClosed() {
-    handleEnd();
+    handleEnded();
     super.handleClosed();
   }
 }

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

@@ -132,6 +132,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();

vertx-core/src/test/java/io/vertx/tests/net/quic/QuicClientTest.java

@@ -432,4 +432,34 @@ public void testInvalidApplicationProtocols() {
       }
     }
   }
+
+  @Test
+  public void testEndHandlerCalledAfterAllEmissions() throws Exception {
+    Buffer  buffer = TestUtils.randomBuffer(1024 * 1024);
+    server.handler(conn -> {
+      conn.handler(stream -> {
+        stream.endHandler(v -> {
+          stream.end(buffer);
+        });
+      });
+    });
+    server.bind(SocketAddress.inetSocketAddress(9999, "localhost")).await();
+    client.bind(SocketAddress.inetSocketAddress(0, "localhost")).await();
+    QuicConnection connection = client.connect(SocketAddress.inetSocketAddress(9999, "localhost")).await();
+    QuicStream stream = connection.openStream().await();
+    AtomicInteger received = new AtomicInteger();
+    AtomicInteger ended = new AtomicInteger();
+    stream.handler(buf -> {
+      received.addAndGet(buf.length());
+      assertEquals(0, ended.get());
+      stream.pause();
+      vertx.setTimer(50, t -> stream.resume());
+    });
+    stream.endHandler(v -> {
+      assertEquals(0, ended.getAndIncrement());
+    });
+    stream.end();
+    assertWaitUntil(() -> received.get() == buffer.length());
+    assertWaitUntil(() -> ended.get() > 0);
+  }
 }