http: pre-handshake WS client listeners + 1MB server WS receive cap

Two riders for the karate-ext single-port noVNC tunnel (D103):

- WsClientOptions.Builder.onMessage/onClose/onError register listeners
  in the WsClient constructor, BEFORE the handshake — for
  server-speaks-first protocols (RFB/VNC behind websockify) a listener
  added after connect() returns can miss the opening frame.
- HttpServerHandler's WebSocketServerHandshakerFactory now allows 1MB
  inbound frames (Netty defaults to 64k) — proxied streams (e.g. a VNC
  clipboard paste) can legitimately exceed the default.
- WsServerClientIntegrationTest: first end-to-end HttpServer+WsClient
  coverage — upgrade handshake, server-first greeting via the
  options-registered listener, >64k binary echo.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: ptrthomas

karate-core/src/main/java/io/karatelabs/http/HttpServerHandler.java

@@ -173,7 +173,9 @@ static boolean isWsUpgrade(FullHttpRequest req) {
 
     private void handleWsUpgrade(ChannelHandlerContext ctx, FullHttpRequest req, HttpRequest request) {
         String wsUrl = "ws://" + req.headers().get(HttpHeaderNames.HOST) + req.uri();
-        WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory(wsUrl, null, true);
+        // 1 MB receive cap (Netty defaults to 64k) — proxied streams (e.g. a VNC clipboard paste
+        // through the websockify tunnel) can legitimately exceed the default
+        WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory(wsUrl, null, true, HttpUtils.MEGABYTE);
         WebSocketServerHandshaker handshaker = factory.newHandshaker(req);
         if (handshaker == null) {
             WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());

karate-core/src/main/java/io/karatelabs/http/WsClient.java

@@ -110,6 +110,17 @@ private WsClient(WsClientOptions options) {
         this.callbackExecutor = options.getCallbackExecutor() != null
                 ? options.getCallbackExecutor()
                 : CALLBACK_EXECUTOR;
+        // options-registered listeners attach BEFORE the handshake so a server-speaks-first
+        // protocol (RFB/VNC) cannot race its opening frame past an empty listener list
+        if (options.getMessageListener() != null) {
+            messageListeners.add(options.getMessageListener());
+        }
+        if (options.getCloseListener() != null) {
+            closeListeners.add(options.getCloseListener());
+        }
+        if (options.getErrorListener() != null) {
+            errorListeners.add(options.getErrorListener());
+        }
     }
 
     private void doConnect() {

karate-core/src/main/java/io/karatelabs/http/WsClientOptions.java

@@ -31,6 +31,7 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
 
 /**
  * Configuration for WebSocket client connections using builder pattern.
@@ -46,6 +47,9 @@ public class WsClientOptions {
     private final boolean trustAllCerts;
     private final SslContext sslContext;
     private final ExecutorService callbackExecutor;
+    private final Consumer<WsFrame> messageListener;
+    private final Runnable closeListener;
+    private final Consumer<Throwable> errorListener;
 
     private WsClientOptions(Builder builder) {
         this.uri = builder.uri;
@@ -59,6 +63,9 @@ private WsClientOptions(Builder builder) {
         this.trustAllCerts = builder.trustAllCerts;
         this.sslContext = builder.sslContext;
         this.callbackExecutor = builder.callbackExecutor;
+        this.messageListener = builder.messageListener;
+        this.closeListener = builder.closeListener;
+        this.errorListener = builder.errorListener;
     }
 
     public static Builder builder(String uri) {
@@ -124,6 +131,18 @@ public ExecutorService getCallbackExecutor() {
         return callbackExecutor;
     }
 
+    public Consumer<WsFrame> getMessageListener() {
+        return messageListener;
+    }
+
+    public Runnable getCloseListener() {
+        return closeListener;
+    }
+
+    public Consumer<Throwable> getErrorListener() {
+        return errorListener;
+    }
+
     public static class Builder {
 
         private final URI uri;
@@ -135,6 +154,9 @@ public static class Builder {
         private boolean trustAllCerts = true;
         private SslContext sslContext;
         private ExecutorService callbackExecutor;
+        private Consumer<WsFrame> messageListener;
+        private Runnable closeListener;
+        private Consumer<Throwable> errorListener;
 
         private Builder(URI uri) {
             if (uri == null) {
@@ -196,6 +218,28 @@ public Builder callbackExecutor(ExecutorService executor) {
             return this;
         }
 
+        /**
+         * Register a message listener BEFORE the connection is opened. For server-speaks-first
+         * protocols (e.g. RFB/VNC behind websockify) a listener added after
+         * {@link WsClient#connect} returns can miss the server's opening frame.
+         */
+        public Builder onMessage(Consumer<WsFrame> listener) {
+            this.messageListener = listener;
+            return this;
+        }
+
+        /** Register a close listener before the connection is opened (see {@link #onMessage}). */
+        public Builder onClose(Runnable listener) {
+            this.closeListener = listener;
+            return this;
+        }
+
+        /** Register an error listener before the connection is opened (see {@link #onMessage}). */
+        public Builder onError(Consumer<Throwable> listener) {
+            this.errorListener = listener;
+            return this;
+        }
+
         public WsClientOptions build() {
             return new WsClientOptions(this);
         }

karate-core/src/test/java/io/karatelabs/http/WsServerClientIntegrationTest.java

@@ -0,0 +1,98 @@
+/*
+ * The MIT License
+ *
+ * Copyright 2026 Karate Labs Inc.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package io.karatelabs.http;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * End-to-end WebSocket coverage of {@link HttpServer} + {@link WsClient}: the upgrade handshake,
+ * a server-speaks-first opening frame caught by an options-registered listener (a listener added
+ * only after {@code connect()} returns can miss it), and a binary echo above Netty's 64k default
+ * receive cap (the handshaker is configured to 1 MB for proxied streams like a websockify tunnel).
+ */
+class WsServerClientIntegrationTest {
+
+    private static HttpServer server;
+
+    @BeforeAll
+    static void start() {
+        server = HttpServer.start(0,
+                req -> HttpResponse.notFound("ws only"),
+                null,
+                (req, connection) -> {
+                    // server speaks FIRST (the RFB/VNC shape), then echoes binary back
+                    connection.send("hello:" + req.getPath());
+                    connection.onBinary(connection::sendBytes);
+                });
+    }
+
+    @AfterAll
+    static void stop() {
+        server.stopAsync();
+    }
+
+    @Test
+    void serverGreetingAndLargeBinaryEchoRoundTrip() throws Exception {
+        byte[] big = new byte[200 * 1024]; // > the 64k Netty default the handshaker overrides
+        for (int i = 0; i < big.length; i++) {
+            big[i] = (byte) i;
+        }
+        CountDownLatch greeting = new CountDownLatch(1);
+        CountDownLatch echoed = new CountDownLatch(1);
+        List<WsFrame> frames = new CopyOnWriteArrayList<>();
+        WsClient client = WsClient.connect(WsClientOptions
+                .builder("ws://localhost:" + server.getPort() + "/tunnel")
+                .onMessage(frame -> {
+                    frames.add(frame);
+                    if (frame.isText()) {
+                        greeting.countDown();
+                    } else {
+                        echoed.countDown();
+                    }
+                })
+                .build());
+        try {
+            // the opening frame arrives even though no listener was added post-connect
+            assertTrue(greeting.await(5, TimeUnit.SECONDS), "server-first greeting frame");
+            assertEquals("hello:/tunnel", frames.get(0).getText());
+            client.send(big);
+            assertTrue(echoed.await(5, TimeUnit.SECONDS), "large binary echo");
+            byte[] back = frames.get(frames.size() - 1).getBytes();
+            assertTrue(Arrays.equals(big, back), "binary payload survives the round-trip");
+        } finally {
+            client.close();
+        }
+    }
+
+}