How to correctly close system when websocket is open?

[Andrzej Giniewicz]

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.

[Rafał Krzewski]

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

[Andrzej Giniewicz]

Hi,

thanks, putting Channel(processing).via(killSwitch.flow) and shutting
it down later worked!

Regards,
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+...@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.