Hello,
I'm building a prototype for application that aims to use akka-http and
websockets for communication. It mostly works, but on close we get
exception:
[ERROR] [10/17/2016 19:38:56.104]
[my-system-akka.actor.default-dispatcher-10]
[akka.actor.ActorSystemImpl(my-system)] WebSocket handler failed with
Processor actor
[Actor[akka://my-system/user/StreamSupervisor-0/flow-4-0-unknown-operation#1709634420]]
terminated abruptly (akka.stream.AbruptTerminationException)
We tried many suggestions over the web, including calling
Http().shutdownAllConnectionPools() (which ends with success) and unbind
(which ends with success as well). Unfortunately it doesn't help - after
calling shutdownAllConnectionPools and unbind the websocket is still up (we
can send messages and receive answers from it). How to correctly close the
system to avoid such issue?
This is how (fragments) of how we build the binding (for now this is single
client app, so we have single actor to process messages):
val processing = system.actorOf(Props[ProcessingActor], "processing-1")
val route = get {
pathEndOrSingleSlash {
complete {
// ...
}
} ~ encodeResponse {
getFromResourceDirectory("")
} ~ path("ws") {
handleWebSocketMessages(Channel(processing))
}
}
val bindingFuture = Http().bindAndHandle(route, "localhost", port)
and then we wait (readLine) after which we close (as I said we have
shutdownAllConnectionPools, unbind and terminate). Page contains html and
Channel is custom flow.
object Channel {
private val bufferSize = 5
def apply(processing: ActorRef)(implicit system: ActorSystem):
Flow[WSMessage, WSMessage, _] = Flow.fromGraph(GraphDSL.create(
Source.actorRef[Message](bufferSize=bufferSize, OverflowStrategy.fail)
) { implicit builder => source =>
val sink = Sink.actorRef[SystemCommand](processing, Close)
val actor = builder.materializedValue.map(a => Initialize(a))
val fromSocket = builder.add(Flow[WSMessage].collect {
// ...
})
val toSocket = builder.add(Flow[Message].map {
// ...
})
val merge = builder.add(Merge[SystemCommand](2))
fromSocket ~> merge.in(0)
actor ~> merge.in(1)
merge ~> sink
source ~> toSocket
FlowShape(fromSocket.in, toSocket.out)
})
}
Any help would be appreciated - and btw, this was one of first attempts and
we got something up and running in just few hours, so even that we have
issues with close, we would like to emphatize that we think Akka is great!
Andrzej.
--
>>>>>>>>>> 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 [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
