diff --git a/README.md b/README.md index cfc0b9c..6479ec9 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,8 @@ module's `scripts/run-all.sh`, never typed by hand. | [`kafka-basics/`](kafka-basics/README.md) | [Spring Boot 4.1 and Apache Kafka: Producer, Consumer and Serialisation from Scratch](https://ankurm.com/spring-boot-4-1-kafka-producer-consumer-serialisation/) | The on-ramp: what the starter gives you, the two Jackson serializer families, where a key lands and why, and which defaults are Kafka's rather than Spring's | | [`kafka-error-handling/`](kafka-error-handling/README.md) | [Kafka Error Handling with Spring Kafka 4.1: DLT, Retry Topics and Poison Pills](https://ankurm.com/spring-kafka-4-1-error-handling-dlt-retry-topics/) | What the default error handler really does, why a poison pill stops a partition, and what blocking retries cost that retry topics do not | | [`rabbitmq/`](rabbitmq/README.md) | [Spring Boot and RabbitMQ: Exchanges, Queues, Bindings and a Working Dead-Letter Queue](https://ankurm.com/spring-boot-rabbitmq-exchanges-dead-letter-queue/) | All four exchange types against a real broker, manual acknowledgement, and a dead-letter path exercised through both rejection and TTL expiry | +| [`sse-websocket/`](sse-websocket/README.md) | [Server-Sent Events and WebSocket on Spring Boot 4: SseEmitter, STOMP, and Which to Pick](https://ankurm.com/spring-boot-4-sse-websocket-stomp/) | The two browser-facing options side by side: the SSE lifecycle and the three ways a stream ends, STOMP's routing defaults, and the size limit that is enforced by Tomcat rather than by Spring | +| [`protocol-comparison/`](protocol-comparison/README.md) | [RSocket vs gRPC vs WebSocket on Spring Boot 4.1: When Each One Wins](https://ankurm.com/rsocket-vs-grpc-vs-websocket-spring-boot-4-1/) | One application serving the same two operations over all three, benchmarked in one JVM, ending in the only measurement that transfers off the box: what each server produces for a consumer that asked for a hundred | | [`broker-comparison/`](broker-comparison/README.md) | [Kafka vs RabbitMQ vs Pulsar for Java Teams: A Decision Framework with Benchmarks](https://ankurm.com/kafka-vs-rabbitmq-vs-pulsar-java-decision-framework/) | The three brokers measured side by side on ordering, replay, consumer scaling and operational footprint, ending in a decision table where every row has a transcript behind it | The brokers make an instructive set. Kafka's consumer holds an offset and the broker remembers @@ -19,6 +21,11 @@ choose how it is read. Almost every difference in how you handle failure follows sentences — which is why Kafka needs a retry topic to do what RabbitMQ does with a queue argument, and why `broker-comparison` is mostly an argument about where a message lives. +The last two modules are about the *other* kind of messaging — a connection rather than a +broker — and they are deliberately adjacent to the broker modules, because the question +"should this be a topic or a stream?" is answered by whether anything needs to survive a dropped +connection. Nothing in `sse-websocket` or `protocol-comparison` persists a single message. + They are meant to be read in order. `kafka-basics` establishes that the default acknowledgement mode is `BATCH` and therefore that delivery is at-least-once; everything the error-handling module does exists because of that one sentence. diff --git a/sse-websocket/README.md b/sse-websocket/README.md new file mode 100644 index 0000000..93efe26 --- /dev/null +++ b/sse-websocket/README.md @@ -0,0 +1,96 @@ +# `sse-websocket` — a live dashboard over SSE and a chat over STOMP + +Companion project for +[**Server-Sent Events and WebSocket on Spring Boot 4: SseEmitter, STOMP, and Which to Pick**](https://ankurm.com/spring-boot-4-sse-websocket-stomp/) +on ankurm.com. + +Seventeen tests against a **real embedded Tomcat**, talking over real sockets. No Docker, no +broker to install, nothing but a JDK. [`./scripts/run-all.sh`](scripts/run-all.sh) regenerates +everything under [`docs/output/`](docs/output/). + +## Versions + +| | Version | Notes | +|---|---|---| +| JDK | 25 (Temurin 25.0.4.1+1) | current LTS | +| Spring Boot | 4.1.1 | latest GA of the 4.1 line | +| Spring Framework | 7.0.9 | Boot-managed | +| Tomcat | 11.0.24 | Boot-managed; its WebSocket buffer is the limit that matters | +| Jackson | 3.1.5 (`tools.jackson`) | why `JacksonJsonMessageConverter`, not `MappingJackson2MessageConverter` | + +Read from `repo1.maven.org/.../maven-metadata.xml` and from Boot's own +`spring-boot-dependencies` POM, not from release announcements. + +## Quickstart + +```bash +./scripts/run-all.sh # every test, then regenerate docs/output/ +mvn test # the same without the capture +mvn spring-boot:run # then: curl -N localhost:8080/sse/metrics +curl -s localhost:8080/diag # open streams, counters, converter order, container buffers +``` + +## Endpoints + +| Path | What it is for | +|---|---| +| `GET /sse/metrics` | the dashboard stream; `SseEmitter(Long.MAX_VALUE)`, fed by a scheduled broadcast | +| `GET /sse/raw` | three events and a close, so the wire format can be captured verbatim | +| `GET /sse/payload` | the same record with and without a media type, plus the same for a `String` | +| `GET /sse/silent?timeoutMs=` | one event then silence, to observe an async timeout from the client side | +| `GET /sse/resume` | echoes `Last-Event-ID` and replays from it | +| `GET /diag` | emitter counters, `SimpUserRegistry` count, converter order, container buffer sizes | +| `GET /diag/async-timeout` | the effective async timeout, read off the live `AsyncContext` | +| `WS /ws` | the STOMP endpoint | +| `WS /ws-sockjs` | the SockJS fallback — a *separate* endpoint, not an option on the first | + +STOMP destinations: `SEND /app/chat.send` → `/topic/room`; `SEND /app/chat.echo` → `/topic/chat.echo` +(no `@SendTo`, on purpose); `SEND /app/chat.whisper` → `/user/queue/whisper`. + +## Profiles + +| Profile | Effect | +|---|---| +| *(none)* | Tomcat's own WebSocket buffers — 8 192 bytes | +| `bigframes` | `ServletServerContainerFactoryBean` with 256 KB buffers; `-Dws.buffer-bytes=` to vary it | + +## Documentation + +1. [Two protocols, one question](docs/01-two-protocols.md) +2. [The SSE lifecycle, and the three ways a stream ends](docs/02-sse-lifecycle.md) +3. [The payload, and a media type that does less than you think](docs/03-the-payload.md) +4. [STOMP: four lines of configuration, three surprising defaults](docs/04-stomp.md) +5. [The limits: origin, size, and which one actually fires](docs/05-limits.md) +6. [Choosing, and when the answer is neither](docs/06-choosing.md) + +## Captured output + +| File | What it shows | +|---|---| +| [`sse-wire-format.txt`](docs/output/sse-wire-format.txt) | the raw bytes of an SSE response, headers included | +| [`sse-payload-conversion.txt`](docs/output/sse-payload-conversion.txt) | `data(x)` vs `data(x, APPLICATION_JSON)` for a record and a `String` | +| [`sse-lifecycle.txt`](docs/output/sse-lifecycle.txt) | an async timeout as the client sees it, and how many writes a dead client absorbs | +| [`async-timeout.txt`](docs/output/async-timeout.txt) | the container default and the property override, read off `AsyncContext` | +| [`sse-concurrency.txt`](docs/output/sse-concurrency.txt) | 40 open streams on a 10-thread connector | +| [`stomp-routing.txt`](docs/output/stomp-routing.txt) | fan-out, the default destination, the controller bypass, user destinations, presence | +| [`websocket-size-limits.txt`](docs/output/websocket-size-limits.txt) | the ceiling binary-searched at four container buffer sizes | +| [`handshake-origin.txt`](docs/output/handshake-origin.txt) | 101 / 101 / 403 for absent, same and foreign `Origin` | +| [`tests.txt`](docs/output/tests.txt) | 17 tests | + +## Six things this module exists to prove + +1. **SSE does not pin a thread per client.** Forty streams stay open on a connector pool capped + at ten, with zero active threads. +2. **An unset `spring.mvc.async.request-timeout` means 30 seconds on Tomcat 11.** Read off the + live `AsyncContext`, not quoted. That is why dashboards reconnect every 30 s with nothing in + the log. +3. **A dead SSE client is discovered by a failed write, and not by the first one.** One full + serialise-and-write succeeded into a socket whose client had already gone. +4. **The media type argument to `SseEmitter.data(..)` is a filter, not a preference.** For a + record it changes nothing; for a `String` it changes nothing either, because + `StringHttpMessageConverter` claims the write first. +5. **A STOMP client can publish straight to `/topic/**` and skip every controller.** The + timestamp the controller would have overwritten arrives untouched. +6. **Spring's 64 KB `messageSizeLimit` is never reached on a stock Tomcat.** The container's + 8 192-byte buffer caps a STOMP frame at about 16 KB and closes the connection with status + 1009 — no exception, no log line from Spring. diff --git a/sse-websocket/docs/01-two-protocols.md b/sse-websocket/docs/01-two-protocols.md new file mode 100644 index 0000000..3724938 --- /dev/null +++ b/sse-websocket/docs/01-two-protocols.md @@ -0,0 +1,81 @@ +# 1. Two protocols, one question + +Next: [2. The SSE lifecycle](02-sse-lifecycle.md) + +--- + +Server-Sent Events and WebSocket get compared as if they were rivals for the same job. They are +not. The question that separates them is much smaller than "which is better", and it is this: + +> **Does the client need to send anything after the first request?** + +If the answer is no — a dashboard, a progress bar, a notification feed, a log tail, a +"your export is ready" ping — SSE is the whole answer, and it is a return type rather than a +subsystem. If the answer is yes, and the messages are frequent or need routing between clients, +you want WebSocket, and almost certainly STOMP on top of it. + +Everything else in the comparison follows from that one asymmetry. + +## What each one costs you to set up + +| | SSE | STOMP over WebSocket | +|---|---|---| +| Dependency | `spring-boot-starter-webmvc` | `spring-boot-starter-websocket` | +| Auto-configuration involved | none | `spring-boot-websocket` | +| Server-side surface | a handler method returning `SseEmitter` | `@EnableWebSocketMessageBroker`, a configurer, a broker, a converter stack | +| Transport | one long-lived HTTP response | an HTTP/1.1 Upgrade to a framed, bidirectional connection | +| Client | `new EventSource(url)`, built into every browser | a STOMP library | +| Reconnect | automatic, with `Last-Event-ID` | yours to write | +| Server knows who is connected | no | yes — `SimpUserRegistry`, connect/subscribe/disconnect events | +| Payload | UTF-8 text only | text or binary | + +The dependency line is the Boot 4 detail worth stating out loud. `spring-boot-starter-websocket` +pulls in `spring-boot-websocket`, which is where the auto-configuration lives. Depending on +`org.springframework:spring-websocket` and `spring-messaging` directly compiles and starts, and +then nothing works — the same rule that catches people with `spring-kafka` in +[`../kafka-basics`](../kafka-basics/README.md) and `spring-rabbit` in +[`../rabbitmq`](../rabbitmq/README.md). In Boot 4, **depending on a library rather than on its +Boot starter means you are missing its auto-configuration.** + +## What SSE is not + +Two things people assume, both wrong, both measured in this module: + +**"It pins a thread per client."** It does not. `SseEmitter` switches the request to asynchronous +mode; the request thread goes back to the pool immediately and the response stays open with +nothing attached to it. [`SseConcurrencyTest`](../src/test/java/com/ankurm/ssews/SseConcurrencyTest.java) +holds forty streams open against a connector whose pool is capped at ten: + +``` +emitters open : 40 +connector pool size : 10 (max 10) +connector active : 0 +``` + +**"You have to pass a media type or it calls `toString()`."** Also not true, and the opposite is +closer to it. See [3. The payload](03-the-payload.md). + +## What SSE genuinely cannot do + +- **Take input.** `EventSource` issues one GET and never sends again. Client-to-server traffic has + to go over ordinary HTTP requests, which is fine for low rates and awful for a chat. +- **Send custom headers.** The browser's `EventSource` has no header API, so bearer-token + authentication needs a cookie or a query parameter. `withCredentials` exists; `Authorization` + does not. +- **Carry binary.** The format is line-oriented UTF-8. Binary means base64, which costs a third. +- **Tell you a client left.** There is no disconnect event, only a write that fails later — + see [2. The SSE lifecycle](02-sse-lifecycle.md). +- **Escape the browser's per-origin connection limit.** Over HTTP/1.1 a browser allows about six + concurrent connections to one origin, and an open SSE stream is one of them, *per tab*. Four + tabs of a dashboard is 4 streams; six is a hung site. HTTP/2 multiplexes and the limit + effectively disappears, which makes "do we serve this over HTTP/2?" a real part of the decision. + +## What STOMP costs + +A broker, a message-converter stack, a session registry, an origin policy, and a set of routing +rules whose defaults are not what you would guess. Chapters +[4](04-stomp.md) and [5](05-limits.md) are that list. + +--- + +Next: [2. The SSE lifecycle](02-sse-lifecycle.md) diff --git a/sse-websocket/docs/02-sse-lifecycle.md b/sse-websocket/docs/02-sse-lifecycle.md new file mode 100644 index 0000000..5dc19fa --- /dev/null +++ b/sse-websocket/docs/02-sse-lifecycle.md @@ -0,0 +1,132 @@ +# 2. The SSE lifecycle, and the three ways a stream ends + +Previous: [1. Two protocols](01-two-protocols.md) · Next: [3. The payload](03-the-payload.md) + +--- + +An `SseEmitter` has exactly one interesting property: **nothing owns it.** Spring does not keep a +list, does not notify you when one dies, and does not clean up after you. Every SSE bug in +production is a variation on that sentence. + +## The wire format + +[`docs/output/sse-wire-format.txt`](output/sse-wire-format.txt), captured from a raw socket by +[`SseWireFormatTest`](../src/test/java/com/ankurm/ssews/SseWireFormatTest.java): + +``` +HTTP/1.1 200 +Content-Type: text/event-stream +Transfer-Encoding: chunked + +:stream open + +id:1 +event:metric +retry:3000 +data:{"seq":1,"host":"node-a","cpu":0.42,"heapMb":512,"at":"2026-09-03T10:00:00Z"} + +id:2 +event:note +data:line one +data:line two +``` + +Points that matter: + +- **No `Content-Length`**, so the response is chunked. Any intermediary that buffers a chunked + response defeats SSE entirely; that is why `X-Accel-Buffering: no` exists for nginx. +- **A blank line terminates an event.** Two `data:` lines in one event are rejoined by the client + with `\n` — a multi-line string does not become two events. +- **A line starting with `:` is a comment** and is discarded. Sending one periodically is the + standard trick for keeping an idle stream alive through a proxy that reaps quiet connections. +- **`retry:` sets the client's reconnect delay** in milliseconds, and it is sticky. +- **`id:` is what comes back as `Last-Event-ID`** on the browser's automatic reconnect. See + `/sse/resume` for the replay shape; almost no server implements it, which is why "SSE + guarantees delivery" is false in practice. + +## Ending 1: the client goes away + +The server does not find out until it writes. `SseDisconnectTest` closes a socket, then +broadcasts, and counts: + +``` +broadcasts before the write failed : 2 +registry.sendsOk : 1 +registry.sendsFailed : 0 -> 1 +``` + +The **first** write after the client vanished succeeded. TCP accepted the bytes into the socket +buffer and only surfaced the reset on the write after that. So on any dashboard, at least one full +serialise-and-write happens into a dead connection, and on a slow-to-reset path it can be several. + +The consequence for your code: **the broadcast loop must catch and drop**, and it must catch two +things. `IOException` is the socket. `IllegalStateException` is the emitter having already been +completed or timed out on a different thread — and if that escapes onto a `@Scheduled` +method, the schedule stops and every client's dashboard freezes at once. + +## Ending 2: the async request times out + +`SseTimeoutTest` opens a stream with a 1.2-second emitter timeout, sends one event, then waits. +[`docs/output/sse-lifecycle.txt`](output/sse-lifecycle.txt): + +``` +HTTP/1.1 200 +Content-Type: text/event-stream +Transfer-Encoding: chunked + +event:hello +data:then silence + +0 +=== stream ended after ~2165 ms === +``` + +**Nothing marks the timeout.** No 503, no error event, no trailer. The response was committed with +200 when the first byte went out, so closing the socket is the only thing left to do, and the +browser's `EventSource` treats that as an ordinary disconnect and reconnects. Server-side it is +visible: `onTimeout` ran and the registry's counter moved. + +Note the ~2.1 s for a 1.2 s timeout. Tomcat checks async timeouts from its background +processor, so expiry is granular to about a second rather than exact. + +### The number that decides how long a stream lives + +Three levels, checked in order: + +1. `new SseEmitter(millis)` — per handler +2. `spring.mvc.async.request-timeout` — per application +3. whatever the servlet container defaults to — invisible + +`AsyncTimeoutUnsetTest` reads the third off the live `AsyncContext` rather than quoting it +([`async-timeout.txt`](output/async-timeout.txt)): + +``` +{"spring.mvc.async.request-timeout":"","servletContainer":"Apache Tomcat/11.0.24","effectiveAsyncTimeoutMs":30000} +{"spring.mvc.async.request-timeout":"5s","servletContainer":"Apache Tomcat/11.0.24","effectiveAsyncTimeoutMs":5000} +``` + +**Thirty seconds.** A dashboard that silently reconnects every thirty seconds, with nothing in the +log, is this and nothing else. `new SseEmitter(Long.MAX_VALUE)` is the usual fix, and it is also +what turns a leaked emitter into a permanent one. + +Reading it took some care: inside the handler method `request.isAsyncStarted()` is still `false`, +because Spring starts async processing *after* the method returns and it has seen the return type. +The first place the real `AsyncContext` exists is +`AsyncHandlerInterceptor.afterConcurrentHandlingStarted`. + +## Ending 3: you complete it + +`emitter.complete()`, or `completeWithError(ex)`. The second writes nothing useful to a stream +that is already committed — if you want the client to know something went wrong, send an +`event:error` of your own *first*, then complete. + +## Therefore: a registry + +[`EmitterRegistry`](../src/main/java/com/ankurm/ssews/EmitterRegistry.java) exists because of the +three endings above. It removes on **all three** callbacks, not just `onCompletion`. Registering +only `onCompletion` is the common version of this bug: a timed-out emitter is never completed by +the client, so it stays in the map for ever and every subsequent broadcast pays for it. + +--- + +Previous: [1. Two protocols](01-two-protocols.md) · Next: [3. The payload](03-the-payload.md) diff --git a/sse-websocket/docs/03-the-payload.md b/sse-websocket/docs/03-the-payload.md new file mode 100644 index 0000000..da943e2 --- /dev/null +++ b/sse-websocket/docs/03-the-payload.md @@ -0,0 +1,83 @@ +# 3. The payload, and a media type that does less than you think + +Previous: [2. The SSE lifecycle](02-sse-lifecycle.md) · Next: [4. STOMP](04-stomp.md) + +--- + +The advice you will find is: always pass a media type to `data(..)`, or your object is written +with `toString()`. Both halves of that turn out to be wrong on Boot 4.1, and the mechanism +underneath explains a real trap that the advice misses. + +## What actually happens + +`SseEmitter.send(Object data, MediaType mediaType)` walks the application's configured +`HttpMessageConverter` list and uses **the first converter whose `canWrite(type, mediaType)` +returns true**. The media type argument is a *filter*, not a preference. A `null` media type +filters nothing, so the first converter that can write the type at all wins. + +`/diag` prints the list in order: + +``` +ByteArrayHttpMessageConverter +StringHttpMessageConverter +ResourceHttpMessageConverter +ResourceRegionHttpMessageConverter +AllEncompassingFormHttpMessageConverter +JacksonJsonHttpMessageConverter +Jaxb2RootElementHttpMessageConverter +``` + +Two consequences, both measured by +[`SseDataConversionTest`](../src/test/java/com/ankurm/ssews/SseDataConversionTest.java) +([`sse-payload-conversion.txt`](output/sse-payload-conversion.txt)): + +**For a record or POJO, the media type changes nothing.** No converter ahead of Jackson claims it, +so Jackson writes it either way: + +``` +=== SseEmitter.event().data(record, APPLICATION_JSON) === +{"seq":7,"host":"node-a","cpu":0.42,"heapMb":512,"at":"2026-09-03T10:00:00Z"} +=== SseEmitter.event().data(record) [no media type] === +{"seq":7,"host":"node-a","cpu":0.42,"heapMb":512,"at":"2026-09-03T10:00:00Z"} +=== POJO: identical? true === +``` + +**For a `String`, the media type *still* changes nothing** — and this is the part that +surprised me enough to rewrite the test. `StringHttpMessageConverter` supports `MediaType.ALL` and +sits ahead of Jackson, so it claims the write even when you explicitly ask for +`application/json`: + +``` +=== String payload, no media type -> {"already":"json"} +=== String payload, APPLICATION_JSON -> {"already":"json"} +``` + +There is no way to make `send()` JSON-quote a `String`. Good news if you are streaming +pre-rendered JSON — it goes out verbatim, not double-encoded. A trap if you assumed a +`String` field would be escaped for you: it will not be, and a payload containing a newline +becomes two `data:` lines rather than one. + +## Jackson 3, and the `Instant` that proves it + +`"at":"2026-09-03T10:00:00Z"` — an ISO-8601 string, not an epoch number. Boot 4.1.1 manages +Jackson **3.1.5** under `tools.jackson`, and `JacksonJsonHttpMessageConverter` is the Jackson 3 +converter. The Jackson 2 sibling (`MappingJackson2HttpMessageConverter`) is still on the +classpath in the messaging stack. + +**Rule for Boot 4: a `2` in a Spring class name means the *previous* Jackson.** The unnumbered +name is current, which is the opposite of the convention you would guess, and it is the same fork +that catches people in [`../kafka-basics`](../kafka-basics/README.md) and +[`../rabbitmq`](../rabbitmq/README.md). For STOMP it is +`JacksonJsonMessageConverter` you want, not `MappingJackson2MessageConverter`. + +## Practical guidance + +- Pass `MediaType.APPLICATION_JSON` anyway. It costs nothing, it documents intent, and it makes + the code correct against a converter list you did not configure yourself. +- Do not stream a `String` and assume it is escaped. Send the object. +- Keep events small. Every subscriber pays for one serialisation per event; see + [6. Choosing](06-choosing.md). + +--- + +Previous: [2. The SSE lifecycle](02-sse-lifecycle.md) · Next: [4. STOMP](04-stomp.md) diff --git a/sse-websocket/docs/04-stomp.md b/sse-websocket/docs/04-stomp.md new file mode 100644 index 0000000..f7468ac --- /dev/null +++ b/sse-websocket/docs/04-stomp.md @@ -0,0 +1,129 @@ +# 4. STOMP: four lines of configuration, three surprising defaults + +Previous: [3. The payload](03-the-payload.md) · Next: [5. The limits](05-limits.md) + +--- + +[`WebSocketConfig`](../src/main/java/com/ankurm/ssews/WebSocketConfig.java) is short. Each line +decides more than it looks like it does. + +```java +registry.enableSimpleBroker("/topic", "/queue"); +registry.setApplicationDestinationPrefixes("/app"); +registry.setUserDestinationPrefix("/user"); +``` + +## Surprise 1: a client can publish straight to a topic + +`setApplicationDestinationPrefixes("/app")` says which destinations reach a `@MessageMapping`. +It does **not** say which destinations a client may send to. A `SEND` to `/topic/room` goes +directly to the broker, and every subscriber receives it. +[`StompRoutingTest`](../src/test/java/com/ankurm/ssews/StompRoutingTest.java) sends a message with +a timestamp the controller would have overwritten, and it arrives untouched: + +``` +=== SEND straight to /topic/room === +received: ChatMessage[from=not-checked-by-anyone, text=unvalidated, at=2000-01-01T00:00:00Z] +=== the controller's Instant.now() rewrite did NOT happen === +``` + +No handler ran. No validation ran. Nothing logged that anything was skipped. If your controller is +where authorisation, sanitisation or rate limiting happens, a client that knows the destination +name walks past all of it. + +The fix is not a routing setting — it is Spring Security's message-level authorisation +(`simpDestMatchers("/topic/**").denyAll()` for client `SEND` frames), or a broker relay whose +own ACLs forbid publishing. Treat a bare simple broker as an open pipe. + +## Surprise 2: a handler with no `@SendTo` is not private + +The default destination for a `@MessageMapping` return value is the broker prefix plus the +mapping. `@MessageMapping("/chat.echo")` with no annotation publishes to **`/topic/chat.echo`**: + +``` +=== @MessageMapping("/chat.echo") with NO @SendTo === +subscribed to /topic/chat.echo, received: ChatMessage[from=echo, text=who can see this?, ...] +``` + +Anyone who can guess the mapping can subscribe to the replies. If a handler's answer is for the +caller only, it needs `@SendToUser`, not silence. + +## Surprise 3: `convertAndSendToUser` needs more than a name + +`/user/**` is not a real destination. On `SUBSCRIBE`, `DefaultUserDestinationResolver` rewrites +`/user/queue/whisper` into a session-scoped destination, which is why two clients subscribed to +the same string get different messages: + +``` +Translated /user/queue/whisper -> [/queue/whisper-userea271ead-...] +Translated /user/queue/whisper -> [/queue/whisper-user84865e50-...] +``` + +With no authenticated `Principal`, the "user" is the STOMP session id — and the three-argument +`convertAndSendToUser(sessionId, dest, payload)` **silently delivers nothing**, because the +resolver looks that name up in a user registry that has never heard of it. No exception, no log +line. The message has to carry the session id too: + +```java +SimpMessageHeaderAccessor out = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE); +out.setSessionId(targetSessionId); +out.setLeaveMutable(true); // without this the accessor is frozen before the resolver looks +template.convertAndSendToUser(targetSessionId, "/queue/whisper", payload, out.getMessageHeaders()); +``` + +And there is a second trap stacked on the first, which cost a debugging round here: + +> **`StompSession.getSessionId()` on the client is not the server's session id.** +> ``` +> alice.getSessionId() [client side] : 08b58570-0fcb-ffaf-d82c-119232e66561 +> server-side session id : f882cf6a-f0d9-48e3-baf3-c573045a47d1 +> ``` +> Whispering to the client-side value delivers nothing, silently, exactly like the missing header +> does. The server-side id is what appears in `SimpMessageHeaderAccessor.getSessionId()`, and it +> has to come from the server. + +With Spring Security in the picture the `Principal` is a real user name and none of this applies +— which is a good argument for putting security in front of a STOMP endpoint even when you +do not think you need authentication yet. + +## What STOMP gives you that SSE cannot + +Presence. [`PresenceListener`](../src/main/java/com/ankurm/ssews/PresenceListener.java) is fifteen +lines and produces: + +``` +| CONNECTED d775c936-8492-4b32-871e-c659346a2fb9 +| SUBSCRIBE d775c936-8492-4b32-871e-c659346a2fb9 -> /topic/room +| DISCONNECT def64a1f-dc10-45bc-92d3-aa2e842ae02f status=CloseStatus[code=1000, reason=null] +``` + +The server knows who is connected, what they subscribed to, and when they left, with a close +status attached. SSE has no equivalent of any of those three. + +## The simple broker is not a broker + +`enableSimpleBroker` is in-memory, single JVM, no persistence, no acknowledgement, and it drops +everything on restart. It is also fine for a great many applications. When it is not, +`enableStompBrokerRelay("/topic", "/queue")` swaps in RabbitMQ or ActiveMQ **without touching a +single controller** — the annotations and destinations stay exactly as they are. The broker +side of that is [`../rabbitmq`](../rabbitmq/README.md). + +Note the asymmetry once you relay: `/topic` and `/queue` become real broker destinations with +real semantics, and the "client can publish straight to a topic" problem becomes the broker's ACL +problem rather than yours. + +## SockJS is a second endpoint, not an option on the first + +```java +registry.addEndpoint("/ws"); +registry.addEndpoint("/ws-sockjs").withSockJS(); +``` + +`.withSockJS()` on the same path does not give you both. SockJS rewrites the URL into +`/{server}/{session}/{transport}`, and a plain WebSocket client pointed at that path fails the +handshake. In 2026, SockJS is mostly legacy — WebSocket support is universal — but it +is still the answer for a corporate proxy that strips `Upgrade`. + +--- + +Previous: [3. The payload](03-the-payload.md) · Next: [5. The limits](05-limits.md) diff --git a/sse-websocket/docs/05-limits.md b/sse-websocket/docs/05-limits.md new file mode 100644 index 0000000..989ff78 --- /dev/null +++ b/sse-websocket/docs/05-limits.md @@ -0,0 +1,119 @@ +# 5. The limits: origin, size, and which one actually fires + +Previous: [4. STOMP](04-stomp.md) · Next: [6. Choosing](06-choosing.md) + +--- + +## Origin: checked by default, and the check is real + +The WebSocket handshake is an ordinary HTTP GET, and the browser's same-origin policy **does not +apply to it**. Without a server-side check, a page on any site could open a WebSocket to your +server with the user's cookies attached. Spring checks `Origin` by default. +[`WebSocketHandshakeOriginTest`](../src/test/java/com/ankurm/ssews/WebSocketHandshakeOriginTest.java) +varies only that header ([`handshake-origin.txt`](output/handshake-origin.txt)): + +``` +no Origin header -> HTTP/1.1 101 +Origin: http://localhost -> HTTP/1.1 101 +Origin: https://evil... -> HTTP/1.1 403 +``` + +The middle row is the policy working. The **first** row is the one to understand: a client that +sends no `Origin` at all is allowed through, which is why every Java test in this module connects +without ceremony — and why `Origin` is a browser control, not an authentication mechanism. +For a real cross-origin front end use `setAllowedOriginPatterns("https://app.example.com")`; +`setAllowedOrigins("*")` is rejected outright when credentials are allowed. + +## Size: two limits, and the documented one is not the one that fires + +`WebSocketTransportRegistration.setMessageSizeLimit` defaults to 64 KB and is what every +guide tells you to raise. Underneath it sits the servlet container's own WebSocket buffer, which +`/diag` reports as: + +``` +"ws.container.defaultMaxTextMessageBufferSize":8192 +``` + +Spring's `SubProtocolWebSocketHandler` does not support partial messages, so the container has to +assemble each text message whole before Spring sees it. Cross the container's ceiling and the +connection is **closed** — WebSocket status 1009, `The decoded text message was too big for +the output buffer and the endpoint does not support partial messages` — before Spring's +limit is ever consulted. + +[`StompSizeLimitDefaultBufferTest`](../src/test/java/com/ankurm/ssews/StompSizeLimitDefaultBufferTest.java) +binary-searches for the exact ceiling +([`websocket-size-limits.txt`](output/websocket-size-limits.txt)): + +| container buffer | largest body delivered | smallest rejected | +|---|---|---| +| 8 192 (Tomcat default) | 16 459 | 16 522 | +| 16 384 | 16 446 | 16 496 | +| 32 768 | 32 615 | 32 665 | +| 65 536 | 65 350 | 65 400 | + +Body length is the JSON string field; the STOMP frame adds about 90 bytes of headers on top, so +each row is a frame just under 16 KB, 16 KB, 32 KB and 64 KB respectively. + +Read that table honestly: **the ceiling behaves like `max(16 KB, buffer)`**. Setting the buffer to +16 KB changes nothing from the default; above 16 KB the ceiling follows it. I have not +reduced the 16 KB floor to a line in Tomcat's source and will not claim a mechanism I have +not read. What is certain, and is the point: + +> **Spring's 64 KB `messageSizeLimit` is not reached on a stock Tomcat.** The real ceiling is +> about 16 KB, it is enforced by the container, and raising Spring's limit alone does +> nothing whatsoever. + +The fix is a `ServletServerContainerFactoryBean` +([`WebSocketBufferConfig`](../src/main/java/com/ankurm/ssews/WebSocketBufferConfig.java), active +under the `bigframes` profile): + +```java +@Bean +public ServletServerContainerFactoryBean createWebSocketContainer() { + ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); + container.setMaxTextMessageBufferSize(256 * 1024); + container.setMaxBinaryMessageBufferSize(256 * 1024); + return container; +} +``` + +### The two failures look different, and only one is diagnosable + +With the buffer at 256 KB, Spring's limit finally decides — and it fails *better*: + +``` + 60 KB body -> delivered, body length 61440 + 80 KB body -> NOT delivered, connection dropped +org.springframework.messaging.simp.stomp.StompConversionException: The configured STOMP buffer size limit of 65536 bytes has been exceeded +``` + +Spring logs the exception, names the limit, and sends a STOMP `ERROR` frame the client can see. +The container's refusal does none of that. So the practical advice is not just "raise the buffer" +— it is **raise the container buffer above Spring's limit deliberately, so that the limit +that fires is the one that tells you it fired.** + +### The client has the same limit + +A `WebSocketStompClient` built on `StandardWebSocketClient` inherits the same 8 KB default, +and hits it on its *own* session first. That produced a false positive while writing these tests: +a message that looked rejected by the server was actually refused by the client. +[`SizeProbe`](../src/test/java/com/ankurm/ssews/SizeProbe.java) raises the client side to 1 MB +before measuring anything. + +## Send buffer and send time + +Two limits that bite in production rather than in tests, both set in `WebSocketConfig`: + +- `setSendBufferSizeLimit` (512 KB here) — how much undelivered output Spring will hold + for one slow client before closing its session. Without it, one client on hotel wifi can grow + the heap until the process dies. +- `setSendTimeLimit` (20 s here) — how long a single send may take before the session is + closed. + +Both exist because a broadcast is only as fast as its slowest subscriber, and a WebSocket session +has no equivalent of TCP's back-pressure reaching your application code. This is precisely the gap +RSocket closes, and the reason the [next article](06-choosing.md) exists. + +--- + +Previous: [4. STOMP](04-stomp.md) · Next: [6. Choosing](06-choosing.md) diff --git a/sse-websocket/docs/06-choosing.md b/sse-websocket/docs/06-choosing.md new file mode 100644 index 0000000..b530849 --- /dev/null +++ b/sse-websocket/docs/06-choosing.md @@ -0,0 +1,80 @@ +# 6. Choosing, and when the answer is neither + +Previous: [5. The limits](05-limits.md) + +--- + +## The decision, in one table + +| If | Use | Because | +|---|---|---| +| Server pushes, client never speaks | **SSE** | A return type. No broker, no converter stack, no origin policy. Reconnect is free. | +| Clients talk to each other, or to the server, often | **STOMP over WebSocket** | Routing, presence, per-user destinations, and a path to a real broker without touching controllers. | +| One-off progress on an operation the client already started | **SSE** | The stream is scoped to the request. Nothing to clean up. | +| Payloads are binary, or large | **WebSocket** | SSE is line-oriented UTF-8; binary costs a third in base64. | +| The consumer is another service, not a browser | **neither** — see below | Browsers are the only reason to accept SSE's or STOMP's constraints. | +| You need the consumer to say "slow down" | **neither** | Covered in the next article. | + +## The costs nobody puts in the table + +**SSE costs a connection per stream, per tab.** Over HTTP/1.1 a browser allows about six +concurrent connections per origin, and an open `EventSource` is one of them. Three tabs of your +dashboard leaves three for everything else on the page. Over HTTP/2 they multiplex over one +connection and the limit effectively disappears — so "are we behind HTTP/2 end to end, +including the proxy?" is a real input to this decision, not a detail. + +**SSE costs one serialisation per subscriber per event.** The fan-out in +[`EmitterRegistry.broadcast`](../src/main/java/com/ankurm/ssews/EmitterRegistry.java) is a loop: +a thousand open dashboards means a thousand `send()` calls on one scheduler thread, every tick. +The obvious fix — serialise once and write the bytes N times — is not something +`SseEmitter` offers. + +**SSE does not pin threads, though.** That objection is wrong, and +[`sse-concurrency.txt`](output/sse-concurrency.txt) is the evidence: + +``` +=== server.tomcat.threads.max=10, clients=40 === +emitters open : 40 +connector pool size : 10 (max 10) +connector active : 0 +one broadcast reached: 40 streams +clients that read it : 40 +``` + +**STOMP costs a stateful server.** Sessions live in one JVM's memory. Two instances behind a load +balancer do not share subscriptions, so a message published on instance A never reaches a +subscriber on instance B — which is the real reason to move to a broker relay, well before +you need persistence. + +## Should you build this at all? + +Three cases where the honest answer is no: + +- **Updates arrive every few minutes.** Polling every thirty seconds is one endpoint, no + lifecycle, no proxy configuration, no reconnect logic and no leaked emitters. The right answer + more often than it is chosen. +- **You want a chat and you have one server and no plan for a second.** Everything in + [chapter 4](04-stomp.md) is real work, and the first time you scale out you will be doing it + again with a broker relay. Decide about the broker first. +- **The client is another service.** SSE's advantage is that browsers implement it. STOMP's + advantage is that browsers can speak it. Neither advantage applies service-to-service, and + neither protocol gives you flow control, deadlines or a schema — which is what the + companion article on RSocket, gRPC and raw WebSocket is about. + +## Everything else, one line each + +- **Behind nginx**, a chunked SSE response is buffered by default and the client sees nothing + until the buffer fills. `proxy_buffering off` or the `X-Accel-Buffering: no` response header. +- **`EventSource` cannot send an `Authorization` header.** Cookie, or a query parameter you then + have to keep out of access logs. +- **Heartbeats.** SSE: send a `:` comment every N seconds. STOMP: `setHeartbeatValue` plus a + `TaskScheduler`, and both sides must agree. +- **Compression.** `text/event-stream` compresses extremely well and Boot's HTTP compression is + off by default; `server.compression.mime-types` must list it explicitly. +- **Surefire does not discover static nested test classes.** Seven of this module's seventeen + tests silently did not run until they were split into top-level classes. Check the count, not + the colour. + +--- + +Previous: [5. The limits](05-limits.md) diff --git a/sse-websocket/docs/output/async-timeout.txt b/sse-websocket/docs/output/async-timeout.txt new file mode 100644 index 0000000..8838412 --- /dev/null +++ b/sse-websocket/docs/output/async-timeout.txt @@ -0,0 +1,4 @@ +=== spring.mvc.async.request-timeout ABSENT === +{"spring.mvc.async.request-timeout":"","servletContainer":"Apache Tomcat/11.0.24","effectiveAsyncTimeoutMs":30000} +=== spring.mvc.async.request-timeout=5s === +{"spring.mvc.async.request-timeout":"5s","servletContainer":"Apache Tomcat/11.0.24","effectiveAsyncTimeoutMs":5000} diff --git a/sse-websocket/docs/output/handshake-origin.txt b/sse-websocket/docs/output/handshake-origin.txt new file mode 100644 index 0000000..7ca5855 --- /dev/null +++ b/sse-websocket/docs/output/handshake-origin.txt @@ -0,0 +1,4 @@ +=== GET /ws handshake, varying only the Origin header === +no Origin header -> HTTP/1.1 101 +Origin: http://localhost -> HTTP/1.1 101 +Origin: https://evil... -> HTTP/1.1 403 diff --git a/sse-websocket/docs/output/sse-concurrency.txt b/sse-websocket/docs/output/sse-concurrency.txt new file mode 100644 index 0000000..691304d --- /dev/null +++ b/sse-websocket/docs/output/sse-concurrency.txt @@ -0,0 +1,6 @@ +=== server.tomcat.threads.max=10, clients=40 === +emitters open : 40 +connector pool size : 10 (max 10) +connector active : 0 +one broadcast reached: 40 streams +clients that read it : 40 diff --git a/sse-websocket/docs/output/sse-lifecycle.txt b/sse-websocket/docs/output/sse-lifecycle.txt new file mode 100644 index 0000000..8e9ab6e --- /dev/null +++ b/sse-websocket/docs/output/sse-lifecycle.txt @@ -0,0 +1,23 @@ +=== GET /sse/silent?timeoutMs=1200 : whole response === +| HTTP/1.1 200 +| Content-Type: text/event-stream +| Transfer-Encoding: chunked +| Date: Thu, 03 Sep 2026 18:59:34 GMT +| Connection: close +| +| 1f +| event:hello +| data:then silence +| +| +| 0 +| +=== stream ended after ~1697 ms === +=== registry.timedOut(): 0 -> 1 === + +=== client closed the socket, then: === +broadcasts before the write failed : 2 +registry.sendsOk : 1 +registry.sendsFailed : 0 -> 1 +registry.open : 0 +=== writes that appeared to succeed after the client was gone: 1 === diff --git a/sse-websocket/docs/output/sse-payload-conversion.txt b/sse-websocket/docs/output/sse-payload-conversion.txt new file mode 100644 index 0000000..2656d77 --- /dev/null +++ b/sse-websocket/docs/output/sse-payload-conversion.txt @@ -0,0 +1,8 @@ +=== SseEmitter.event().data(record, APPLICATION_JSON) === +{"seq":7,"host":"node-a","cpu":0.42,"heapMb":512,"at":"2026-09-03T10:00:00Z"} +=== SseEmitter.event().data(record) [no media type] === +{"seq":7,"host":"node-a","cpu":0.42,"heapMb":512,"at":"2026-09-03T10:00:00Z"} +=== POJO: identical? true === +=== String payload, no media type -> {"already":"json"} +=== String payload, APPLICATION_JSON -> {"already":"json"} +=== String payload, APPLICATION_JSON -> {"already":"json"} diff --git a/sse-websocket/docs/output/sse-wire-format.txt b/sse-websocket/docs/output/sse-wire-format.txt new file mode 100644 index 0000000..32f52fd --- /dev/null +++ b/sse-websocket/docs/output/sse-wire-format.txt @@ -0,0 +1,22 @@ +=== GET /sse/raw : raw bytes, one line per row === +| HTTP/1.1 200 +| Content-Type: text/event-stream +| Transfer-Encoding: chunked +| Date: Thu, 03 Sep 2026 18:59:33 GMT +| Connection: close +| +| ac +| :stream open +| +| id:1 +| event:metric +| retry:3000 +| data:{"seq":1,"host":"node-a","cpu":0.42,"heapMb":512,"at":"2026-09-03T10:00:00Z"} +| +| id:2 +| event:note +| data:line one +| data:line two +| +| +| 0 diff --git a/sse-websocket/docs/output/stomp-routing.txt b/sse-websocket/docs/output/stomp-routing.txt new file mode 100644 index 0000000..353344c --- /dev/null +++ b/sse-websocket/docs/output/stomp-routing.txt @@ -0,0 +1,23 @@ +=== SEND /app/chat.send -> @SendTo("/topic/room") === +alice received : ChatMessage[from=alice, text=is this thing on?, at=2026-09-03T18:59:36.632180396Z] +bob received : ChatMessage[from=alice, text=is this thing on?, at=2026-09-03T18:59:36.632180396Z] +=== presence events the server saw === +| CONNECTED 122616fa-6e84-4c09-a317-6dee68f17794 +| CONNECTED 22633c5f-f886-4ff2-94d0-156e43d92870 +| SUBSCRIBE 122616fa-6e84-4c09-a317-6dee68f17794 -> /topic/room +| SUBSCRIBE 22633c5f-f886-4ff2-94d0-156e43d92870 -> /topic/room +| DISCONNECT 122616fa-6e84-4c09-a317-6dee68f17794 status=CloseStatus[code=1002, reason=null] + +=== @MessageMapping("/chat.echo") with NO @SendTo === +subscribed to /topic/chat.echo, received: ChatMessage[from=echo, text=who can see this?, at=2026-09-03T18:59:38.089388303Z] + +=== SEND straight to /topic/room === +received: ChatMessage[from=not-checked-by-anyone, text=unvalidated, at=2000-01-01T00:00:00Z] +=== the controller's Instant.now() rewrite did NOT happen === + +alice.getSessionId() [client side] : 97d9d368-53ab-0f29-0dd4-2c0515124971 +server-side session id : c6ddbdda-36a6-4a9e-8a15-6714379ecdb9 +=== convertAndSendToUser(aliceSession, "/queue/whisper", ..) === +both subscribed to the SAME string : /user/queue/whisper +alice received : ChatMessage[from=42eda7c9-16b6-4ffc-bf68-60e8edfe4239, text=just for you, at=2026-09-03T19:00:15.135586620Z] +bob received : null diff --git a/sse-websocket/docs/output/tests.txt b/sse-websocket/docs/output/tests.txt new file mode 100644 index 0000000..69f9a29 --- /dev/null +++ b/sse-websocket/docs/output/tests.txt @@ -0,0 +1,16 @@ +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 25.88 s -- in com.ankurm.ssews.StompSizeLimitTunedBufferTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.620 s -- in com.ankurm.ssews.AsyncTimeoutUnsetTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.044 s -- in com.ankurm.ssews.SseWireFormatTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.042 s -- in com.ankurm.ssews.WebSocketHandshakeOriginTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.416 s -- in com.ankurm.ssews.AsyncTimeoutPropertyTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.144 s -- in com.ankurm.ssews.SseDisconnectTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.342 s -- in com.ankurm.ssews.SseConcurrencyTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 2.023 s -- in com.ankurm.ssews.SseTimeoutTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 1.146 s -- in com.ankurm.ssews.StompChatTest +Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.647 s -- in com.ankurm.ssews.StompRoutingTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 13.52 s -- in com.ankurm.ssews.StompSizeLimitSpringLimitTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.040 s -- in com.ankurm.ssews.SseDataConversionTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.042 s -- in com.ankurm.ssews.WebSocketContainerRaisedTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 22.80 s -- in com.ankurm.ssews.StompSizeLimitDefaultBufferTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 2.143 s -- in com.ankurm.ssews.StompUserDestinationTest +Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.025 s -- in com.ankurm.ssews.WebSocketContainerDefaultsTest diff --git a/sse-websocket/docs/output/websocket-size-limits.txt b/sse-websocket/docs/output/websocket-size-limits.txt new file mode 100644 index 0000000..4f1a86c --- /dev/null +++ b/sse-websocket/docs/output/websocket-size-limits.txt @@ -0,0 +1,36 @@ +The limit that fires first is the servlet container's WebSocket buffer, not Spring's +messageSizeLimit. 'body' below is the length of the JSON string field; the STOMP frame +carries about 90 bytes of headers on top of it. + +=== container buffer: Tomcat default (8192, see /diag) === + 8 KB body -> delivered, body length 8192 + 16 KB body -> delivered, body length 16384 + 60 KB body -> NOT delivered, connection dropped +largest body delivered : 16459 bytes +smallest body rejected : 16522 bytes + +=== container buffer set to 16384 === +largest body delivered : 16446 bytes +smallest body rejected : 16496 bytes +=== container buffer set to 32768 === +largest body delivered : 32615 bytes +smallest body rejected : 32665 bytes +=== container buffer set to 65536 === +largest body delivered : 65350 bytes +smallest body rejected : 65400 bytes + +=== bigframes profile: container buffer 256 KB, Spring limit 64 KB === + 16 KB body -> delivered, body length 16384 + 60 KB body -> delivered, body length 61440 +desti..], byteCount=82068, last=true] in session 1655473b-76fe-4489-99d6-e3f8200b41b4. Sending STOMP ERROR to client. +org.springframework.messaging.simp.stomp.StompConversionException: The configured STOMP buffer size limit of 65536 bytes has been exceeded + 80 KB body -> NOT delivered, connection dropped +desti..], byteCount=204949, last=true] in session 8dcce90c-48db-43a5-8d32-659da1fe32d8. Sending STOMP ERROR to client. +org.springframework.messaging.simp.stomp.StompConversionException: The configured STOMP buffer size limit of 65536 bytes has been exceeded +200 KB body -> NOT delivered, connection dropped + +Spring's own limit, once the container is out of the way: +org.springframework.messaging.simp.stomp.StompConversionException: The configured STOMP buffer size limit of 65536 bytes has been exceeded + +"ws.container.defaultMaxTextMessageBufferSize":262144 +"ws.container.defaultMaxTextMessageBufferSize":8192 diff --git a/sse-websocket/pom.xml b/sse-websocket/pom.xml new file mode 100644 index 0000000..889abf5 --- /dev/null +++ b/sse-websocket/pom.xml @@ -0,0 +1,59 @@ + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 4.1.1 + + + + com.ankurm + sse-websocket + 1.0 + jar + + + 25 + UTF-8 + + + + + + org.springframework.boot + spring-boot-starter-webmvc + + + + + org.springframework.boot + spring-boot-starter-websocket + + + + org.springframework.boot + spring-boot-starter-jackson + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/sse-websocket/scripts/run-all.sh b/sse-websocket/scripts/run-all.sh new file mode 100755 index 0000000..b914234 --- /dev/null +++ b/sse-websocket/scripts/run-all.sh @@ -0,0 +1,63 @@ +#!/usr/bin/env bash +# Regenerate every file under docs/output/. Nothing to install beyond a JDK: every test starts an +# embedded Tomcat on a random port and talks to it over a real socket. +set -eu +cd "$(dirname "$0")/.." +LOG=/tmp/sse-websocket-test.log + +mvn -B test > "$LOG" 2>&1 + +# The buffer sweep: the same binary search at three container buffer sizes, so the ceiling can be +# shown as a function of the setting rather than as one number. +: > /tmp/sse-buffer-sweep.log +for b in 16384 32768 65536; do + mvn -B test -Dtest=StompSizeLimitTunedBufferTest -Dwsbuf="$b" 2>&1 \ + | grep -E '^=== container buffer|^largest body|^smallest body' >> /tmp/sse-buffer-sweep.log +done + +sed -n '/=== GET \/sse\/raw/,/^| 0$/p' "$LOG" > docs/output/sse-wire-format.txt +sed -n '/=== SseEmitter.event()/,/=== String payload, APPLICATION_JSON/p' "$LOG" \ + > docs/output/sse-payload-conversion.txt + +{ + sed -n '/=== GET \/sse\/silent/,/registry.timedOut/p' "$LOG" + echo + sed -n '/=== client closed the socket/,/appeared to succeed/p' "$LOG" +} > docs/output/sse-lifecycle.txt + +sed -n '/=== server.tomcat.threads.max=10/,/clients that read it/p' "$LOG" \ + > docs/output/sse-concurrency.txt + +grep -E '^=== spring.mvc.async|^\{"spring.mvc.async' "$LOG" > docs/output/async-timeout.txt + +{ + sed -n '/=== SEND \/app\/chat.send/,/^| DISCONNECT/p' "$LOG"; echo + sed -n '/=== @MessageMapping/,/^subscribed to/p' "$LOG"; echo + sed -n '/=== SEND straight to/,/did NOT happen/p' "$LOG"; echo + sed -n '/^alice.getSessionId/,/^bob received/p' "$LOG" +} > docs/output/stomp-routing.txt + +{ + echo "The limit that fires is the servlet container's WebSocket buffer, not Spring's" + echo "messageSizeLimit. Body length below is the JSON string field; the STOMP frame carries" + echo "about 90 bytes of headers on top." + echo + sed -n '/=== container buffer: Tomcat default/,/smallest body rejected/p' "$LOG" + echo + cat /tmp/sse-buffer-sweep.log + echo + sed -n '/=== bigframes profile/,/^200 KB body/p' "$LOG" \ + | grep -vE '^\s+at |^2026-|^$|Initializing|Completed initialization' + echo + echo "Spring's own limit, once the container is out of the way:" + grep -m1 -A1 'Failed to parse TextMessage' "$LOG" | sed 's/^\[[^]]*\] //' + grep -m1 'STOMP buffer size limit' "$LOG" + echo + grep -oE '"ws.container.defaultMaxTextMessageBufferSize":[0-9]+' "$LOG" | sort -u +} > docs/output/websocket-size-limits.txt + +sed -n '/=== GET \/ws handshake/,/Origin: https:\/\/evil/p' "$LOG" > docs/output/handshake-origin.txt + +grep -E 'Tests run:.*in com\.ankurm' "$LOG" | sed 's/^\[INFO\] //' > docs/output/tests.txt + +echo "regenerated:"; ls -1 docs/output/ diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/AsyncTimeoutProbe.java b/sse-websocket/src/main/java/com/ankurm/ssews/AsyncTimeoutProbe.java new file mode 100644 index 0000000..989ab4b --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/AsyncTimeoutProbe.java @@ -0,0 +1,86 @@ +package com.ankurm.ssews; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; + +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Configuration; +import org.springframework.stereotype.Component; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.context.request.async.DeferredResult; +import org.springframework.web.servlet.AsyncHandlerInterceptor; +import org.springframework.web.servlet.config.annotation.InterceptorRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; + +/** + * Answers "how long does my SSE stream actually get?" without waiting for it to happen. + * + *

The chain is: {@code SseEmitter}'s own constructor argument, else + * {@code spring.mvc.async.request-timeout}, else whatever the servlet container defaults to. + * Only the last one is invisible. + * + *

Reading it needs a little care. Inside the handler method {@code isAsyncStarted()} is still + * {@code false} — Spring starts async processing after the method returns, once + * it sees the return type. The first place the real {@code AsyncContext} exists is + * {@link AsyncHandlerInterceptor#afterConcurrentHandlingStarted}, so that is where the timeout + * is captured. + */ +@RestController +public class AsyncTimeoutProbe { + + static final AtomicReference LAST_TIMEOUT = new AtomicReference<>(""); + + private final String configured; + + AsyncTimeoutProbe(@Value("${spring.mvc.async.request-timeout:}") String configured) { + this.configured = configured; + } + + /** Start async, so the interceptor below can read the effective timeout, then answer. */ + @GetMapping("/diag/async-start") + public DeferredResult start() { + DeferredResult result = new DeferredResult<>(); + result.setResult("started"); + return result; + } + + @GetMapping("/diag/async-timeout") + public Map asyncTimeout(HttpServletRequest request) { + Map m = new LinkedHashMap<>(); + m.put("spring.mvc.async.request-timeout", configured); + m.put("servletContainer", request.getServletContext().getServerInfo()); + m.put("effectiveAsyncTimeoutMs", LAST_TIMEOUT.get()); + return m; + } + + @Component + static class CaptureAsyncTimeout implements AsyncHandlerInterceptor { + + @Override + public void afterConcurrentHandlingStarted(HttpServletRequest request, + HttpServletResponse response, Object handler) { + if (request.isAsyncStarted()) { + LAST_TIMEOUT.set(request.getAsyncContext().getTimeout()); + } + } + } + + @Configuration + static class RegisterInterceptor implements WebMvcConfigurer { + + private final CaptureAsyncTimeout interceptor; + + RegisterInterceptor(CaptureAsyncTimeout interceptor) { + this.interceptor = interceptor; + } + + @Override + public void addInterceptors(InterceptorRegistry registry) { + registry.addInterceptor(interceptor); + } + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/ChatController.java b/sse-websocket/src/main/java/com/ankurm/ssews/ChatController.java new file mode 100644 index 0000000..ea74a40 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/ChatController.java @@ -0,0 +1,70 @@ +package com.ankurm.ssews; + +import java.security.Principal; +import java.time.Instant; + +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.simp.SimpMessageHeaderAccessor; +import org.springframework.messaging.simp.SimpMessageType; +import org.springframework.messaging.simp.SimpMessagingTemplate; +import org.springframework.stereotype.Controller; + +/** + * The chat. Three handlers, each demonstrating a different routing rule. + * + * @see docs/04-stomp.md + */ +@Controller +public class ChatController { + + private final SimpMessagingTemplate template; + + ChatController(SimpMessagingTemplate template) { + this.template = template; + } + + /** Explicit fan-out destination. Client SENDs to /app/chat.send, subscribers get /topic/room. */ + @MessageMapping("/chat.send") + @SendTo("/topic/room") + public ChatMessage send(@Payload ChatMessage in, SimpMessageHeaderAccessor headers) { + String from = in.from() == null ? headers.getSessionId() : in.from(); + return new ChatMessage(from, in.text(), Instant.now()); + } + + /** + * No {@code @SendTo}. The return value still goes somewhere: the default destination is the + * broker prefix plus the mapping, i.e. {@code /topic/chat.echo}. That default is why a + * handler you thought was private is readable by anyone who guesses the path. + */ + @MessageMapping("/chat.echo") + public ChatMessage echo(@Payload ChatMessage in) { + return new ChatMessage("echo", in.text(), Instant.now()); + } + + /** + * Point-to-point. Sent to one session's private destination; every other subscriber to + * /user/queue/whisper gets nothing, because the broker rewrote the destination per session. + * + *

The three extra lines are not optional and their absence is silent. With no + * authenticated {@link Principal} the "user" is the STOMP session id, and + * {@code DefaultUserDestinationResolver} can only turn a session id into a real destination + * if the message also carries that session id in its headers. Call the two-argument + * {@code convertAndSendToUser(sessionId, dest, payload)} instead and the message is resolved + * against a user registry that has never heard of that name: no exception, no log line, no + * delivery. {@code setLeaveMutable(true)} is required too — without it the accessor is + * immutable by the time the resolver looks, and the session id is lost again. + */ + @MessageMapping("/chat.whisper") + public void whisper(@Payload ChatMessage in, SimpMessageHeaderAccessor headers, Principal principal) { + String targetSessionId = in.from(); + SimpMessageHeaderAccessor out = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE); + out.setSessionId(targetSessionId); + out.setLeaveMutable(true); + template.convertAndSendToUser(targetSessionId, "/queue/whisper", + new ChatMessage(principal == null ? headers.getSessionId() : principal.getName(), + in.text(), Instant.now()), + out.getMessageHeaders()); + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/ChatMessage.java b/sse-websocket/src/main/java/com/ankurm/ssews/ChatMessage.java new file mode 100644 index 0000000..a53e2d0 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/ChatMessage.java @@ -0,0 +1,7 @@ +package com.ankurm.ssews; + +import java.time.Instant; + +/** A single chat line, carried as the STOMP frame body. */ +public record ChatMessage(String from, String text, Instant at) { +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/DashboardSseController.java b/sse-websocket/src/main/java/com/ankurm/ssews/DashboardSseController.java new file mode 100644 index 0000000..82b8040 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/DashboardSseController.java @@ -0,0 +1,117 @@ +package com.ankurm.ssews; + +import java.io.IOException; +import java.time.Instant; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicLong; + +import org.springframework.http.MediaType; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * The live dashboard, and four smaller endpoints that exist only so the article can quote a + * transcript rather than assert a behaviour. + * + * @see docs/02-sse-lifecycle.md + */ +@RestController +public class DashboardSseController { + + private final EmitterRegistry registry; + private final AtomicLong seq = new AtomicLong(); + + DashboardSseController(EmitterRegistry registry) { + this.registry = registry; + } + + /** + * The dashboard stream. Subscribers get whatever the scheduled broadcaster pushes. + * + *

{@code Long.MAX_VALUE} disables the async timeout for this handler. That is the right + * call for a stream that is supposed to stay open, and it is also the line that turns a + * leaked emitter into a permanent leak — which is why {@link EmitterRegistry} removes + * on all three callbacks. + */ + @GetMapping(path = "/sse/metrics", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter metrics() { + return registry.register(UUID.randomUUID().toString(), new SseEmitter(Long.MAX_VALUE)); + } + + /** + * Three events and a close, written from the request thread, so a curl transcript shows the + * exact wire format: field names, the blank-line record separator, and where {@code retry} + * and {@code id} land. + */ + @GetMapping(path = "/sse/raw", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter raw() throws IOException { + SseEmitter emitter = new SseEmitter(10_000L); + emitter.send(SseEmitter.event().comment("stream open")); + emitter.send(SseEmitter.event() + .id("1").name("metric").reconnectTime(3000) + .data(new MetricSnapshot(1, "node-a", 0.42, 512, Instant.parse("2026-09-03T10:00:00Z")), + MediaType.APPLICATION_JSON)); + // Multi-line data is split into several data: lines by the SSE spec, and the browser + // rejoins them with "\n". Worth seeing once. + emitter.send(SseEmitter.event().id("2").name("note").data("line one\nline two")); + emitter.complete(); + return emitter; + } + + /** + * Sends the same record twice: once with an explicit JSON media type and once with none. + * The difference is the whole of {@code docs/03-the-payload.md}. + */ + @GetMapping(path = "/sse/payload", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter payload() throws IOException { + SseEmitter emitter = new SseEmitter(10_000L); + MetricSnapshot snap = + new MetricSnapshot(7, "node-a", 0.42, 512, Instant.parse("2026-09-03T10:00:00Z")); + emitter.send(SseEmitter.event().name("with-json").data(snap, MediaType.APPLICATION_JSON)); + emitter.send(SseEmitter.event().name("no-media-type").data(snap)); + // The same two calls for a String payload, where the answer is different: a String is + // written by StringHttpMessageConverter unless a JSON media type forces Jackson, and + // Jackson then quotes and escapes it. + emitter.send(SseEmitter.event().name("string-plain").data("{\"already\":\"json\"}")); + emitter.send(SseEmitter.event().name("string-json") + .data("{\"already\":\"json\"}", MediaType.APPLICATION_JSON)); + emitter.complete(); + return emitter; + } + + /** + * Opens a stream, writes one event, then never writes again. With a short timeout this shows + * what a client actually observes when the async request times out mid-stream. + */ + @GetMapping(path = "/sse/silent", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter silent(@RequestParam(defaultValue = "1500") long timeoutMs) throws IOException { + SseEmitter emitter = registry.register(UUID.randomUUID().toString(), new SseEmitter(timeoutMs)); + emitter.send(SseEmitter.event().name("hello").data("then silence")); + return emitter; + } + + /** + * Echoes the {@code Last-Event-ID} request header. The browser sends it automatically on + * reconnect; the server is expected to replay from there, and almost nothing does. + */ + @GetMapping(path = "/sse/resume", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter resume(@RequestHeader(name = "Last-Event-ID", required = false) String lastEventId) + throws IOException { + SseEmitter emitter = new SseEmitter(10_000L); + long from = lastEventId == null ? 0 : Long.parseLong(lastEventId); + emitter.send(SseEmitter.event().name("resumed") + .data("Last-Event-ID=" + lastEventId + " -> replaying from " + (from + 1))); + for (long i = from + 1; i <= from + 2; i++) { + emitter.send(SseEmitter.event().id(Long.toString(i)).name("metric").data("event " + i)); + } + emitter.complete(); + return emitter; + } + + long nextSeq() { + return seq.incrementAndGet(); + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/DiagnosticsController.java b/sse-websocket/src/main/java/com/ankurm/ssews/DiagnosticsController.java new file mode 100644 index 0000000..8db8472 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/DiagnosticsController.java @@ -0,0 +1,81 @@ +package com.ankurm.ssews; + +import java.util.LinkedHashMap; +import java.util.Map; + +import java.util.List; + +import org.springframework.beans.factory.annotation.Value; +import jakarta.servlet.ServletContext; +import jakarta.websocket.server.ServerContainer; +import org.springframework.http.converter.HttpMessageConverter; +import org.springframework.messaging.simp.user.SimpUserRegistry; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RestController; + +/** + * Prints the runtime state that is otherwise invisible: how many SSE streams are open, how many + * writes discovered a dead one, and how many WebSocket sessions the broker believes it has. + * + *

Delete this before shipping. It is here because "the emitter map is leaking" and "the + * broker still thinks that user is connected" are both conditions you cannot see from the + * outside, and both are much easier to argue about with a number in front of you. + */ +@RestController +public class DiagnosticsController { + + private final EmitterRegistry registry; + private final SimpUserRegistry userRegistry; + private final List> converters; + private final String asyncTimeout; + private final String tomcatMaxThreads; + + DiagnosticsController(EmitterRegistry registry, + SimpUserRegistry userRegistry, + org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter adapter, + @Value("${spring.mvc.async.request-timeout:}") String asyncTimeout, + @Value("${server.tomcat.threads.max:}") String tomcatMaxThreads) { + this.registry = registry; + this.userRegistry = userRegistry; + this.converters = adapter.getMessageConverters(); + this.asyncTimeout = asyncTimeout; + this.tomcatMaxThreads = tomcatMaxThreads; + } + + @org.springframework.beans.factory.annotation.Autowired + ServletContext servletContext; + + @GetMapping("/diag") + public Map diag() { + Map m = new LinkedHashMap<>(); + m.put("sse.open", registry.open()); + m.put("sse.opened", registry.opened()); + m.put("sse.completed", registry.completed()); + m.put("sse.timedOut", registry.timedOut()); + m.put("sse.errored", registry.errored()); + m.put("sse.sendsOk", registry.sendsOk()); + m.put("sse.sendsFailed", registry.sendsFailed()); + m.put("ws.users", userRegistry.getUserCount()); + m.put("config.spring.mvc.async.request-timeout", asyncTimeout); + m.put("config.server.tomcat.threads.max", tomcatMaxThreads); + // The order that decides what SseEmitter.send(data, mediaType) actually does: the first + // converter whose canWrite(type, mediaType) returns true wins, and the media type you + // passed is only a filter, never a preference. + m.put("http.messageConverters", + converters.stream().map(c -> c.getClass().getSimpleName()).toList()); + + // The number that actually decides how big a STOMP frame can be. It belongs to the + // servlet container, not to Spring, and Spring's own messageSizeLimit never sees a + // frame the container refused to assemble. + Object sc = servletContext.getAttribute("jakarta.websocket.server.ServerContainer"); + if (sc instanceof ServerContainer container) { + m.put("ws.container.defaultMaxTextMessageBufferSize", + container.getDefaultMaxTextMessageBufferSize()); + m.put("ws.container.defaultMaxBinaryMessageBufferSize", + container.getDefaultMaxBinaryMessageBufferSize()); + m.put("ws.container.defaultMaxSessionIdleTimeout", + container.getDefaultMaxSessionIdleTimeout()); + } + return m; + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/EmitterRegistry.java b/sse-websocket/src/main/java/com/ankurm/ssews/EmitterRegistry.java new file mode 100644 index 0000000..ff6c143 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/EmitterRegistry.java @@ -0,0 +1,98 @@ +package com.ankurm.ssews; + +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * The piece every SSE tutorial leaves out: something that owns the open emitters and removes + * them again. + * + *

An {@link SseEmitter} is not a subscription. Nothing in Spring keeps a list of them, and + * nothing tells you when one dies — you find out on the next write, as an + * {@link IOException} thrown from {@code send()} on whichever thread happened to be + * broadcasting. If that thread is a scheduler thread and the exception escapes, the schedule + * stops and the whole dashboard silently freezes for everyone. + * + *

The counters here are exposed by {@link DiagnosticsController} so the article can quote + * real numbers for "how many streams were open, and how many writes discovered a dead one". + * + * @see docs/02-sse-lifecycle.md + */ +@Component +public class EmitterRegistry { + + private static final Logger log = LoggerFactory.getLogger(EmitterRegistry.class); + + private final Map emitters = new ConcurrentHashMap<>(); + + private final AtomicLong opened = new AtomicLong(); + private final AtomicLong completed = new AtomicLong(); + private final AtomicLong timedOut = new AtomicLong(); + private final AtomicLong errored = new AtomicLong(); + private final AtomicLong sendsOk = new AtomicLong(); + private final AtomicLong sendsFailed = new AtomicLong(); + + public SseEmitter register(String id, SseEmitter emitter) { + emitters.put(id, emitter); + opened.incrementAndGet(); + + // All three callbacks must remove the emitter. Registering only onCompletion is the + // most common variant of this bug: a timed-out emitter is never completed by the + // client, so it stays in the map for ever and every broadcast keeps paying for it. + emitter.onCompletion(() -> { emitters.remove(id); completed.incrementAndGet(); }); + emitter.onTimeout(() -> { + emitters.remove(id); + timedOut.incrementAndGet(); + // Spring completes the async request on timeout, but call complete() so the + // onCompletion callback semantics stay consistent for our own bookkeeping. + emitter.complete(); + }); + emitter.onError(t -> { emitters.remove(id); errored.incrementAndGet(); }); + return emitter; + } + + /** + * Broadcast to everyone, dropping the streams that turn out to be gone. + * + *

The try/catch is not defensive programming; it is the only way a dead client is + * detected. Note also that this is where SSE's cost lives: one serialise-and-write per + * open stream, on the calling thread. + */ + public int broadcast(String eventName, MetricSnapshot payload) { + int delivered = 0; + for (Map.Entry e : emitters.entrySet()) { + try { + e.getValue().send(SseEmitter.event() + .id(Long.toString(payload.seq())) + .name(eventName) + .reconnectTime(3000) + .data(payload, org.springframework.http.MediaType.APPLICATION_JSON)); + sendsOk.incrementAndGet(); + delivered++; + } + catch (IOException | IllegalStateException ex) { + // IOException -> the socket is gone (client closed the tab, proxy reaped it). + // IllegalState -> the emitter already completed or timed out on another thread. + sendsFailed.incrementAndGet(); + emitters.remove(e.getKey()); + log.debug("dropping emitter {}: {}", e.getKey(), ex.getClass().getSimpleName()); + } + } + return delivered; + } + + public int open() { return emitters.size(); } + public long opened() { return opened.get(); } + public long completed() { return completed.get(); } + public long timedOut() { return timedOut.get(); } + public long errored() { return errored.get(); } + public long sendsOk() { return sendsOk.get(); } + public long sendsFailed() { return sendsFailed.get(); } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/MetricSnapshot.java b/sse-websocket/src/main/java/com/ankurm/ssews/MetricSnapshot.java new file mode 100644 index 0000000..f4c4c09 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/MetricSnapshot.java @@ -0,0 +1,14 @@ +package com.ankurm.ssews; + +import java.time.Instant; + +/** + * One row of the live dashboard. + * + *

{@link Instant} is here on purpose: it is the field that fails to serialise if the wrong + * converter is chosen for the SSE payload, which is what {@code SseDataConversionTest} pins. + * + * @see docs/03-the-payload.md + */ +public record MetricSnapshot(long seq, String host, double cpu, long heapMb, Instant at) { +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/MetricsBroadcaster.java b/sse-websocket/src/main/java/com/ankurm/ssews/MetricsBroadcaster.java new file mode 100644 index 0000000..447db3b --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/MetricsBroadcaster.java @@ -0,0 +1,42 @@ +package com.ankurm.ssews; + +import java.time.Instant; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicLong; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +/** + * Pushes one snapshot to every open dashboard stream on a fixed schedule. + * + *

Fan-out happens on the scheduler thread, which is the shape almost every SSE dashboard + * ends up with and also its first scaling limit: 1 000 open streams means 1 000 + * serialise-and-write calls in a row on one thread, every tick. + * {@code docs/06-choosing.md} has the measurement. + */ +@Component +@ConditionalOnProperty(name = "dashboard.broadcast.enabled", havingValue = "true", matchIfMissing = true) +public class MetricsBroadcaster { + + private final EmitterRegistry registry; + private final AtomicLong seq = new AtomicLong(); + + MetricsBroadcaster(EmitterRegistry registry) { + this.registry = registry; + } + + @Scheduled(fixedRateString = "${dashboard.broadcast.interval-ms:500}") + public void tick() { + if (registry.open() == 0) { + return; + } + registry.broadcast("metric", new MetricSnapshot( + seq.incrementAndGet(), + "node-a", + Math.round(ThreadLocalRandom.current().nextDouble() * 10000) / 10000.0, + 256 + ThreadLocalRandom.current().nextInt(256), + Instant.now())); + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/PresenceListener.java b/sse-websocket/src/main/java/com/ankurm/ssews/PresenceListener.java new file mode 100644 index 0000000..b878fda --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/PresenceListener.java @@ -0,0 +1,47 @@ +package com.ankurm.ssews; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.springframework.context.event.EventListener; +import org.springframework.messaging.simp.stomp.StompHeaderAccessor; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.messaging.SessionConnectedEvent; +import org.springframework.web.socket.messaging.SessionDisconnectEvent; +import org.springframework.web.socket.messaging.SessionSubscribeEvent; + +/** + * Presence, and the reason people reach for STOMP over SSE in the first place: the server knows + * who is connected, who subscribed to what, and when they left. + * + *

SSE has no equivalent. There is no disconnect event — only a write that fails later. + */ +@Component +public class PresenceListener { + + private final List log = new CopyOnWriteArrayList<>(); + + @EventListener + public void onConnected(SessionConnectedEvent e) { + log.add("CONNECTED " + StompHeaderAccessor.wrap(e.getMessage()).getSessionId()); + } + + @EventListener + public void onSubscribe(SessionSubscribeEvent e) { + StompHeaderAccessor h = StompHeaderAccessor.wrap(e.getMessage()); + log.add("SUBSCRIBE " + h.getSessionId() + " -> " + h.getDestination()); + } + + @EventListener + public void onDisconnect(SessionDisconnectEvent e) { + log.add("DISCONNECT " + e.getSessionId() + " status=" + e.getCloseStatus()); + } + + public List events() { + return List.copyOf(log); + } + + public void clear() { + log.clear(); + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/SseWebSocketApplication.java b/sse-websocket/src/main/java/com/ankurm/ssews/SseWebSocketApplication.java new file mode 100644 index 0000000..5f2848e --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/SseWebSocketApplication.java @@ -0,0 +1,25 @@ +package com.ankurm.ssews; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.scheduling.annotation.EnableScheduling; + +/** + * Both halves of the article in one application: a Server-Sent Events dashboard under + * {@code /sse/**} and a STOMP-over-WebSocket chat under {@code /ws}. + * + *

They coexist deliberately. The point of the companion article is that they are not + * competitors so much as different answers to "who needs to talk", and running them side by + * side makes the asymmetry visible: the SSE endpoints are plain MVC handler methods with a + * return type, while the chat needs a broker, a message-converter stack and a session registry. + * + * @see docs/01-two-protocols.md + */ +@SpringBootApplication +@EnableScheduling +public class SseWebSocketApplication { + + public static void main(String[] args) { + SpringApplication.run(SseWebSocketApplication.class, args); + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/WebSocketBufferConfig.java b/sse-websocket/src/main/java/com/ankurm/ssews/WebSocketBufferConfig.java new file mode 100644 index 0000000..58ebf21 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/WebSocketBufferConfig.java @@ -0,0 +1,35 @@ +package com.ankurm.ssews; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Profile; +import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean; + +/** + * The other half of the message size limit — the half that is not in Spring. + * + *

{@code WebSocketTransportRegistration.setMessageSizeLimit(64 * 1024)} is not the limit that + * bites first. Spring's {@code SubProtocolWebSocketHandler} does not support partial messages, + * so the servlet container has to buffer each WebSocket text message whole before handing it + * over — and Tomcat's buffer defaults to 8 192 bytes. Cross that and + * Tomcat closes the connection with WebSocket status 1009 before Spring ever + * sees the frame, so raising Spring's limit alone changes nothing. + * + *

Active under the {@code bigframes} profile so the companion test can show both behaviours + * from one build. + * + * @see docs/05-limits.md + */ +@Configuration +@Profile("bigframes") +public class WebSocketBufferConfig { + + @Bean + public ServletServerContainerFactoryBean createWebSocketContainer( + @org.springframework.beans.factory.annotation.Value("${ws.buffer-bytes:262144}") int bytes) { + ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); + container.setMaxTextMessageBufferSize(bytes); + container.setMaxBinaryMessageBufferSize(bytes); + return container; + } +} diff --git a/sse-websocket/src/main/java/com/ankurm/ssews/WebSocketConfig.java b/sse-websocket/src/main/java/com/ankurm/ssews/WebSocketConfig.java new file mode 100644 index 0000000..e815398 --- /dev/null +++ b/sse-websocket/src/main/java/com/ankurm/ssews/WebSocketConfig.java @@ -0,0 +1,62 @@ +package com.ankurm.ssews; + +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.simp.config.MessageBrokerRegistry; +import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; +import org.springframework.web.socket.config.annotation.StompEndpointRegistry; +import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; +import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration; + +/** + * Four lines of configuration that decide more than they look like they do. + * + * @see docs/04-stomp.md + */ +@Configuration +@EnableWebSocketMessageBroker +public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { + + @Override + public void registerStompEndpoints(StompEndpointRegistry registry) { + // No setAllowedOrigins call at all. The handshake then permits same-origin requests + // only -- and "same origin" is decided from the Origin header, which a non-browser + // client does not send, so curl and the Java client are allowed while a page on + // another site is not. Add setAllowedOriginPatterns("https://app.example.com") for a + // real cross-origin front end; setAllowedOrigins("*") is rejected outright when + // credentials are allowed. + registry.addEndpoint("/ws"); + + // The SockJS variant is a SECOND endpoint on its own path. Registering .withSockJS() + // on the same path does not give you both: SockJS wraps the URL in /{server}/{session} + // and a plain WebSocket client pointed at it fails the handshake. + registry.addEndpoint("/ws-sockjs").withSockJS(); + } + + @Override + public void configureMessageBroker(MessageBrokerRegistry registry) { + // The simple broker. In-memory, single JVM, no persistence, no acknowledgement, and it + // drops everything on restart. enableStompBrokerRelay(..) swaps in RabbitMQ or ActiveMQ + // without touching a controller -- see ../rabbitmq for the broker side. + registry.enableSimpleBroker("/topic", "/queue"); + + // Destinations a CLIENT sends to. /app/chat.send is routed to @MessageMapping; anything + // starting with /topic or /queue goes straight to the broker and is NEVER seen by a + // controller. Getting this backwards produces a message that vanishes with no error. + registry.setApplicationDestinationPrefixes("/app"); + + // The prefix a client subscribes under for messages addressed to it alone. The client + // subscribes to /user/queue/whisper; the broker rewrites that to a session-scoped + // destination that no other session can subscribe to. + registry.setUserDestinationPrefix("/user"); + } + + @Override + public void configureWebSocketTransport(WebSocketTransportRegistration registration) { + // Defaults, restated so the numbers are visible in one place rather than discovered + // when a 70 KB frame disappears. WebSocketTransportRegistration's own defaults are + // 64 KB for both; see WebSocketLimitsTest for what exceeding them looks like. + registration.setMessageSizeLimit(64 * 1024); + registration.setSendBufferSizeLimit(512 * 1024); + registration.setSendTimeLimit(20_000); + } +} diff --git a/sse-websocket/src/main/resources/application.yaml b/sse-websocket/src/main/resources/application.yaml new file mode 100644 index 0000000..7730cd9 --- /dev/null +++ b/sse-websocket/src/main/resources/application.yaml @@ -0,0 +1,19 @@ +spring: + application: + name: sse-websocket + # spring.mvc.async.request-timeout is deliberately ABSENT, not set to blank. When it is + # absent Spring never calls AsyncContext.setTimeout(..), so the servlet container's own + # default applies. /diag/async-timeout reports what that turns out to be on this container; + # an SseEmitter constructor argument overrides both. + +server: + port: 8080 + +dashboard: + broadcast: + enabled: true + interval-ms: 500 + +logging: + level: + org.springframework.web.socket: INFO diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/AsyncTimeoutPropertyTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/AsyncTimeoutPropertyTest.java new file mode 100644 index 0000000..17324ad --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/AsyncTimeoutPropertyTest.java @@ -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"); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/AsyncTimeoutUnsetTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/AsyncTimeoutUnsetTest.java new file mode 100644 index 0000000..814c999 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/AsyncTimeoutUnsetTest.java @@ -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\":\"\""); + assertThat(body).contains("\"effectiveAsyncTimeoutMs\":30000"); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/SizeProbe.java b/sse-websocket/src/test/java/com/ankurm/ssews/SizeProbe.java new file mode 100644 index 0000000..3d05804 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/SizeProbe.java @@ -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. + * + *

The client's own buffers are raised to 1 MB first. Without that the client hits the + * identical 8 KB default and closes its own session, which looks exactly like the server + * rejecting the message — 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 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}; + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/SseClient.java b/sse-websocket/src/test/java/com/ankurm/ssews/SseClient.java new file mode 100644 index 0000000..13e88f2 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/SseClient.java @@ -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. + * + *

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 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 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 readToEnd() throws Exception { + String line; + while ((line = reader.readLine()) != null) { + lines.add(line); + } + return List.copyOf(lines); + } + + List lines() { + return List.copyOf(lines); + } + + @Override + public void close() throws Exception { + socket.close(); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/SseConcurrencyTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/SseConcurrencyTest.java new file mode 100644 index 0000000..c79bd3c --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/SseConcurrencyTest.java @@ -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. + * + *

This is the test that answers the objection people raise first — "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. + * + *

What SSE does cost is a socket 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 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-. 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) { } + } + } + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/SseDataConversionTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/SseDataConversionTest.java new file mode 100644 index 0000000..d310aef --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/SseDataConversionTest.java @@ -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 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 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); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/SseDisconnectTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/SseDisconnectTest.java new file mode 100644 index 0000000..209f1fb --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/SseDisconnectTest.java @@ -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. + * + *

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) + " ==="); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/SseTimeoutTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/SseTimeoutTest.java new file mode 100644 index 0000000..638409d --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/SseTimeoutTest.java @@ -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. + * + *

The answer is the point: nothing. No error event, no status code, no trailer — 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 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); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/SseWireFormatTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/SseWireFormatTest.java new file mode 100644 index 0000000..9aed982 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/SseWireFormatTest.java @@ -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 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"); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/StompChatTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/StompChatTest.java new file mode 100644 index 0000000..ad8dbeb --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/StompChatTest.java @@ -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 aliceInbox = + StompTestSupport.subscribe(alice, "/topic/room", ChatMessage.class); + BlockingQueue 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(); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/StompRoutingTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/StompRoutingTest.java new file mode 100644 index 0000000..fabd67c --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/StompRoutingTest.java @@ -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 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 — subscribers just get + * whatever the client sent. + */ + @Test + void sendingToTopicSkipsTheController() throws Exception { + WebSocketStompClient client = StompTestSupport.client(); + StompSession session = StompTestSupport.connect(client, port); + + BlockingQueue 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(); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitDefaultBufferTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitDefaultBufferTest.java new file mode 100644 index 0000000..5313993 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitDefaultBufferTest.java @@ -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. + * + *

The documented knob is {@code WebSocketTransportRegistration.setMessageSizeLimit}, 64 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 — 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); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitSpringLimitTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitSpringLimitTest.java new file mode 100644 index 0000000..ecb3c9a --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitSpringLimitTest.java @@ -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 KB, Spring's own 64 KB limit is finally the one that + * decides — 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"); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitTunedBufferTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitTunedBufferTest.java new file mode 100644 index 0000000..56e7d9e --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/StompSizeLimitTunedBufferTest.java @@ -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=} to move it; the + * committed transcript covers 8 KB (untouched), 16 KB, 32 KB and 64 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); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/StompTestSupport.java b/sse-websocket/src/test/java/com/ankurm/ssews/StompTestSupport.java new file mode 100644 index 0000000..63c5bfb --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/StompTestSupport.java @@ -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 — handshake, CONNECT/CONNECTED, SUBSCRIBE, SEND, MESSAGE. + * + *

Note the converter: {@link JacksonJsonMessageConverter} is the Jackson 3 one. + * {@code MappingJackson2MessageConverter} is still on the classpath and is the Jackson 2 + * one; under Boot 4 a "2" in a Spring class name means the previous 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 BlockingQueue subscribe(StompSession session, String destination, Class type) { + BlockingQueue 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; + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/StompUserDestinationTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/StompUserDestinationTest.java new file mode 100644 index 0000000..73c1373 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/StompUserDestinationTest.java @@ -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. + * + *

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 aliceInbox = + StompTestSupport.subscribe(alice, "/user/queue/whisper", ChatMessage.class); + BlockingQueue 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 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(); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketContainerDefaultsTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketContainerDefaultsTest.java new file mode 100644 index 0000000..0ffe16e --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketContainerDefaultsTest.java @@ -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"); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketContainerRaisedTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketContainerRaisedTest.java new file mode 100644 index 0000000..9f44d15 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketContainerRaisedTest.java @@ -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"); + } +} diff --git a/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketHandshakeOriginTest.java b/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketHandshakeOriginTest.java new file mode 100644 index 0000000..ccf31b2 --- /dev/null +++ b/sse-websocket/src/test/java/com/ankurm/ssews/WebSocketHandshakeOriginTest.java @@ -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. + * + *

This matters because the WebSocket handshake is a plain HTTP GET, and the browser's + * same-origin policy does not 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 lines = new ArrayList<>(); + String line; + while ((line = in.readLine()) != null && !line.isEmpty()) { + lines.add(line); + } + return lines.isEmpty() ? "" : lines.get(0).trim(); + } + } +}