Server WebSocket Support

WebSocket is a protocol that provides a bi-directional channel between browser and webserver usually run over an upgraded HTTP(S) connection. Data is exchanged in messages whereby a message can either be binary data or Unicode text.

Akka HTTP provides a stream-based implementation of the WebSocket protocol that hides the low-level details of the underlying binary framing wire-protocol and provides a simple API to implement services using WebSocket.

Model

The basic unit of data exchange in the WebSocket protocol is a message. A message can either be binary message, i.e. a sequence of octets or a text message, i.e. a sequence of Unicode code points.

In the data model the two kinds of messages, binary and text messages, are represented by the two classes BinaryMessageBinaryMessage and TextMessageTextMessage deriving from a common superclass MessageMessage. The subclasses BinaryMessageBinaryMessage and TextMessageTextMessage contain methods to access the data. The superclass MessageMessage contains isText and isBinary methods to distinguish a message and asBinaryMessage and asTextMessage methods to cast a message. Take the API of TextMessageTextMessage as an example (BinaryMessageBinaryMessage is very similar with String replaced by ByteStringByteString):

Scala

/**
 * The ADT for WebSocket messages. A message can either be a binary or a text message.
 */
sealed trait Message extends akka.http.javadsl.model.ws.Message

/**
 * Represents a WebSocket text message. A text message can either be a [[TextMessage.Strict]] in which case
 * the complete data is already available or it can be [[TextMessage.Streamed]] in which case `textStream`
 * will return a Source streaming the data as it comes in.
 */
sealed trait TextMessage extends akka.http.javadsl.model.ws.TextMessage with Message {
  /**
   * The contents of this message as a stream.
   */
  def textStream: Source[String, _]

  /** Java API */
  override def getStreamedText: javadsl.Source[String, _] = textStream.asJava
  override def asScala: TextMessage = this
}
sealed trait BinaryMessage extends akka.http.javadsl.model.ws.BinaryMessage with Message {
  /**
   * The contents of this message as a stream.
   */
  def dataStream: Source[ByteString, _]

  /** Java API */
  override def getStreamedData: javadsl.Source[ByteString, _] = dataStream.asJava
  override def asScala: BinaryMessage = this
}

Java

abstract class TextMessage extends Message {
  /**
   * Returns a source of the text message data.
   */
  def getStreamedText: Source[String, _]

  /** Is this message a strict one? */
  def isStrict: Boolean

  /**
   * Returns the strict message text if this message is strict, throws otherwise.
   */
  def getStrictText: String
}

The data of a message is provided as a stream because WebSocket messages do not have a predefined size and could (in theory) be infinitely long. However, only one message can be open per direction of the WebSocket connection, so that many application level protocols will want to make use of the delineation into (small) messages to transport single application-level data units like “one event” or “one chat message”.

Many messages are small enough to be sent or received in one go. As an opportunity for optimization, the model provides the notion of a “strict” message to represent cases where a whole message was received in one go. Strict messages are represented with the Strict subclass for each kind of message which contains data as a strict, i.e. non-streamed, ByteStringByteString or String. If TextMessage.isStrict returns true, the complete data is already available and can be accessed with TextMessage.getStrictText (analogously for BinaryMessageBinaryMessage).

When receiving data from the network connection the WebSocket implementation tries to create a strict message whenever possible, i.e. when the complete data was received in one chunk. However, the actual chunking of messages over a network connection and through the various streaming abstraction layers is not deterministic from the perspective of the application. Therefore, application code must be able to handle both streamed and strict messages and not expect certain messages to be strict. (Particularly, note that tests against localhost will behave differently than tests against remote peers where data is received over a physical network connection.)

For sending data, you can use TextMessage.apply(text: String)TextMessage.create(String) to create a strict message if the complete message has already been assembled. Otherwise, use TextMessage.apply(textStream: Source[String, \_])TextMessage.create(Source<String, ?>) to create a streaming message from an Akka Stream source.

Server API

The entrypoint for the WebSocket API is the synthetic UpgradeToWebSocketUpgradeToWebSocket header which is added to a request if Akka HTTP encounters a WebSocket upgrade request.

The WebSocket specification mandates that details of the WebSocket connection are negotiated by placing special-purpose HTTP-headers into request and response of the HTTP upgrade. In Akka HTTP these HTTP-level details of the WebSocket handshake are hidden from the application and don’t need to be managed manually.

Instead, the synthetic UpgradeToWebSocketUpgradeToWebSocket represents a valid WebSocket upgrade request. An application can detect a WebSocket upgrade request by looking for the UpgradeToWebSocketUpgradeToWebSocket header. It can choose to accept the upgrade and start a WebSocket connection by responding to that request with an HttpResponseHttpResponse generated by one of the UpgradeToWebSocket.handleMessagesWith methods. In its most general form this method expects two arguments: first, a handler Flow<Message, Message, Any>Flow[Message, Message, Any]Flow<Message, Message, ?>Flow[Message, Message, ?] that will be used to handle WebSocket messages on this connection. Second, the application can optionally choose one of the proposed application-level sub-protocols by inspecting the values of UpgradeToWebSocket.requestedProtocolsUpgradeToWebSocket.getRequestedProtocols and pass the chosen protocol value to handleMessageshandleMessagesWith.

Handling Messages

A message handler is expected to be implemented as a Flow<Message, Message, Any>Flow[Message, Message, Any]Flow<Message, Message, ?>Flow[Message, Message, ?]. For typical request-response scenarios this fits very well and such a FlowFlow can be constructed from a simple function by using Flow[Message].map or Flow[Message].mapAsyncFlow.<Message>create().map or Flow.<Message>create().mapAsync.

There are other use-cases, e.g. in a server-push model, where a server message is sent spontaneously, or in a true bi-directional scenario where input and output aren’t logically connected. Providing the handler as a FlowFlow in these cases may not fit. Another method named UpgradeToWebSocket.handleMessagesWithSinkSourceAn overload of UpgradeToWebSocket.handleMessagesWith is provided, instead, which allows to pass an output-generating Source<Message, ?>Source[Message, _] and an input-receiving Sink<Message, ?>Sink[Message, _] independently.

Note that a handler is required to consume the data stream of each message to make place for new messages. Otherwise, subsequent messages may be stuck and message traffic in this direction will stall.

Example

Let’s look at an exampleexample.

WebSocket requests come in like any other requests. In the example, requests to /greeter are expected to be WebSocket requests:

Scala

val requestHandler: HttpRequest => HttpResponse = {
  case req @ HttpRequest(GET, Uri.Path("/greeter"), _, _, _) =>
    req.header[UpgradeToWebSocket] match {
      case Some(upgrade) => upgrade.handleMessages(greeterWebSocketService)
      case None          => HttpResponse(400, entity = "Not a valid websocket request!")
    }
  case r: HttpRequest =>
    r.discardEntityBytes() // important to drain incoming HTTP Entity stream
    HttpResponse(404, entity = "Unknown resource!")
}

Java

public static HttpResponse handleRequest(HttpRequest request) {
  System.out.println("Handling request to " + request.getUri());

  if (request.getUri().path().equals("/greeter")) {
    final Flow<Message, Message, NotUsed> greeterFlow = greeter();
    return WebSocket.handleWebSocketRequestWith(request, greeterFlow);
  } else {
    return HttpResponse.create().withStatus(404);
  }
}

In the example, the passed handler expects text messages where each message is expected to contain a (person’s) name and then responds with another text message that contains a greeting:

Scala

// The Greeter WebSocket Service expects a "name" per message and
// returns a greeting message for that name
val greeterWebSocketService =
  Flow[Message]
    .mapConcat {
      // we match but don't actually consume the text message here,
      // rather we simply stream it back as the tail of the response
      // this means we might start sending the response even before the
      // end of the incoming message has been received
      case tm: TextMessage => TextMessage(Source.single("Hello ") ++ tm.textStream) :: Nil
      case bm: BinaryMessage =>
        // ignore binary messages but drain content to avoid the stream being clogged
        bm.dataStream.runWith(Sink.ignore)
        Nil
    }

Java

/**
 * A handler that treats incoming messages as a name,
 * and responds with a greeting to that name
 */
public static Flow<Message, Message, NotUsed> greeter() {
  return
    Flow.<Message>create()
      .collect(new JavaPartialFunction<Message, Message>() {
        @Override
        public Message apply(Message msg, boolean isCheck) throws Exception {
          if (isCheck) {
            if (msg.isText()) {
              return null;
            } else {
              throw noMatch();
            }
          } else {
            return handleTextMessage(msg.asTextMessage());
          }
        }
      });
}

public static TextMessage handleTextMessage(TextMessage msg) {
  if (msg.isStrict()) // optimization that directly creates a simple response...
  {
    return TextMessage.create("Hello " + msg.getStrictText());
  } else // ... this would suffice to handle all text messages in a streaming fashion
  {
    return TextMessage.create(Source.single("Hello ").concat(msg.getStreamedText()));
  }
}

Routing support

The routing DSL provides the handleWebSocketMessages directive to install a WebSocket handler if a request is a WebSocket request. Otherwise, the directive rejects the request.

Let’s look at how the above example can be rewritten using the high-level routing DSL.

Instead of writing the request handler manually, the routing behavior of the app is defined by a route that uses the handleWebSocketRequests directive in place of the WebSocket.handleWebSocketRequestWith:

Scala

def 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

public Route createRoute() {
  return
    path("greeter", () ->
      handleWebSocketMessages(greeter())
    );
}

The handling code itself will be the same as with using the low-level API.

See the full routing example.