Add the sse-websocket module

2026-09-04 00:52:18 +05:30
parent e1f8aa7402
commit 5224afdad2
51 changed files with 2761 additions and 0 deletions
@@ -0,0 +1,30 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.test.web.servlet.client.RestTestClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** The property overrides the container default, in milliseconds, exactly as written. */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"dashboard.broadcast.enabled=false", "spring.mvc.async.request-timeout=5s"})
+class AsyncTimeoutPropertyTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void propertyOverridesTheContainer() {
+        RestTestClient client = RestTestClient.bindToServer()
+                .baseUrl("http://localhost:" + port).build();
+        client.get().uri("/diag/async-start").exchange().expectStatus().isOk();
+        String body = client.get().uri("/diag/async-timeout").exchange()
+                .expectStatus().isOk().expectBody(String.class).returnResult().getResponseBody();
+
+        System.out.println("=== spring.mvc.async.request-timeout=5s ===");
+        System.out.println(body);
+        assertThat(body).contains("\"effectiveAsyncTimeoutMs\":5000");
+    }
+}
@@ -0,0 +1,35 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.test.web.servlet.client.RestTestClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * With {@code spring.mvc.async.request-timeout} absent, the servlet container's default decides
+ * how long an SSE stream lives. This reads that number off the live {@code AsyncContext} rather
+ * than quoting it.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class AsyncTimeoutUnsetTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void containerDefaultApplies() {
+        RestTestClient client = RestTestClient.bindToServer()
+                .baseUrl("http://localhost:" + port).build();
+        client.get().uri("/diag/async-start").exchange().expectStatus().isOk();
+        String body = client.get().uri("/diag/async-timeout").exchange()
+                .expectStatus().isOk().expectBody(String.class).returnResult().getResponseBody();
+
+        System.out.println("=== spring.mvc.async.request-timeout ABSENT ===");
+        System.out.println(body);
+        assertThat(body).contains("\"spring.mvc.async.request-timeout\":\"<unset>\"");
+        assertThat(body).contains("\"effectiveAsyncTimeoutMs\":30000");
+    }
+}
@@ -0,0 +1,67 @@
+package com.ankurm.ssews;
+
+import java.time.Instant;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import jakarta.websocket.ContainerProvider;
+import jakarta.websocket.WebSocketContainer;
+import org.springframework.messaging.converter.JacksonJsonMessageConverter;
+import org.springframework.messaging.simp.stomp.StompSession;
+import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
+import org.springframework.web.socket.client.standard.StandardWebSocketClient;
+import org.springframework.web.socket.messaging.WebSocketStompClient;
+
+/**
+ * Sends one STOMP message of a given body size and reports whether it arrived.
+ *
+ * <p>The client's own buffers are raised to 1&nbsp;MB first. Without that the client hits the
+ * identical 8&nbsp;KB default and closes its own session, which looks exactly like the server
+ * rejecting the message &mdash; a false positive that cost a debugging round while writing this.
+ */
+final class SizeProbe {
+
+    private SizeProbe() {
+    }
+
+    static String attempt(int port, int size) throws Exception {
+        WebSocketContainer container = ContainerProvider.getWebSocketContainer();
+        container.setDefaultMaxTextMessageBufferSize(1024 * 1024);
+        WebSocketStompClient client = new WebSocketStompClient(new StandardWebSocketClient(container));
+        client.setMessageConverter(new JacksonJsonMessageConverter());
+        client.setInboundMessageSizeLimit(1024 * 1024);
+
+        StompSession session = client.connectAsync("ws://localhost:" + port + "/ws",
+                new StompSessionHandlerAdapter() { }).get();
+        BlockingQueue<ChatMessage> inbox =
+                StompTestSupport.subscribe(session, "/topic/room", ChatMessage.class);
+        Thread.sleep(200);
+
+        String result;
+        try {
+            session.send("/app/chat.send", new ChatMessage("bulk", "x".repeat(size), Instant.now()));
+            ChatMessage received = inbox.poll(4, TimeUnit.SECONDS);
+            result = received != null
+                    ? "delivered, body length " + received.text().length()
+                    : "NOT delivered, connection dropped";
+        }
+        catch (Exception ex) {
+            Throwable root = ex;
+            while (root.getCause() != null) {
+                root = root.getCause();
+            }
+            result = "NOT delivered: " + root.getMessage();
+        }
+        client.stop();
+        return result;
+    }
+
+    /** Binary search for the largest body that still arrives. */
+    static int[] ceiling(int port, int lo, int hi) throws Exception {
+        while (hi - lo > 64) {
+            int mid = (lo + hi) / 2;
+            if (attempt(port, mid).startsWith("delivered")) { lo = mid; } else { hi = mid; }
+        }
+        return new int[] {lo, hi};
+    }
+}
@@ -0,0 +1,71 @@
+package com.ankurm.ssews;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A raw-socket SSE client.
+ *
+ * <p>Deliberately not an HTTP client library: the article quotes the bytes on the wire, and any
+ * client that parses events for you hides exactly the thing being shown. It also lets a test
+ * abandon the socket mid-stream, which is how the disconnect case is reproduced.
+ */
+final class SseClient implements AutoCloseable {
+
+    private final Socket socket;
+    private final BufferedReader reader;
+    private final List<String> lines = new ArrayList<>();
+
+    SseClient(int port, String path) throws Exception {
+        this(port, path, null);
+    }
+
+    SseClient(int port, String path, String lastEventId) throws Exception {
+        this.socket = new Socket("127.0.0.1", port);
+        this.socket.setSoTimeout(15_000);
+        StringBuilder req = new StringBuilder()
+                .append("GET ").append(path).append(" HTTP/1.1\r\n")
+                .append("Host: 127.0.0.1:").append(port).append("\r\n")
+                .append("Accept: text/event-stream\r\n");
+        if (lastEventId != null) {
+            req.append("Last-Event-ID: ").append(lastEventId).append("\r\n");
+        }
+        req.append("Connection: close\r\n\r\n");
+        OutputStream out = socket.getOutputStream();
+        out.write(req.toString().getBytes(StandardCharsets.UTF_8));
+        out.flush();
+        this.reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
+    }
+
+    /** Read until the stream closes or {@code max} lines have arrived. Records everything. */
+    List<String> read(int max) throws Exception {
+        String line;
+        while (lines.size() < max && (line = reader.readLine()) != null) {
+            lines.add(line);
+        }
+        return List.copyOf(lines);
+    }
+
+    /** Read everything until the server closes the connection. */
+    List<String> readToEnd() throws Exception {
+        String line;
+        while ((line = reader.readLine()) != null) {
+            lines.add(line);
+        }
+        return List.copyOf(lines);
+    }
+
+    List<String> lines() {
+        return List.copyOf(lines);
+    }
+
+    @Override
+    public void close() throws Exception {
+        socket.close();
+    }
+}
@@ -0,0 +1,95 @@
+package com.ankurm.ssews;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.apache.tomcat.util.threads.ThreadPoolExecutor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.boot.tomcat.TomcatWebServer;
+import org.springframework.boot.web.server.context.WebServerApplicationContext;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Forty open SSE streams on a ten-thread Tomcat.
+ *
+ * <p>This is the test that answers the objection people raise first &mdash; "doesn't SSE pin a
+ * thread per client?" It does not. {@code SseEmitter} puts the request into asynchronous mode,
+ * the request thread returns to the pool, and the response stays open with no thread attached to
+ * it. Forty concurrent streams on a pool of ten is the cheapest way to show that; the same test
+ * with a blocking handler deadlocks at the eleventh client.
+ *
+ * <p>What SSE does cost is a <em>socket</em> per stream, plus one serialise-and-write per stream
+ * per broadcast, both of which the transcript makes visible.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"dashboard.broadcast.enabled=false", "server.tomcat.threads.max=10"})
+class SseConcurrencyTest {
+
+    private static final int CLIENTS = 40;
+
+    @LocalServerPort
+    int port;
+
+    @Autowired
+    EmitterRegistry registry;
+
+    @Autowired
+    WebServerApplicationContext context;
+
+    @Test
+    void fortyStreamsOnTenThreads() throws Exception {
+        List<SseClient> clients = new ArrayList<>();
+        try {
+            for (int i = 0; i < CLIENTS; i++) {
+                clients.add(new SseClient(port, "/sse/metrics"));
+            }
+            for (int i = 0; i < 200 && registry.open() < CLIENTS; i++) {
+                Thread.sleep(25);
+            }
+
+            // Read the pool size off THIS context's own connector. Counting threads by name
+            // does not work here: other @SpringBootTest contexts in the same JVM have their own
+            // http-nio pools, and with a random port Tomcat names its threads http-nio-auto-N,
+            // not http-nio-<actual port>. That mismatch made this assertion pass alone and fail
+            // in the full suite.
+            TomcatWebServer server = (TomcatWebServer) context.getWebServer();
+            ThreadPoolExecutor pool = (ThreadPoolExecutor) server.getTomcat().getConnector()
+                    .getProtocolHandler().getExecutor();
+
+            System.out.println("=== server.tomcat.threads.max=10, clients=" + CLIENTS + " ===");
+            System.out.println("emitters open        : " + registry.open());
+            System.out.println("connector pool size  : " + pool.getPoolSize()
+                    + " (max " + pool.getMaximumPoolSize() + ")");
+            System.out.println("connector active     : " + pool.getActiveCount());
+
+            assertThat(registry.open()).isEqualTo(CLIENTS);
+            assertThat(pool.getMaximumPoolSize()).isEqualTo(10);
+            assertThat(pool.getPoolSize()).isLessThanOrEqualTo(10);
+
+            int delivered = registry.broadcast("metric",
+                    new MetricSnapshot(1, "node-a", 0.5, 300, Instant.now()));
+            System.out.println("one broadcast reached: " + delivered + " streams");
+            assertThat(delivered).isEqualTo(CLIENTS);
+
+            int received = 0;
+            for (SseClient c : clients) {
+                if (String.join("\n", c.read(12)).contains("event:metric")) {
+                    received++;
+                }
+            }
+            System.out.println("clients that read it : " + received);
+            assertThat(received).isEqualTo(CLIENTS);
+        }
+        finally {
+            for (SseClient c : clients) {
+                try { c.close(); } catch (Exception ignored) { }
+            }
+        }
+    }
+}
@@ -0,0 +1,70 @@
+package com.ankurm.ssews;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * {@code .data(object)} and {@code .data(object, APPLICATION_JSON)} are not the same call.
+ * This test prints both results side by side.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class SseDataConversionTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void mediaTypeDecidesTheConverter() throws Exception {
+        List<String> lines;
+        try (SseClient client = new SseClient(port, "/sse/payload")) {
+            lines = client.readToEnd();
+        }
+
+        String withJson = dataAfter(lines, "event:with-json");
+        String noMediaType = dataAfter(lines, "event:no-media-type");
+
+        System.out.println("=== SseEmitter.event().data(record, APPLICATION_JSON) ===");
+        System.out.println(withJson);
+        System.out.println("=== SseEmitter.event().data(record)  [no media type] ===");
+        System.out.println(noMediaType);
+
+        assertThat(withJson).startsWith("{").contains("\"seq\":7").contains("\"host\":\"node-a\"");
+
+        // The finding: for a POJO they are IDENTICAL. No media type does not mean toString().
+        // send() asks the configured HttpMessageConverters which one canWrite(type, null), and
+        // for a record the first (and only) answer is the Jackson converter.
+        System.out.println("=== POJO: identical? " + withJson.equals(noMediaType) + " ===");
+        assertThat(noMediaType).isEqualTo(withJson);
+
+        String stringPlain = dataAfter(lines, "event:string-plain");
+        String stringJson = dataAfter(lines, "event:string-json");
+        System.out.println("=== String payload, no media type   -> " + stringPlain);
+        System.out.println("=== String payload, APPLICATION_JSON -> " + stringJson);
+
+        // The finding, and it is the opposite of what I expected when writing this test: for a
+        // String the media type argument changes NOTHING. StringHttpMessageConverter supports
+        // MediaType.ALL and sits ahead of the Jackson converter in the list, so it claims the
+        // write even when you ask for application/json. There is no way to make send() quote a
+        // String -- which is good news if you are streaming pre-rendered JSON, and a trap if you
+        // expected a String field to be escaped for you.
+        assertThat(stringPlain).isEqualTo("{\"already\":\"json\"}");
+        assertThat(stringJson).isEqualTo(stringPlain);
+    }
+
+    private static String dataAfter(List<String> lines, String eventLine) {
+        int i = lines.indexOf(eventLine);
+        assertThat(i).as("event line %s present", eventLine).isGreaterThanOrEqualTo(0);
+        for (int j = i; j < lines.size(); j++) {
+            if (lines.get(j).startsWith("data:")) {
+                return lines.get(j).substring("data:".length());
+            }
+        }
+        throw new AssertionError("no data line after " + eventLine);
+    }
+}
@@ -0,0 +1,61 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import java.time.Instant;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * A disconnected SSE client is discovered by a failed write, not by an event.
+ *
+ * <p>This test opens a stream, walks away from the socket, and then counts how many broadcasts
+ * it takes before the server notices. The number is not one, and that is the whole problem: for
+ * that many ticks the server serialised a payload and wrote it into a socket nobody was reading.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class SseDisconnectTest {
+
+    @LocalServerPort
+    int port;
+
+    @Autowired
+    EmitterRegistry registry;
+
+    @Test
+    void deadClientIsFoundOnTheNextWriteOrTheOneAfter() throws Exception {
+        SseClient client = new SseClient(port, "/sse/metrics");
+        // Wait for the emitter to be registered.
+        for (int i = 0; i < 100 && registry.open() == 0; i++) {
+            Thread.sleep(20);
+        }
+        assertThat(registry.open()).isEqualTo(1);
+
+        long failedBefore = registry.sendsFailed();
+        client.close();   // the browser tab closes
+
+        int broadcastsUntilNoticed = 0;
+        for (int i = 1; i <= 50 && registry.open() > 0; i++) {
+            registry.broadcast("metric", new MetricSnapshot(i, "node-a", 0.1, 256, Instant.now()));
+            broadcastsUntilNoticed = i;
+            Thread.sleep(50);
+        }
+
+        System.out.println("=== client closed the socket, then: ===");
+        System.out.println("broadcasts before the write failed : " + broadcastsUntilNoticed);
+        System.out.println("registry.sendsOk                   : " + registry.sendsOk());
+        System.out.println("registry.sendsFailed               : " + failedBefore + " -> " + registry.sendsFailed());
+        System.out.println("registry.open                      : " + registry.open());
+
+        assertThat(registry.open()).isZero();
+        assertThat(registry.sendsFailed()).isGreaterThan(failedBefore);
+        // The interesting assertion: it took more than one write. TCP accepted the first
+        // payloads into the socket buffer and only reported the reset afterwards.
+        System.out.println("=== writes that appeared to succeed after the client was gone: "
+                + (broadcastsUntilNoticed - 1) + " ===");
+    }
+}
@@ -0,0 +1,60 @@
+package com.ankurm.ssews;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * What a client sees when the async request times out mid-stream.
+ *
+ * <p>The answer is the point: nothing. No error event, no status code, no trailer &mdash; the
+ * response was committed with 200 when the first byte went out, so the only thing left to do is
+ * close the socket. A browser's EventSource treats that as a normal disconnect and reconnects,
+ * which is why an SSE endpoint with a 30-second container timeout looks like it works.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class SseTimeoutTest {
+
+    @LocalServerPort
+    int port;
+
+    @Autowired
+    EmitterRegistry registry;
+
+    @Test
+    void timeoutIsInvisibleToTheClient() throws Exception {
+        long before = registry.timedOut();
+
+        List<String> lines;
+        long start = System.nanoTime();
+        try (SseClient client = new SseClient(port, "/sse/silent?timeoutMs=1200")) {
+            lines = client.readToEnd();
+        }
+        long elapsedMs = (System.nanoTime() - start) / 1_000_000;
+
+        System.out.println("=== GET /sse/silent?timeoutMs=1200 : whole response ===");
+        lines.forEach(l -> System.out.println("| " + l));
+        System.out.println("=== stream ended after ~" + elapsedMs + " ms ===");
+
+        String whole = String.join("\n", lines);
+        assertThat(whole).contains("HTTP/1.1 200");
+        assertThat(whole).contains("data:then silence");
+
+        // Nothing marks the timeout. No 503, no error event, no "AsyncRequestTimeoutException".
+        assertThat(whole).doesNotContain("503");
+        assertThat(whole).doesNotContain("event:error");
+        assertThat(whole).doesNotContainIgnoringCase("timeout");
+
+        // Server-side it is not invisible: onTimeout ran.
+        Thread.sleep(300);
+        System.out.println("=== registry.timedOut(): " + before + " -> " + registry.timedOut() + " ===");
+        assertThat(registry.timedOut()).isGreaterThan(before);
+        assertThat(elapsedMs).isBetween(900L, 6000L);
+    }
+}
@@ -0,0 +1,53 @@
+package com.ankurm.ssews;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * What SSE actually looks like on the wire. Everything in the article's "the format" section is
+ * a quote from this test's output.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class SseWireFormatTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void wireFormat() throws Exception {
+        List<String> lines;
+        try (SseClient client = new SseClient(port, "/sse/raw")) {
+            lines = client.readToEnd();
+        }
+
+        System.out.println("=== GET /sse/raw : raw bytes, one line per row ===");
+        lines.forEach(l -> System.out.println("| " + l));
+
+        String head = String.join("\n", lines);
+
+        // Content-Type is text/event-stream, and there is NO Content-Length: the response is
+        // chunked because its length is unknowable when the headers are written.
+        assertThat(head).contains("HTTP/1.1 200");
+        assertThat(head.toLowerCase()).contains("content-type: text/event-stream");
+        assertThat(head.toLowerCase()).doesNotContain("content-length:");
+        assertThat(head.toLowerCase()).contains("transfer-encoding: chunked");
+
+        // The four field names, and the comment line that keeps proxies from reaping an idle
+        // stream. A line starting with ':' is a comment and is discarded by the client.
+        assertThat(lines).anyMatch(l -> l.equals(":stream open"));
+        assertThat(lines).anyMatch(l -> l.equals("id:1"));
+        assertThat(lines).anyMatch(l -> l.equals("event:metric"));
+        assertThat(lines).anyMatch(l -> l.equals("retry:3000"));
+        assertThat(lines).anyMatch(l -> l.startsWith("data:{"));
+
+        // Multi-line data becomes SEVERAL data: lines; the client rejoins them with "\n".
+        assertThat(lines).contains("data:line one");
+        assertThat(lines).contains("data:line two");
+    }
+}
@@ -0,0 +1,74 @@
+package com.ankurm.ssews;
+
+import java.time.Instant;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.messaging.simp.stomp.StompSession;
+import org.springframework.web.socket.messaging.WebSocketStompClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Two clients, one room. Everything a chat needs that SSE cannot do: the client speaks, and the
+ * server knows who is connected.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class StompChatTest {
+
+    @LocalServerPort
+    int port;
+
+    @Autowired
+    PresenceListener presence;
+
+    @Test
+    void twoClientsInOneRoom() throws Exception {
+        presence.clear();
+        WebSocketStompClient client = StompTestSupport.client();
+
+        StompSession alice = StompTestSupport.connect(client, port);
+        StompSession bob = StompTestSupport.connect(client, port);
+
+        BlockingQueue<ChatMessage> aliceInbox =
+                StompTestSupport.subscribe(alice, "/topic/room", ChatMessage.class);
+        BlockingQueue<ChatMessage> bobInbox =
+                StompTestSupport.subscribe(bob, "/topic/room", ChatMessage.class);
+        Thread.sleep(300);
+
+        alice.send("/app/chat.send", new ChatMessage("alice", "is this thing on?", Instant.now()));
+
+        ChatMessage atAlice = aliceInbox.poll(5, TimeUnit.SECONDS);
+        ChatMessage atBob = bobInbox.poll(5, TimeUnit.SECONDS);
+
+        System.out.println("=== SEND /app/chat.send -> @SendTo(\"/topic/room\") ===");
+        System.out.println("alice received : " + atAlice);
+        System.out.println("bob   received : " + atBob);
+
+        // The sender gets its own message back. There is no echo suppression: a topic is a fan-out
+        // to every subscriber, and the sender is one of them.
+        assertThat(atAlice).isNotNull();
+        assertThat(atBob).isNotNull();
+        assertThat(atAlice.text()).isEqualTo("is this thing on?");
+        assertThat(atAlice).isEqualTo(atBob);
+
+        Thread.sleep(300);
+        System.out.println("=== presence events the server saw ===");
+        presence.events().forEach(e -> System.out.println("| " + e));
+        assertThat(presence.events()).anyMatch(e -> e.startsWith("CONNECTED"));
+        assertThat(presence.events()).anyMatch(e -> e.contains("SUBSCRIBE") && e.contains("/topic/room"));
+
+        alice.disconnect();
+        bob.disconnect();
+        Thread.sleep(500);
+        presence.events().stream().filter(e -> e.startsWith("DISCONNECT"))
+                .forEach(e -> System.out.println("| " + e));
+        assertThat(presence.events()).anyMatch(e -> e.startsWith("DISCONNECT"));
+        client.stop();
+    }
+}
@@ -0,0 +1,76 @@
+package com.ankurm.ssews;
+
+import java.time.Instant;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.messaging.simp.stomp.StompSession;
+import org.springframework.web.socket.messaging.WebSocketStompClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The two routing rules that eat an afternoon each.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class StompRoutingTest {
+
+    @LocalServerPort
+    int port;
+
+    /**
+     * A handler with no {@code @SendTo} is not private. Its return value goes to the broker
+     * prefix plus the mapping.
+     */
+    @Test
+    void defaultDestinationIsTopicPlusMapping() throws Exception {
+        WebSocketStompClient client = StompTestSupport.client();
+        StompSession session = StompTestSupport.connect(client, port);
+
+        BlockingQueue<ChatMessage> guessed =
+                StompTestSupport.subscribe(session, "/topic/chat.echo", ChatMessage.class);
+        Thread.sleep(300);
+
+        session.send("/app/chat.echo", new ChatMessage("mallory", "who can see this?", Instant.now()));
+
+        ChatMessage received = guessed.poll(5, TimeUnit.SECONDS);
+        System.out.println("=== @MessageMapping(\"/chat.echo\") with NO @SendTo ===");
+        System.out.println("subscribed to /topic/chat.echo, received: " + received);
+        assertThat(received).isNotNull();
+        assertThat(received.text()).isEqualTo("who can see this?");
+        client.stop();
+    }
+
+    /**
+     * Sending straight to a broker destination bypasses every controller. No handler runs, no
+     * validation runs, and nothing reports that anything was skipped &mdash; subscribers just get
+     * whatever the client sent.
+     */
+    @Test
+    void sendingToTopicSkipsTheController() throws Exception {
+        WebSocketStompClient client = StompTestSupport.client();
+        StompSession session = StompTestSupport.connect(client, port);
+
+        BlockingQueue<ChatMessage> inbox =
+                StompTestSupport.subscribe(session, "/topic/room", ChatMessage.class);
+        Thread.sleep(300);
+
+        // Note the destination: /topic/room, NOT /app/chat.send. The controller never sees it.
+        ChatMessage forged = new ChatMessage("not-checked-by-anyone", "unvalidated", Instant.parse("2000-01-01T00:00:00Z"));
+        session.send("/topic/room", forged);
+
+        ChatMessage received = inbox.poll(5, TimeUnit.SECONDS);
+        System.out.println("=== SEND straight to /topic/room ===");
+        System.out.println("received: " + received);
+        assertThat(received).isNotNull();
+        // The controller would have replaced the timestamp with Instant.now(). It did not run.
+        assertThat(received.at()).isEqualTo(Instant.parse("2000-01-01T00:00:00Z"));
+        assertThat(received.from()).isEqualTo("not-checked-by-anyone");
+        System.out.println("=== the controller's Instant.now() rewrite did NOT happen ===");
+        client.stop();
+    }
+}
@@ -0,0 +1,39 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Where a STOMP message really stops being deliverable, with nothing configured.
+ *
+ * <p>The documented knob is {@code WebSocketTransportRegistration.setMessageSizeLimit}, 64&nbsp;KB
+ * by default and set explicitly in {@link WebSocketConfig}. The limit that actually fires is the
+ * servlet container's WebSocket buffer, and it fires as a connection close (WebSocket status
+ * 1009) rather than as an error the application can see &mdash; so raising Spring's limit alone
+ * changes nothing at all.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class StompSizeLimitDefaultBufferTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void theCeilingIsWellUnderSpringsSixtyFourKilobytes() throws Exception {
+        System.out.println("=== container buffer: Tomcat default (8192, see /diag) ===");
+        System.out.println("  8 KB body -> " + SizeProbe.attempt(port, 8 * 1024));
+        System.out.println(" 16 KB body -> " + SizeProbe.attempt(port, 16 * 1024));
+        System.out.println(" 60 KB body -> " + SizeProbe.attempt(port, 60 * 1024));
+
+        int[] edge = SizeProbe.ceiling(port, 1024, 64 * 1024);
+        System.out.println("largest body delivered   : " + edge[0] + " bytes");
+        System.out.println("smallest body rejected   : " + edge[1] + " bytes");
+
+        // Measured, not remembered. Spring's messageSizeLimit is 65 536 and is never reached.
+        assertThat(edge[0]).isBetween(15_000, 17_000);
+    }
+}
@@ -0,0 +1,33 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * With the container buffer at 256&nbsp;KB, Spring's own 64&nbsp;KB limit is finally the one that
+ * decides &mdash; and it fails differently. The container's refusal is a bare close; Spring's is
+ * a STOMP {@code ERROR} frame, logged, with the byte count in it. Two limits, two symptoms, and
+ * only the second one tells you what happened.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"dashboard.broadcast.enabled=false", "spring.profiles.active=bigframes"})
+class StompSizeLimitSpringLimitTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void springsLimitDecidesOnceTheContainerIsOutOfTheWay() throws Exception {
+        System.out.println("=== bigframes profile: container buffer 256 KB, Spring limit 64 KB ===");
+        System.out.println(" 16 KB body -> " + SizeProbe.attempt(port, 16 * 1024));
+        System.out.println(" 60 KB body -> " + SizeProbe.attempt(port, 60 * 1024));
+        System.out.println(" 80 KB body -> " + SizeProbe.attempt(port, 80 * 1024));
+        System.out.println("200 KB body -> " + SizeProbe.attempt(port, 200 * 1024));
+
+        assertThat(SizeProbe.attempt(port, 60 * 1024)).startsWith("delivered");
+        assertThat(SizeProbe.attempt(port, 80 * 1024)).doesNotStartWith("delivered");
+    }
+}
@@ -0,0 +1,31 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The same binary search with the container buffer raised, which is what turns "raise the
+ * buffer" into a rule rather than folklore. Run with {@code -Dwsbuf=<bytes>} to move it; the
+ * committed transcript covers 8&nbsp;KB (untouched), 16&nbsp;KB, 32&nbsp;KB and 64&nbsp;KB.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"dashboard.broadcast.enabled=false",
+                "spring.profiles.active=bigframes", "ws.buffer-bytes=${wsbuf:32768}"})
+class StompSizeLimitTunedBufferTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void ceilingTracksTheContainerBuffer() throws Exception {
+        String buf = System.getProperty("wsbuf", "32768");
+        System.out.println("=== container buffer set to " + buf + " ===");
+        int[] edge = SizeProbe.ceiling(port, 1024, 200 * 1024);
+        System.out.println("largest body delivered   : " + edge[0] + " bytes");
+        System.out.println("smallest body rejected   : " + edge[1] + " bytes");
+        assertThat(edge[0]).isGreaterThan(30_000);
+    }
+}
@@ -0,0 +1,58 @@
+package com.ankurm.ssews;
+
+import java.lang.reflect.Type;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.springframework.messaging.converter.JacksonJsonMessageConverter;
+import org.springframework.messaging.simp.stomp.StompFrameHandler;
+import org.springframework.messaging.simp.stomp.StompHeaders;
+import org.springframework.messaging.simp.stomp.StompSession;
+import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
+import org.springframework.web.socket.client.standard.StandardWebSocketClient;
+import org.springframework.web.socket.messaging.WebSocketStompClient;
+
+/**
+ * A real STOMP-over-WebSocket client, so the chat tests exercise the same code path a browser
+ * does &mdash; handshake, CONNECT/CONNECTED, SUBSCRIBE, SEND, MESSAGE.
+ *
+ * <p>Note the converter: {@link JacksonJsonMessageConverter} is the Jackson&nbsp;3 one.
+ * {@code MappingJackson2MessageConverter} is still on the classpath and is the Jackson&nbsp;2
+ * one; under Boot 4 a "2" in a Spring class name means the <em>previous</em> Jackson. Picking
+ * the wrong one here fails on the {@code Instant} field, exactly as it does for Kafka and AMQP
+ * in the sibling modules.
+ */
+final class StompTestSupport {
+
+    private StompTestSupport() {
+    }
+
+    static WebSocketStompClient client() {
+        WebSocketStompClient client = new WebSocketStompClient(new StandardWebSocketClient());
+        client.setMessageConverter(new JacksonJsonMessageConverter());
+        return client;
+    }
+
+    static StompSession connect(WebSocketStompClient client, int port) throws Exception {
+        return client.connectAsync("ws://localhost:" + port + "/ws",
+                new StompSessionHandlerAdapter() { }).get();
+    }
+
+    /** Subscribe and collect frames of one payload type into a queue the test can poll. */
+    static <T> BlockingQueue<T> subscribe(StompSession session, String destination, Class<T> type) {
+        BlockingQueue<T> queue = new LinkedBlockingQueue<>();
+        session.subscribe(destination, new StompFrameHandler() {
+            @Override
+            public Type getPayloadType(StompHeaders headers) {
+                return type;
+            }
+
+            @Override
+            @SuppressWarnings("unchecked")
+            public void handleFrame(StompHeaders headers, Object payload) {
+                queue.add((T) payload);
+            }
+        });
+        return queue;
+    }
+}
@@ -0,0 +1,81 @@
+package com.ankurm.ssews;
+
+import java.time.Instant;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.messaging.simp.stomp.StompSession;
+import org.springframework.web.socket.messaging.WebSocketStompClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Point-to-point delivery, and the thing that makes it work: {@code /user/**} is not a real
+ * destination. The broker rewrites it per session, so two clients subscribed to the same string
+ * are subscribed to different queues.
+ *
+ * <p>Without Spring Security in the picture, the "user" is the STOMP session id. That is enough
+ * to demonstrate the routing and is also a trap worth naming: with no {@code Principal}, a user
+ * destination is session-scoped, so a second browser tab is a different user and a reconnect
+ * loses the mailbox.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class StompUserDestinationTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void onlyTheAddressedSessionReceives() throws Exception {
+        WebSocketStompClient client = StompTestSupport.client();
+
+        StompSession alice = StompTestSupport.connect(client, port);
+        StompSession bob = StompTestSupport.connect(client, port);
+
+        BlockingQueue<ChatMessage> aliceInbox =
+                StompTestSupport.subscribe(alice, "/user/queue/whisper", ChatMessage.class);
+        BlockingQueue<ChatMessage> bobInbox =
+                StompTestSupport.subscribe(bob, "/user/queue/whisper", ChatMessage.class);
+        Thread.sleep(400);
+
+        // Discovering Alice's session id is itself the lesson. StompSession.getSessionId() on
+        // the CLIENT returns the client's own identifier, which is NOT the id the server keeps
+        // in SimpMessageHeaderAccessor and NOT the one the user destination is built from.
+        // Whispering to it delivers nothing, silently. The server-side id has to come from the
+        // server: /app/chat.send echoes it back when "from" is null.
+        BlockingQueue<ChatMessage> aliceRoom =
+                StompTestSupport.subscribe(alice, "/topic/room", ChatMessage.class);
+        Thread.sleep(200);
+        alice.send("/app/chat.send", new ChatMessage(null, "who am i?", Instant.now()));
+        ChatMessage identity = aliceRoom.poll(5, TimeUnit.SECONDS);
+        assertThat(identity).isNotNull();
+        String aliceServerSession = identity.from();
+
+        System.out.println("alice.getSessionId()  [client side] : " + alice.getSessionId());
+        System.out.println("server-side session id             : " + aliceServerSession);
+        assertThat(aliceServerSession).isNotEqualTo(alice.getSessionId());
+
+        bob.send("/app/chat.whisper",
+                new ChatMessage(aliceServerSession, "just for you", Instant.now()));
+
+        ChatMessage atAlice = aliceInbox.poll(5, TimeUnit.SECONDS);
+        ChatMessage atBob = bobInbox.poll(1500, TimeUnit.MILLISECONDS);
+
+        System.out.println("=== convertAndSendToUser(aliceSession, \"/queue/whisper\", ..) ===");
+        System.out.println("both subscribed to the SAME string : /user/queue/whisper");
+        System.out.println("alice received : " + atAlice);
+        System.out.println("bob   received : " + atBob);
+
+        assertThat(atAlice).isNotNull();
+        assertThat(atAlice.text()).isEqualTo("just for you");
+        assertThat(atBob).as("the other session must not see it").isNull();
+
+        alice.disconnect();
+        bob.disconnect();
+        client.stop();
+    }
+}
@@ -0,0 +1,30 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.test.web.servlet.client.RestTestClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reads the container's own WebSocket buffer sizes rather than quoting a default from memory.
+ * Nothing in this application sets them.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class WebSocketContainerDefaultsTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void printThem() {
+        String body = RestTestClient.bindToServer().baseUrl("http://localhost:" + port).build()
+                .get().uri("/diag").exchange().expectStatus().isOk()
+                .expectBody(String.class).returnResult().getResponseBody();
+        System.out.println("=== /diag with no bigframes profile ===");
+        System.out.println(body);
+        assertThat(body).contains("\"ws.container.defaultMaxTextMessageBufferSize\":8192");
+    }
+}
@@ -0,0 +1,27 @@
+package com.ankurm.ssews;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.test.web.servlet.client.RestTestClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** The {@code bigframes} profile's ServletServerContainerFactoryBean, verified to have landed. */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = {"dashboard.broadcast.enabled=false", "spring.profiles.active=bigframes"})
+class WebSocketContainerRaisedTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void printThem() {
+        String body = RestTestClient.bindToServer().baseUrl("http://localhost:" + port).build()
+                .get().uri("/diag").exchange().expectStatus().isOk()
+                .expectBody(String.class).returnResult().getResponseBody();
+        System.out.println("=== /diag with the bigframes profile ===");
+        System.out.println(body);
+        assertThat(body).contains("\"ws.container.defaultMaxTextMessageBufferSize\":262144");
+    }
+}
@@ -0,0 +1,76 @@
+package com.ankurm.ssews;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * What {@code registry.addEndpoint("/ws")} with no {@code setAllowedOrigins} actually permits.
+ *
+ * <p>This matters because the WebSocket handshake is a plain HTTP GET, and the browser's
+ * same-origin policy does <em>not</em> apply to it: a page on any site can open a WebSocket to
+ * your server, with the user's cookies attached, unless the server checks {@code Origin}. Spring
+ * checks it by default. A non-browser client sends no {@code Origin} at all and is allowed
+ * through, which is why the Java tests in this module connect without ceremony.
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+        properties = "dashboard.broadcast.enabled=false")
+class WebSocketHandshakeOriginTest {
+
+    @LocalServerPort
+    int port;
+
+    @Test
+    void originIsCheckedByDefault() throws Exception {
+        System.out.println("=== GET /ws handshake, varying only the Origin header ===");
+        String noOrigin = handshake(null);
+        String sameOrigin = handshake("http://localhost:" + port);
+        String foreignOrigin = handshake("https://evil.example.com");
+
+        System.out.println("no Origin header          -> " + noOrigin);
+        System.out.println("Origin: http://localhost  -> " + sameOrigin);
+        System.out.println("Origin: https://evil...   -> " + foreignOrigin);
+
+        assertThat(noOrigin).startsWith("HTTP/1.1 101");
+        assertThat(sameOrigin).startsWith("HTTP/1.1 101");
+        assertThat(foreignOrigin).startsWith("HTTP/1.1 403");
+    }
+
+    private String handshake(String origin) throws Exception {
+        try (Socket socket = new Socket("127.0.0.1", port)) {
+            socket.setSoTimeout(10_000);
+            StringBuilder req = new StringBuilder()
+                    .append("GET /ws HTTP/1.1\r\n")
+                    .append("Host: localhost:").append(port).append("\r\n")
+                    .append("Upgrade: websocket\r\n")
+                    .append("Connection: Upgrade\r\n")
+                    .append("Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n")
+                    .append("Sec-WebSocket-Version: 13\r\n");
+            if (origin != null) {
+                req.append("Origin: ").append(origin).append("\r\n");
+            }
+            req.append("\r\n");
+            OutputStream out = socket.getOutputStream();
+            out.write(req.toString().getBytes(StandardCharsets.UTF_8));
+            out.flush();
+            BufferedReader in = new BufferedReader(
+                    new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
+            List<String> lines = new ArrayList<>();
+            String line;
+            while ((line = in.readLine()) != null && !line.isEmpty()) {
+                lines.add(line);
+            }
+            return lines.isEmpty() ? "<no response>" : lines.get(0).trim();
+        }
+    }
+}