diff --git a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java index b135fdeb1..e820ef001 100644 --- a/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java +++ b/src/main/java/io/vertx/core/http/impl/Http1xClientConnection.java @@ -67,6 +67,10 @@ class Http1xClientConnection extends Http1xConnectionBase impleme private static final Logger log = LoggerFactory.getLogger(Http1xClientConnection.class); + private static final Handler INVALID_MSG_HANDLER = msg -> { + throw new IllegalStateException("Invalid object " + msg); + }; + private final ConnectionListener listener; private final HttpClientImpl client; private final HttpClientOptions options; @@ -78,7 +82,8 @@ class Http1xClientConnection extends Http1xConnectionBase impleme private StreamImpl head; private StreamImpl requestInProgress; - + + private Handler invalidMessageHandler = INVALID_MSG_HANDLER; private boolean close; private Promise netSocketPromise; private int keepAliveTimeout; @@ -509,7 +514,7 @@ class Http1xClientConnection extends Http1xConnectionBase impleme } else if (msg instanceof WebSocketFrame) { handleWsFrame((WebSocketFrame) msg); } else { - throw new IllegalStateException("Invalid object " + msg); + invalidMessageHandler.handle(msg); } } @@ -568,6 +573,11 @@ class Http1xClientConnection extends Http1xConnectionBase impleme if (inflater != null) { pipeline.remove(inflater); } + + // 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 + Deque pending = new ArrayDeque<>(); + invalidMessageHandler = pending::add; pipeline.remove("codec"); // replace the old handler with one that handle plain sockets @@ -583,11 +593,13 @@ class Http1xClientConnection extends Http1xConnectionBase impleme socket.metric(metric()); pipeline.replace("handler", "handler", VertxHandler.create(socket)); - // 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 - // Handle back response promise.complete(socket); + + // Redeliver pending messages + for (Object msg : pending) { + pipeline.fireChannelRead(msg); + } } else { promise.fail("Server responded with " + response.statusCode() + " code instead of 200"); }