Akka-http websockets connection fails 95% of the time

[Luc Klaassen]

Hello all,

I'm having trouble setting up a basic websocket connection to an akka-http websocket server. Almost all connections seem to fail, and are timed out by akka-http after 5 seconds. Sometimes (seems completely random) the connection succeeds though and the message sending works flawlessly. This is really confusing me, since the problem is not 100% reproducable and seems completely random.

See stackoverflow for my current code:

http://stackoverflow.com/questions/31579385/connection-between-js-and-akka-http-websockets-fails-95-of-the-time

Any suggestions are greatly appreciated.

[Johannes Rudolph]

Hi Luc,

I'd like to get to the bottom of this problem to make sure we find the problem if there's one in akka-http. Could you provide an executable reproduction?

Johannes

[Luc Klaassen]

I've pushed the repository to github at https://github.com/Luckl/AkkaHttpWebsockets

Luc

Op vrijdag 24 juli 2015 10:07:53 UTC+2 schreef Johannes Rudolph:

[Johannes Rudolph]

Thanks Luc. I'll have a look next week.

[Johannes Rudolph]

Hi Luc,

I can see the log messages you are seeing:

[DEBUG] [07/27/2015 12:01:01.059]
[api-akka.actor.default-dispatcher-9]
[akka://api/user/$a/flow-47-3-publisherSource-prefixAndTail]
Cancelling akka.stream.impl.MultiStreamOutputProcessor$SubstreamOutput@eda0238
(after: 5000 ms)

but I don't see any connections failing. In which way do you observe
connections to fail?


--
Johannes

-----------------------------------------------
Johannes Rudolph
http://virtual-void.net

[Endre Varga]

On Mon, Jul 27, 2015 at 12:22 PM, 'Johannes Rudolph' via Akka User List <akka...@googlegroups.com> wrote:

Hi Luc,

I can see the log messages you are seeing:

[DEBUG] [07/27/2015 12:01:01.059]
[api-akka.actor.default-dispatcher-9]
[akka://api/user/$a/flow-47-3-publisherSource-prefixAndTail]
Cancelling akka.stream.impl.MultiStreamOutputProcessor$SubstreamOutput@eda0238
(after: 5000 ms)

This looks like an unsubscribed substream. Are the entity bodies properly consumed?

-Endre

 

but I don't see any connections failing. In which way do you observe
connections to fail?


--
Johannes

-----------------------------------------------
Johannes Rudolph
http://virtual-void.net

--
>>>>>>>>>>      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 http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[Luc Klaassen]

The connection does not give an error on the client side, but each time a message is sent it should print the message server side and then return a response with the reversed string. None of this is happening most of the time. It simply says the websocket connection is open, and none of the messages that are sent are received server-side.

 

I've tried the example you have on github, the chat application at https://github.com/jrudolph/akka-http-scala-js-websocket-chat and this works without problems. (tried this to rule out misconfiguration on my pc, there's some old settings from previous projects)

I'll try to run it inside a virtual tonight to make sure this is definitely not the issue here.

Op maandag 27 juli 2015 12:22:33 UTC+2 schreef Johannes Rudolph:

[Johannes Rudolph]

Hi Luc,

On Mon, Jul 27, 2015 at 4:49 PM, Luc Klaassen <luc.kl...@gmail.com> wrote:
> The connection does not give an error on the client side, but each time a
> message is sent it should print the message server side and then return a
> response with the reversed string. None of this is happening most of the
> time. It simply says the websocket connection is open, and none of the
> messages that are sent are received server-side.

Thanks for confirming. That's not what I observe, for me it seems to work.

However (also, as hinted on the SO issue) your use of `collect` is a
risk as it won't work in all cases. A message is not guaranteed to be
strict, so you may miss streamed messages (see [1]). Try

def websocketActorFlow: Flow[Message, Message, Unit] =
Flow[Message].mapConcat{
case TextMessage.Strict(msg) =>
println(msg)
TextMessage.Strict(msg.reverse) :: Nil

case other: TextMessage =>
println(s"Got other text $other")
other.textStream.runWith(Sink.ignore)
Nil

case other: BinaryMessage =>
println(s"Got other binary $other")
other.dataStream.runWith(Sink.ignore)

Nil
}

and see if you get streamed messages some times. If you get a
non-strict `TextMessage` you either need to handle it in a streaming
fashion or you need to buffer the complete message and only run your
processing, then, on the aggregated string. The second alternative has
the usual risk of using unbounded memory.

That said, I guess that we could need some convenience helpers that
would help with aggregating messages allowing you to specify the
maximum number of characters and maximum duration for which the
aggregation should run per message.

Johannes

[1] http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/http/routing-dsl/websocket-support.html#Model

[Luc Klaassen]

I've tried your suggestion, but i'm afraid it doesn't change anything. When the connection succeeds it will log the message and return, otherwise nothing is logged server side, so no other TextMessage or BinaryMessages received.

Luc

Op maandag 27 juli 2015 17:03:50 UTC+2 schreef Johannes Rudolph:

[Johannes Rudolph]

Hi Luc,

in that case, can you provide a more information about the kind of
system you are running on? And it would be nice if we had a
tcpdump/wireshark dump that shows what's going on, so we can see at
least at which side things start to go wrong?

Johannes

[Luc Klaassen]

Hello,

If you check the git repository, i made some dumps with rawcap (to be able to capture localhost tcp). The messages are simply not being sent for some reason. When i tried sending messages with an interval from client side, it shows that the connection eventually gets set up properly and it handles all sent messages immediately. Therefore i think it's safe to assume it's an issue with my local setup. I also asked a friend of mine to test it and for him it also works without issues.

Op maandag 27 juli 2015 17:27:59 UTC+2 schreef Johannes Rudolph:

[Johannes Rudolph]

Hi Luc,

On Mon, Jul 27, 2015 at 8:23 PM, Luc Klaassen <luc.kl...@gmail.com> wrote:
> If you check the git repository, i made some dumps with rawcap (to be able
> to capture localhost tcp).

Thanks. I wonder if you are running servers on different ports? Why
are there servers running on ports 8080, 55175, and 49162 in the
dumps?

> The messages are simply not being sent for some
> reason.

I only see this in the connection to 8080 (dumpfile.pcap). In the
other connection to the server on port 49162 (dumpfile.pcap) several
frames are sent by the client but never answered. In the connection to
port 55175 (dumpfile-ping.pcap) I also see all of the data to be sent
ny the client immediately but the responses are delayed and only
dispatched much later (starting with second 23) in one go.
dumpfile-ping.pcap is unfortunately missing the connection
establishment and mixes in some other connection as well so it's
really hard to see what's really going on.

[Luc Klaassen]

I've added a new pcap file, again while sending pings each second, and this time i also captured the initial connection. Again it sends all initial requests at once after a few seconds (this time 6 seconds)

I honestly have no idea what the other connections are that can be seen. I'm guessing it's some old leftover stuff from previous projects, or something running in the background. I guess this is also what's causing the problems since noone else seems to have them. (this still doesnt explain why the chat application works though, or why without sending pings the message never gets sent)

Op dinsdag 28 juli 2015 08:16:09 UTC+2 schreef Johannes Rudolph:

[Johannes Rudolph]

Hi Luc,

in the new dump there are again these two connections:

port 59385 -> 57649

and

port 59386 -> 8080

It looks as if you are running some kind of transparent proxy (a
desktop firewall sofware?) that redirects all the traffic through a
middleman. So, if you look at the connection from 59385 (the browser?)
it indeed directly sends the packets. However, on the second
connection all these data is buffered for 6 seconds before it is
relayed to the actual server at 8080.

Johannes

[Corey Auger]

@Johannes,

I have run into this problem you are describing ...  I am wondering if there was any work done on "helpers" or what the best strategy to use for aggregating messages.

Would a FoldSink do the trick?  Or am I off track.  Appreciate any guidance you have .. thanks !

[Endre Varga]

Hi Corey,

For aggregating non-strict entities (of any kind) there are the following tools:

 - completionTimeout - to put an upper time limit on an aggregation

 - limit (coming in this PR: https://github.com/akka/akka/pull/18965) - to put an upper limit on size

 - fold - to aggregate the result

src.limit(1024).completionTimeout.runFold(...)(...)

-Endre

--

[ac2ep...@gmail.com]

Reporting: server side drops connection with timeout in 5 sec (on same host):

Code: 

      case TextMessage.Streamed(stream)  => {

        val future = stream.limit(100).runFold("")(_ + _)

        val txt = Await.result(future, 5 seconds).asInstanceOf[String]

        process(txt)

      }   

Question: 

Does anybody in the world know in practice, how to collect this Source in reasonably convenient way?

[Akka Team]

Hi,

Don't use Await inside of a stream as it will block the entire stream from processing elements, instead use for example mapAsync like this to collect the complete text messages into strings (note that these will be entirely in memory so you might want to add some limits in there):

val messageToStringFlow: Flow[Message, String, NotUsed] = 

  Flow[Message].mapAsync(1) {

    case TextMessage.Strict(text) =>       

      Future.successful(text)

    case streamed: TextMessage.Streamed =>

      streamed.textStream.runFold("")(_ ++ _)

  }

I hope this helps

--

Johan

Akka Team

--
>>>>>>>>>> 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+unsubscribe@googlegroups.com.

To post to this group, send email to akka...@googlegroups.com.

Visit this group at https://groups.google.com/group/akka-user.