HTTP/1 client connection unsolicited messages should be treated as invalid.

Motivation:

The current implementation of HTTP/1 client connection fails the connection when it receives unsolicited messages and does not release them. Such messages should be properly released to avoid leaking referenced counted messages.

Changes:

Update the HTTP/1 client connection to handle unsolicited messages as invalid, since invalid messages are released by the invalid message handler.

Author: vietj

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

@@ -75,10 +75,7 @@ public class Http1xClientConnection extends Http1xConnectionBase<WebSocketImpl>
 
   private static final Logger log = LoggerFactory.getLogger(Http1xClientConnection.class);
 
-  private static final Handler<Object> INVALID_MSG_HANDLER = msg -> {
-    ReferenceCountUtil.release(msg);
-    throw new IllegalStateException("Invalid object " + msg);
-  };
+  private static final Handler<Object> INVALID_MSG_HANDLER = ReferenceCountUtil::release;
 
   private final HttpClientBase client;
   private final HttpClientOptions options;
@@ -137,6 +134,12 @@ public HttpClientConnection evictionHandler(Handler<Void> handler) {
     return this;
   }
 
+  @Override
+  public HttpClientConnection invalidMessageHandler(Handler<Object> handler) {
+    invalidMessageHandler = handler;
+    return this;
+  }
+
   @Override
   public HttpClientConnection concurrencyChangeHandler(Handler<Long> handler) {
     // Never changes
@@ -800,6 +803,7 @@ public void handleMessage(Object msg) {
       handleWsFrame((WebSocketFrame) msg);
     } else {
       invalidMessageHandler.handle(msg);
+      fail(new VertxException("Received an invalid message: " + msg.getClass().getName()));
     }
   }
 
@@ -809,7 +813,8 @@ private void handleHttpMessage(HttpObject obj) {
       stream = responses.peekFirst();
     }
     if (stream == null) {
-      fail(new VertxException("Received HTTP message with no request in progress"));
+      invalidMessageHandler.handle(obj);
+      fail(new VertxException("Received an HTTP message with no request in progress: " + obj.getClass().getName()));
     } else if (obj instanceof io.netty.handler.codec.http.HttpResponse) {
       io.netty.handler.codec.http.HttpResponse response = (io.netty.handler.codec.http.HttpResponse) obj;
       HttpVersion version;
@@ -892,9 +897,7 @@ private void removeChannelHandlers() {
     // removing this codec might fire pending buffers in the HTTP decoder
     // this happens when the channel reads the HTTP response and the following data in a single buffer
     Handler<Object> prev = invalidMessageHandler;
-    invalidMessageHandler = msg -> {
-      ReferenceCountUtil.release(msg);
-    };
+    invalidMessageHandler = INVALID_MSG_HANDLER;
     try {
       pipeline.remove("codec");
     } finally {

src/main/java/io/vertx/core/http/impl/Http2ClientConnection.java

@@ -60,6 +60,11 @@ public Http2ClientConnection evictionHandler(Handler<Void> handler) {
     return this;
   }
 
+  @Override
+  public HttpClientConnection invalidMessageHandler(Handler<Object> handler) {
+    return this;
+  }
+
   @Override
   public Http2ClientConnection concurrencyChangeHandler(Handler<Long> handler) {
     concurrencyChangeHandler = handler;

src/main/java/io/vertx/core/http/impl/Http2UpgradeClientConnection.java

@@ -60,6 +60,7 @@ public class Http2UpgradeClientConnection implements HttpClientConnection {
   private Handler<Throwable> exceptionHandler;
   private Handler<Buffer> pingHandler;
   private Handler<Void> evictionHandler;
+  private Handler<Object> invalidMessageHandler;
   private Handler<Long> concurrencyChangeHandler;
   private Handler<Http2Settings> remoteSettingsHandler;
 
@@ -871,6 +872,15 @@ public HttpClientConnection evictionHandler(Handler<Void> handler) {
     return this;
   }
 
+  @Override
+  public HttpClientConnection invalidMessageHandler(Handler<Object> handler) {
+    if (current instanceof Http1xClientConnection) {
+      invalidMessageHandler = handler;
+    }
+    current.invalidMessageHandler(handler);
+    return this;
+  }
+
   @Override
   public HttpClientConnection concurrencyChangeHandler(Handler<Long> handler) {
     if (current instanceof Http1xClientConnection) {

src/main/java/io/vertx/core/http/impl/HttpClientConnection.java

@@ -43,6 +43,14 @@ public interface HttpClientConnection extends HttpConnection {
    */
   HttpClientConnection evictionHandler(Handler<Void> handler);
 
+  /**
+   * Set a {@code handler} called when the connection receives invalid messages.
+   *
+   * @param handler the handler
+   * @return a reference to this, so the API can be used fluently
+   */
+  HttpClientConnection invalidMessageHandler(Handler<Object> handler);
+
   /**
    * Set a {@code handler} called when the connection concurrency changes.
    * The handler is called with the new concurrency.

src/test/java/io/vertx/core/http/Http1xTest.java

@@ -5485,4 +5485,39 @@ public void testFailPendingRequestAllocationWhenConnectionIsClosed() throws Exce
     }
     await();
   }
+
+  @Test
+  public void testUnsolicitedMessagesAreTreatedAsInvalid() throws Exception {
+    NetServer server = vertx
+      .createNetServer()
+      .connectHandler(so -> {
+        so.write("HTTP/1.1 200 OK\r\n" +
+          "Content-Type: text/plain\r\n" +
+          "Content-Length: 11\r\n" +
+          "\r\n" +
+          "Hello World");
+      });
+    server
+      .listen(testAddress)
+      .toCompletionStage()
+      .toCompletableFuture()
+      .get(20, TimeUnit.SECONDS);
+
+    HttpClient client = vertx.httpClientBuilder().withConnectHandler(conn -> {
+      List<Object> invalidMessages = new ArrayList<>();
+      ((HttpClientConnection) conn).invalidMessageHandler(invalidMessages::add);
+      conn.exceptionHandler(err -> {
+      });
+      conn.closeHandler(v -> {
+        assertEquals(2, invalidMessages.size());
+        assertTrue(invalidMessages.get(0) instanceof io.netty.handler.codec.http.HttpResponse);
+        assertTrue(invalidMessages.get(1) instanceof io.netty.handler.codec.http.LastHttpContent);
+        testComplete();
+      });
+    }).build();
+
+    client.request(requestOptions);
+
+    await();
+  }
 }