Client-Side WebSocket Support

Client side WebSocket support is available through Http.singleWebSocketRequest , Http.webSocketClientFlow and Http.webSocketClientLayer.

A WebSocket consists of two streams of messages, incoming messages (a Sink) and outgoing messages (a Source) where either may be signalled first; or even be the only direction in which messages flow during the lifetime of the connection. Therefore a WebSocket connection is modelled as either something you connect a Flow<Message, Message, Mat> to or a Flow<Message, Message, Mat> that you connect a Source<Message, Mat> and a Sink<Message, Mat> to.

A WebSocket request starts with a regular HTTP request which contains an Upgrade header (and possibly other regular HTTP request properties), so in addition to the flow of messages there also is an initial response from the server, this is modelled with WebSocketUpgradeResponse.

The methods of the WebSocket client API handle the upgrade to WebSocket on connection success and materializes the connected WebSocket stream. If the connection fails, for example with a 404 NotFound error, this regular HTTP result can be found in WebSocketUpgradeResponse.response

注釈

Make sure to read and understand the section about Half-Closed WebSockets as the behavior when using WebSockets for one-way communication may not be what you would expect.

Message

Messages sent and received over a WebSocket can be either TextMessage s or BinaryMessage s and each of those can be either strict (all data in one chunk) or streamed. In typical applications messages will be strict as WebSockets are usually deployed to communicate using small messages not stream data, the protocol does however allow this (by not marking the first fragment as final, as described in rfc 6455 section 5.2).

The strict text is available from TextMessage.getStrictText and strict binary data from BinaryMessage.getStrictData.

For streamed messages BinaryMessage.getStreamedData and TextMessage.getStreamedText is used to access the data. In these cases the data is provided as a Source<ByteString, NotUsed> for binary and Source<String, NotUsed> for text messages.

singleWebSocketRequest

singleWebSocketRequest takes a WebSocketRequest and a flow it will connect to the source and sink of the WebSocket connection. It will trigger the request right away and returns a tuple containing a CompletionStage<WebSocketUpgradeResponse> and the materialized value from the flow passed to the method.

The future will succeed when the WebSocket connection has been established or the server returned a regular HTTP response, or fail if the connection fails with an exception.

Simple example sending a message and printing any incoming message:

ActorSystem system = ActorSystem.create();
Materializer materializer = ActorMaterializer.create(system);
Http http = Http.get(system);

// print each incoming text message
// would throw exception on non strict or binary message
final Sink<Message, CompletionStage<Done>> printSink =
  Sink.foreach((message) ->
    System.out.println("Got message: " + message.asTextMessage().getStrictText())
  );

// send this as a message over the WebSocket
final Source<Message, NotUsed> helloSource =
  Source.single(TextMessage.create("hello world"));

// the CompletionStage<Done> is the materialized value of Sink.foreach
// and it is completed when the stream completes
final Flow<Message, Message, CompletionStage<Done>> flow =
  Flow.fromSinkAndSourceMat(printSink, helloSource, Keep.left());

final Pair<CompletionStage<WebSocketUpgradeResponse>, CompletionStage<Done>> pair =
  http.singleWebSocketRequest(
    WebSocketRequest.create("ws://echo.websocket.org"),
    flow,
    materializer
  );

// The first value in the pair is a CompletionStage<WebSocketUpgradeResponse> that
// completes when the WebSocket request has connected successfully (or failed)
final CompletionStage<Done> connected = pair.first().thenApply(upgrade -> {
  // just like a regular http request we can access response status which is available via upgrade.response.status
  // status code 101 (Switching Protocols) indicates that server support WebSockets
  if (upgrade.response().status().equals(StatusCodes.SWITCHING_PROTOCOLS)) {
    return Done.getInstance();
  } else {
    throw new RuntimeException("Connection failed: " + upgrade.response().status());
  }
});

// the second value is the completion of the sink from above
// in other words, it completes when the WebSocket disconnects
final CompletionStage<Done> closed = pair.second();

// in a real application you would not side effect here
// and handle errors more carefully
connected.thenAccept(done -> System.out.println("Connected"));
closed.thenAccept(done -> System.out.println("Connection closed"));

The websocket request may also include additional headers, like in this example, HTTP Basic Auth:

http.singleWebSocketRequest(
  WebSocketRequest.create("ws://example.com:8080/some/path")
    .addHeader(Authorization.basic("johan", "correcthorsebatterystaple")),
  flow,
  materializer);

webSocketClientFlow

webSocketClientFlow takes a request, and returns a Flow<Message, Message, CompletionStage<WebSocketUpgradeResponse>>.

The future that is materialized from the flow will succeed when the WebSocket connection has been established or the server returned a regular HTTP response, or fail if the connection fails with an exception.

注釈

The Flow that is returned by this method can only be materialized once. For each request a new flow must be acquired by calling the method again.

Simple example sending a message and printing any incoming message:

ActorSystem system = ActorSystem.create();
Materializer materializer = ActorMaterializer.create(system);
Http http = Http.get(system);

// print each incoming text message
// would throw exception on non strict or binary message
Sink<Message, CompletionStage<Done>> printSink =
  Sink.foreach((message) ->
      System.out.println("Got message: " + message.asTextMessage().getStrictText())
  );

// send this as a message over the WebSocket
Source<Message, NotUsed> helloSource =
  Source.single(TextMessage.create("hello world"));


Flow<Message, Message, CompletionStage<WebSocketUpgradeResponse>> webSocketFlow =
  http.webSocketClientFlow(WebSocketRequest.create("ws://echo.websocket.org"));


Pair<CompletionStage<WebSocketUpgradeResponse>, CompletionStage<Done>> pair =
  helloSource.viaMat(webSocketFlow, Keep.right())
    .toMat(printSink, Keep.both())
    .run(materializer);


// The first value in the pair is a CompletionStage<WebSocketUpgradeResponse> that
// completes when the WebSocket request has connected successfully (or failed)
CompletionStage<WebSocketUpgradeResponse> upgradeCompletion = pair.first();

// the second value is the completion of the sink from above
// in other words, it completes when the WebSocket disconnects
CompletionStage<Done> closed = pair.second();

CompletionStage<Done> connected = upgradeCompletion.thenApply(upgrade->
{
  // just like a regular http request we can access response status which is available via upgrade.response.status
  // status code 101 (Switching Protocols) indicates that server support WebSockets
  if (upgrade.response().status().equals(StatusCodes.SWITCHING_PROTOCOLS)) {
    return Done.getInstance();
  } else {
    throw new RuntimeException(("Connection failed: " + upgrade.response().status()));
  }
});

// in a real application you would not side effect here
// and handle errors more carefully
connected.thenAccept(done -> System.out.println("Connected"));
closed.thenAccept(done -> System.out.println("Connection closed"));

webSocketClientLayer

Just like the Stand-Alone HTTP Layer Usage for regular HTTP requests, the WebSocket layer can be used fully detached from the underlying TCP interface. The same scenarios as described for regular HTTP requests apply here.

The returned layer forms a BidiFlow<Message, SslTlsOutbound, SslTlsInbound, Message, CompletionStage<WebSocketUpgradeResponse>>.

Half-Closed WebSockets

The Akka HTTP WebSocket API does not support half-closed connections which means that if the either stream completes the entire connection is closed (after a "Closing Handshake" has been exchanged or a timeout of 3 seconds has passed). This may lead to unexpected behavior, for example if we are trying to only consume messages coming from the server, like this:

// we may expect to be able to to just tail
// the server websocket output like this
final Flow<Message, Message, NotUsed> flow =
  Flow.fromSinkAndSource(
    Sink.foreach(System.out::println),
    Source.empty());

http.singleWebSocketRequest(
  WebSocketRequest.create("ws://example.com:8080/some/path"),
  flow,
  materializer);

This will in fact quickly close the connection because of the Source.empty being completed immediately when the stream is materialized. To solve this you can make sure to not complete the outgoing source by using for example Source.maybe like this:

// using Source.maybe materializes into a completable future
// which will allow us to complete the source later
final Flow<Message, Message, CompletableFuture<Optional<Message>>> flow =
  Flow.fromSinkAndSourceMat(
    Sink.foreach(System.out::println),
    Source.maybe(),
    Keep.right());

final Pair<CompletionStage<WebSocketUpgradeResponse>, CompletableFuture<Optional<Message>>> pair =
  http.singleWebSocketRequest(
    WebSocketRequest.create("ws://example.com:8080/some/path"),
    flow,
    materializer);

// at some later time we want to disconnect
pair.second().complete(Optional.empty());

This will keep the outgoing source from completing, but without emitting any elements until the CompletableFuture is manually completed which makes the Source complete and the connection to close.

The same problem holds true if emitting a finite number of elements, as soon as the last element is reached the Source will close and cause the connection to close. To avoid that you can concatenate Source.maybe to the finite stream:

// emit "one" and then "two" and then keep the source from completing
final Source<Message, CompletableFuture<Optional<Message>>> source =
  Source.from(Arrays.<Message>asList(TextMessage.create("one"), TextMessage.create("two")))
    .concatMat(Source.maybe(), Keep.right());

final Flow<Message, Message, CompletableFuture<Optional<Message>>> flow =
  Flow.fromSinkAndSourceMat(
    Sink.foreach(System.out::println),
    source,
    Keep.right());

final Pair<CompletionStage<WebSocketUpgradeResponse>, CompletableFuture<Optional<Message>>> pair =
  http.singleWebSocketRequest(
    WebSocketRequest.create("ws://example.com:8080/some/path"),
    flow,
    materializer);

// at some later time we want to disconnect
pair.second().complete(Optional.empty());

Scenarios that exist with the two streams in a WebSocket and possible ways to deal with it:

Scenario	Possible solution
Two-way communication	`Flow.fromSinkAndSource`, or `Flow.map` for a request-response protocol
Infinite incoming stream, no outgoing	`Flow.fromSinkAndSource(someSink, Source.maybe())`
Infinite outgoing stream, no incoming	`Flow.fromSinkAndSource(Sink.ignore(), yourSource)`

Contents

Client-Side WebSocket Support

Message

singleWebSocketRequest

webSocketClientFlow

webSocketClientLayer

Half-Closed WebSockets

Akka

Contribute

Company