A server {@code WebSocketHandler} is mapped to requests with + * Handler for a WebSocket messages. You can use it as follows: + *
Use {@link WebSocketSession#receive() session.receive()} to compose on - * the inbound message stream, and {@link WebSocketSession#send(Publisher) - * session.send(publisher)} for the outbound message stream. Below is an - * example, combined flow to process inbound and to send outbound messages: + *
{@link WebSocketSession#receive() session.receive()} handles inbound + * messages, while {@link WebSocketSession#send(Publisher) session.send} + * sends outbound messages. Below is an example of handling inbound messages + * and responding to every message: * *
- * class ExampleHandler implements WebSocketHandler {
+ * class ExampleHandler implements WebSocketHandler {
*
- * @Override
- * public Mono<Void> handle(WebSocketSession session) {
- *
- * Flux<WebSocketMessage> output = session.receive()
- * .doOnNext(message -> {
- * // This is for side effects such as
- * // - Logging incoming messages
- * // - Updating some metrics or counters
- * // - Performing access checks or validations (non-blocking)
- * System.out.println("Got message: " + message.getPayloadAsText());
- * })
- * .concatMap(message -> {
- * // This is where you handle the actual processing of the incoming message. It
- * // might involve:
- * // - Parsing the message content (e.g., JSON parsing)
- * // - Invoking a reactive service (e.g., database, HTTP call, etc.)
- * // - Returning a transformed value, typically a Mono<String> or Mono<SomeType>
- * // if you're mapping to another data format
- * return Mono.just(message.getPayloadAsText());
- * })
- * .map(value -> {
- * // This is where you produce one or more responses for the message
- * return session.textMessage("Echo " + value));
- * });
- *
- * return session.send(output);
- * }
- * }
+ * @Override
+ * public Mono<Void> handle(WebSocketSession session) {
+ * Flux<WebSocketMessage> output = session.receive()
+ * .doOnNext(message -> {
+ * // Imperative calls without a return value:
+ * // perform access checks, log, validate, update metrics.
+ * // ...
+ * })
+ * .concatMap(message -> {
+ * // Async, non-blocking calls:
+ * // parse messages, call a database, make remote calls.
+ * // Return the same message, or a transformed value
+ * // ...
+ * });
+ * return session.send(output);
+ * }
+ * }
*
*
* If processing inbound and sending outbound messages are independent * streams, they can be joined together with the "zip" operator: * *
- * class ExampleHandler implements WebSocketHandler {
+ * class ExampleHandler implements WebSocketHandler {
*
- * @Override
- * public Mono<Void> handle(WebSocketSession session) {
+ * @Override
+ * public Mono<Void> handle(WebSocketSession session) {
*
- * Mono<Void> input = session.receive()
- * .doOnNext(message -> {
- * // ...
- * })
- * .concatMap(message -> {
- * // ...
- * })
- * .then();
+ * Mono<Void> input = session.receive()
+ * .doOnNext(message -> {
+ * // ...
+ * })
+ * .concatMap(message -> {
+ * // ...
+ * })
+ * .then();
*
- * Flux<String> source = ... ;
- * Mono<Void> output = session.send(source.map(session::textMessage));
+ * Flux<String> source = ... ;
+ * Mono<Void> output = session.send(source.map(session::textMessage));
*
- * return Mono.zip(input, output).then();
- * }
- * }
+ * return Mono.zip(input, output).then();
+ * }
+ * }
*
*
* A {@code WebSocketHandler} must compose the inbound and outbound streams