How to stream messages through websockets in Akka HTTP?

Akka Akka HTTP
1

I’m trying to setup a relatively simple websockets system where clients connect, send a message, and the server processes it and periodically sends out messages to the client. Then, the server closes the connection.

I’m really stuck on the step for sending responses back to the client.

The documentation gives the following code for handling incoming messages, then just sending them back to the client:

val handler: Flow[Message, TextMessage, NotUsed] = Flow[Message].mapConcat {
    case msg: TextMessage =>
     //Incoming message, send it back as is
      TextMessage(Source.single(Hi:  ") ++ msg.textStream ) :: Nil
     //Snipped

But there is no documentation for how to send out messages without them being in response to existing messages from the client.

I think what I need is to return a source, which, rather than being Source.single(), is an indefinite source which doesn’t close. But I cannot figure out how to create one of those - Source() only takes an iterable which means it expects a finite set of values.

Any help would be appreciated.

1 Like
2

Flow.fromSinkAndSource or Flow.fromSinkAndSourceCoupled are your friends here, they allow an arbitrary/decoupled pair of Sink[Message, _] and Source[Message, _] to be combined into a flow you can handle the websocket connection with.

3

Hey @johanandren

Would it be possible for you to please provide an example of how to use Flow.fromSinkAndSource or Flow.fromSinkAndSourceCoupled to send messages back to the user for websockets?

4

Alternatively, if there was a way to use the ActorFlow from play ( https://www.playframework.com/documentation/2.7.x/ScalaWebSockets ) that should also work well - but that one only returns a Flow[In,Out] while websockets in Akka HTTP require Flow[In,Out,None] which makes them incompatible.

If there was any way to use an actor, and send msgs to that actor and have those pushed back onto the websocket connection, that’d work really well.

Otherwise I’m facing a rewrite of my app to Play…

5

Here’s a, somewhat dated, example of interacting with an actor from websockets, perhaps that can help figuring it out: https://markatta.com/codemonkey/posts/chat-with-akka-http-websockets-old/