akka-http web socket issue

[Eric Swenson]

I have a web service, implemented in akka-http, to which I’ve added web socket support.  I have an akka-http client that is able to connect to the service and receive a single message, thereupon the web socket is, for some reason, disconnected.  I do not understand why it becomes disconnected. There is no intentional code on the server side that should disconnect the web socket.  On the client, I’m using virtually identical to the code found here:

http://doc.akka.io/docs/akka/2.4.7/scala/http/client-side/websocket-support.html#websocketclientflow

I’m finding that this code:

1. val connected = upgradeResponse.flatMap { upgrade =>
2. if (upgrade.response.status == StatusCodes.OK) {
3. Future.successful(Done)
4. } else {
5. throw new RuntimeException(s"Connection failed: ${upgrade.response.status}")
6. }
7. }

upon connection, raises the above RuntimeException. The value of upgrade.response.status is "101 Switching Protocols”. I do receive one message from server-to-client and displayed by the client before the “closed” Future is completed:

val (upgradeResponse, closed) =
  outgoing
    .viaMat(webSocketFlow)(Keep.right) // keep the materialized Future[WebSocketUpgradeResponse]
    .toMat(incoming)(Keep.both) // also keep the Future[Done]
    .run()

When the closed Future completes, this code is executed:

closed.foreach(_ => println("closed"))

and I see this “closed” message displayed.  However, before it is displayed, I do get a message sent by the service to the client.

My questions:  1) isn’t the “101 Switching Protocols” status expected?  Why is that not handled in the if clause of the upgrade response handling logic?  I did change the conditional to:

if (upgrade.response.status == StatusCodes.OK || upgrade.response.status == StatusCodes.SwitchingProtocols)

But while that prevents the runtime exception, the connection still automatically closes after receiving the first message.

On the server side, I have logic that looks like this:

class WebSocketConnection(experimentInstanceId: String, httpActorSystem: ActorSystem, clusterActorSystem: ActorSystem) {
  
  private[this] val experimentInstanceRegion = ClusterSharding(clusterActorSystem).shardRegion(ExperimentInstance.shardName)

  def websocketFlow: Flow[Message, Message, _] =
    Flow.fromGraph(
      GraphDSL.create(Source.actorRef[WsMessage](bufferSize = 5, OverflowStrategy.fail)) { implicit builder =>
        wsMessageSource => //source provided as argument

          // flow used as input from web socket
          val fromWebsocket = builder.add(
            Flow[Message].collect {
              case TextMessage.Strict(txt) =>
                println(s"txt=$txt")
                WsIncomingMessage(experimentInstanceId, txt)
            })

          // flow used as output, it returns Messages
          val backToWebsocket = builder.add(
            Flow[WsMessage].map {
              case WsMessage(author, text) => TextMessage(s"[$author]: $text")
            }
          )

          // send messages to the actor, if send also Disconnected(experimentInstanceId) before stream completes.
          val actorSink = Sink.actorRef[WebSocketEvent](experimentInstanceRegion, WsDisconnected(experimentInstanceId))

          // merges both pipes
          val merge = builder.add(Merge[WebSocketEvent](2))

          val actorAsSource = builder.materializedValue.map(actor => WsConnected(experimentInstanceId, actor))

          fromWebsocket ~> merge.in(0)

          actorAsSource ~> merge.in(1)

          merge ~> actorSink

          wsMessageSource ~> backToWebsocket

          // expose ports

          FlowShape(fromWebsocket.in, backToWebsocket.outlet)
      })

  def sendMessage(message: WsMessage): Unit = experimentInstanceRegion ! message
}

In the actor associated with the experimentInstanceRegion (e.g. a cluster sharing region), I see log messages indicating a WsConnected message is received.  And then shortly after, I see that the WsDisconnected message is received.  

I haven’t touched the web socket support logic in quite some time.  However, this all used to work — in other words, the client would stay active and display messages received from the server, and the server would continue sending messages to the client over the web socket connection until the client disconnected (on purpose).  This was all working in akka 2.3 and may have broken when I upgraded to the various 2.4 releases and is still broken using 2.4.7.  

Any suggestions on how to debug this?  Any reason for the auto-disconnection?  And what about the 101 Switching Protocols handling (mentioned above)?

Thanks. — Eric

[Eric Swenson]

I wrote a simple python client:

$ cat ws-test.py

import os

from websocket import create_connection

eiid=os.getenv("EIID")

token=os.getenv("BEARER_TOKEN")

header="Authorization: Bearer %s" % token

ws = create_connection("ws://localhost:9000/ws-connect/%s" % eiid, header=[header])

while True:

    x = ws.recv()

    print(x)

And used it with my server.  It does not disconnect and correctly responds to all messages sent by the server.  So that rules out problems with the akka-http-based server.  it would appear that the issue is with the akka-http client (whose code I got from this web page:

http://doc.akka.io/docs/akka/2.4.7/scala/http/client-side/websocket-support.html#websocketclientflow

Is there some reason why this code does not work?  As mentioned previously, the code connects, retrieves and displays one message sent by the server just after connection, and then promptly disconnects the websocket connection.

Are there known issues with client-side web socket support in 2.4.7?

— Eric

[Giovanni Alberto Caporaletti]

Are you keeping the outbound client flow open? 

http://doc.akka.io/docs/akka/2.4.7/scala/http/client-side/websocket-support.html#Half-Closed_WebSockets

[Eric Swenson]

Thanks, Giovanni. That was precisely the problem and while I didn't see the article you quoted, I "fixed" the problem on my end by having my client use a Source.fromFuture of a promise which I didn't complete.  I'll switch to using the approach in the article you referenced.  Still, I think there will be lots of people who may use the web socket client in the documentation as a starting point. They will expect that the client will continue to display messages received from the service while the client is running.  I think the example should either a) document that the client closes after processing a single message and how to make it maintain the web socket connection open, or better, in my opinion, b) use Source.maybe to ensure the connection doesn't close.

On a related note, if the web socket server side is implemented using akka-http, the example still closes the connection after 1 minute due to the server side's not having gotten any TCP packets over the connection. The documentation should either a) include a description of how to change this timeout or b) have the client include heartbeat logic to keep the connection alive.  

-- Eric

[Akka Team]

I have created a ticket for updating the docs about the client example mentioned: 

https://github.com/akka/akka/issues/20938

As always, contributions are welcome!

--

Johan

--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[allo...@gmail.com]

does this work when receiving large json data..? i have problem receiving large json data because sometimes it returns TextMessage.Streamed instead of TextMessage.Strict and another issue is when connecting to wss://s2.ripple.com:443 i can't connect using this url but can connect to other websocket servers..

[Konrad Malawski]

Yes it works. What's wrong about Streamed?

That's a feature that it's streaming the data – consume it using the dataBytes contained in the Streamed message.

-- 
Konrad `ktoso` Malawski
Akka @ Lightbend

[vauldex....@gmail.com]

I will re-open this issue because I am having problems regarding `s2.ripple.com`'s reply.


I was able to connect using either `webSocketClientFlow` and `singleWebSocketRequest`.

Having code block as my outgoing message:

```
val outgoing: Source[TextMessage.Streamed, NotUsed] =

      Source.single(TextMessage.Streamed(Source(List("""{"command": "subscribe", "stream": "transactions"}"""))))
```

I was expecting a `TextMessage.Streamed` reply but it replied a `TextMessage.Strict` message, naturally this command `{"command": "subscribe", "stream": "transactions"}` when sent to `s2.ripple.com` would return a stream of data, but it only replies a single message after message is sent.

Thanks in advance and God bless.