How do WebSocketSession.receive() and WebSocketSession.send() work with Flux<WebSocketMessage>, and how do you wire them together in a handler?

## The two streams of a session `WebSocketSession` models the live connection as a duplex pair of reactive streams: - **Inbound:** `Flux<WebSocketMessage> receive()` — emits one `WebSocketMessage` per frame the client sends, completes when the client closes the connection, errors on transport failure. - **Outbound:** `Mono<Void> send(Publisher<WebSocketMessage> messages)` — you hand it a publisher; the framework writes each emitted message to the wire, applying backpressure, and the returned `Mono<Void>` completes when your source `Publisher` completes. ## WebSocketMessage `WebSocketMessage` wraps a payload plus a `Type` (`TEXT`, `BINARY`, `PING`, `PONG`). Reading: - `getPayloadAsText()` → decodes the payload `DataBuffer` as UTF-8 text. - `getPayload()` → the raw `org.springframework.core.io.buffer.DataBuffer`. - `getType()` → the frame type. Creating messages — **use the session factory methods** so the correct `DataBufferFactory` is used: - `session.textMessage(String)` - `session.binaryMessage(Function<DataBufferFactory, DataBuffer>)` - `session.pingMessage(...)` / `session.pongMessage(...)` ## Wiring patterns **1. Echo (inbound drives outbound):** ```java return session.send(session.receive().map(m -> session.textMessage(m.getPayloadAsText()))); ``` **2. Server push / outbound-only** (e.g., a ticker) — you ignore `receive()` but note the caveat below: ```java Flux<WebSocketMessage> ticks = Flux.interval(Duration.ofSeconds(1)) .map(i -> session.textMessage("tick-" + i)); return session.send(ticks); ``` **3. Bidirectional / independent in and out.** `receive()` and `send()` are separate streams, but you must return **one** `Mono<Void>`. Combine them so both run and the session closes when appropriate: ```java Mono<Void> input = session.receive() .doOnNext(m -> process(m.getPayloadAsText())) .then(); // completes, emits nothing Mono<Void> output = session.send(outboundFlux); return Mono.zip(input, output).then(); // or Flux.merge(input, output).then() ``` Use `Mono.zip`/`Flux.merge` when you want the session to end when *either*/*both* streams end depending on your semantics. ## Backpressure and threading - Backpressure is honored: `send()` only pulls from your outbound publisher as fast as the socket can write; if the client is slow, demand slows upstream. Unbounded hot sources (e.g., a broadcast sink) should use `onBackpressureBuffer`/`onBackpressureDrop` to avoid overwhelming a slow client. - **You generally get one `send()` subscription per session.** Don't call `send()` multiple times concurrently and don't write to the session from multiple threads — sessions are not designed for concurrent writes. Merge everything into a single outbound `Flux`. ## DataBuffer lifecycle — the classic leak Inbound `WebSocketMessage` payloads are backed by **pooled** `DataBuffer`s (especially on Reactor Netty). The framework **releases** the buffer once the message has been consumed within your `receive()` subscription. Two consequences: - If you **cache/delay/defer** a received message beyond the immediate stream (e.g., buffer, replay, hand to another async boundary), you must **retain** it (`DataBufferUtils.retain(...)`) and release later, or you get use-after-free / leak warnings. - If you call `getPayloadAsText()` eagerly in the stream (the common case), you're safe because you extract the String before the buffer is released. - Also: if you **never subscribe to `receive()`** in an inbound-capable handler, incoming frames may not be drained. For a pure push handler this is usually fine, but if the client sends data, prefer `session.receive().then()` merged into your pipeline to drain and release buffers. ## When to use raw receive()/send() Use it for low-level, per-frame control (custom binary protocols, simple echo/push, proxies). For structured pub/sub messaging use STOMP or RSocket (different leaves).