Hi, just a quick suggestion: mabe a KillSwitch [1] in Channel flow would help?
cheers, Rafał [1] http://doc.akka.io/api/akka/2.4/?_ga=1.230565852.1388425308.1467760779#akka.stream.KillSwitch W dniu poniedziałek, 17 października 2016 21:52:55 UTC+2 użytkownik Andrzej Giniewicz napisał: > > 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 akka-user+unsubscr...@googlegroups.com. To post to this group, send email to akka-user@googlegroups.com. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
