handleWebSocketMessages

def handleWebSocketMessages(handler: Flow[Message, Message, Any]): Route 
def handleWebSocketMessages( handler: Flow[Message, Message, Any], shouldCompress: Message 

Description

The directive first checks if the request was a valid WebSocket handshake request and if yes, it completes the request with the passed handler. Otherwise, the request is rejected with an ExpectedWebSocketRequestRejectionExpectedWebSocketRequestRejection.

The overload that accepts a shouldCompress filter can select compression separately for each outbound message after permessage-deflate is negotiated. It is evaluated synchronously once for each outbound text or binary message, and its result applies to every fragment of that message. The filter is not invoked for control frames or when compression was not negotiated. It should be fast and non-blocking; throwing from it fails the WebSocket stream.

For a streamed message, the complete payload and its final size are not available when the filter is evaluated. Applications that select compression based on message size therefore need an explicit policy for streamed messages.

WebSocket subprotocols offered in the Sec-WebSocket-Protocol header of the request are ignored. If you want to support several protocols use the handleWebSocketMessagesForProtocol directive, instead.

For more information about the WebSocket support, see Server-Side WebSocket Support.

Example

Scala

copysourcedef greeter: Flow[Message, Message, Any] =
  Flow[Message].mapConcat {
    case tm: TextMessage =>
      TextMessage(Source.single("Hello ") ++ tm.textStream ++ Source.single("!")) :: Nil
    case bm: BinaryMessage =>
      // ignore binary messages but drain content to avoid the stream being clogged
      bm.dataStream.runWith(Sink.ignore)
      Nil
  }
val websocketRoute =
  path("greeter") {
    handleWebSocketMessages(greeter)
  }

// tests:
// create a testing probe representing the client-side
val wsClient = WSProbe()

// WS creates a WebSocket request for testing
WS("/greeter", wsClient.flow) ~> websocketRoute ~>
check {
  // check response for WS Upgrade headers
  isWebSocketUpgrade shouldEqual true

  // manually run a WS conversation
  wsClient.sendMessage("Peter")
  wsClient.expectMessage("Hello Peter!")

  wsClient.sendMessage(BinaryMessage(ByteString("abcdef")))
  wsClient.expectNoMessage(100.millis)

  wsClient.sendMessage("John")
  wsClient.expectMessage("Hello John!")

  wsClient.sendCompletion()
  wsClient.expectCompletion()
}

Java

copysourceimport static org.apache.pekko.http.javadsl.server.Directives.path;
import static org.apache.pekko.http.javadsl.server.Directives.handleWebSocketMessages;

final Flow<Message, Message, NotUsed> greeter =
    Flow.of(Message.class)
        .mapConcat(
            msg -> {
              if (msg instanceof TextMessage tm) {
                final TextMessage ret =
                    TextMessage.create(
                        Source.single("Hello ")
                            .concat(tm.getStreamedText())
                            .concat(Source.single("!")));
                return List.of(ret);
              } else if (msg instanceof BinaryMessage bm) {
                bm.getStreamedData().runWith(Sink.ignore(), materializer());
                return Collections.emptyList();
              } else {
                throw new IllegalArgumentException("Unsupported message type!");
              }
            });

final Route websocketRoute = path("greeter", () -> handleWebSocketMessages(greeter));

// create a testing probe representing the client-side
final WSProbe wsClient = WSProbe.create(system(), materializer());

// WS creates a WebSocket request for testing
testRoute(websocketRoute)
    .run(WS(Uri.create("/greeter"), wsClient.flow(), materializer()))
    .assertStatusCode(StatusCodes.SWITCHING_PROTOCOLS);

// manually run a WS conversation
wsClient.sendMessage("Peter");
wsClient.expectMessage("Hello Peter!");

wsClient.sendMessage(BinaryMessage.create(ByteString.fromString("abcdef")));
wsClient.expectNoMessage(Duration.ofMillis(100));

wsClient.sendMessage("John");
wsClient.expectMessage("Hello John!");

wsClient.sendCompletion();
wsClient.expectCompletion();