How to listen websocket server close events using akka-http websocket client

Question

I have websocket client connected to akka-http websocket server , how can i listen to connection close events happened on server( i.e server shutdown/ server closed websocket connection) ?

object Client extends App {


  implicit val actorSystem = ActorSystem("akka-system")
  implicit val flowMaterializer = ActorMaterializer()

  val config = actorSystem.settings.config
  val interface = config.getString("app.interface")

  val port = config.getInt("app.port")


  // print each incoming strict text message
  val printSink: Sink[Message, Future[Done]] =
    Sink.foreach {
      case message: TextMessage.Strict =>
        println(message.text)

      case _ => {
        sourceQueue.map(q => {
          println(s"offering message on client")
          q.offer(TextMessage("received unknown"))
        })
        println(s"received unknown message format")
      }
    }

  val (source, sourceQueue) = {
    val p = Promise[SourceQueue[Message]]
    val s = Source.queue[Message](Int.MaxValue, OverflowStrategy.backpressure).mapMaterializedValue(m => {
      p.trySuccess(m)
      m
    })
      .keepAlive(FiniteDuration(1, TimeUnit.SECONDS), () => TextMessage.Strict("Heart Beat"))
    (s, p.future)
  }

  val flow =
    Flow.fromSinkAndSourceMat(printSink, source)(Keep.right)


  val (upgradeResponse, sourceClose) =
    Http().singleWebSocketRequest(WebSocketRequest("ws://localhost:8080/ws-echo"), flow)

  val connected = upgradeResponse.map { upgrade =>
    // just like a regular http request we can get 404 NotFound,
    // with a response body, that will be available from upgrade.response
    if (upgrade.response.status == StatusCodes.SwitchingProtocols || upgrade.response.status == StatusCodes.SwitchingProtocols) {
      Done
    } else {
      throw new RuntimeException(s"Connection failed: ${upgrade.response.status}")
    }
  }


  connected.onComplete(println)

}

score 4 · Accepted Answer · answered Jun 09 '16 at 16:25

4

Websocket connection termination is modeled as a regular stream completion, therefore in your case you can use the materialized Future[Done] yielded by Sink.foreach:

val flow = Flow.fromSinkAndSourceMat(printSink, source)(Keep.both)

val (upgradeResponse, (sinkClose, sourceClose)) =
  Http().singleWebSocketRequest(..., flow)

sinkClose.onComplete {
  case Success(_) => println("Connection closed gracefully")
  case Failure(e) => println("Connection closed with an error: $e")
}

answered Jun 09 '16 at 16:25

Vladimir Matveev

120,085
34
287
296

That depends on your architecture. The simplest way would be, I think, to extract both `singleWebSocketRequest()` invocation and `sinkClose.onComplete` handler to a method and call this method recursively from the `onComplete` handler. – Vladimir Matveev Jun 10 '16 at 07:49
1

ok ty , i wish they support this out of the box like socket.io http://stackoverflow.com/questions/13797262/how-to-reconnect-to-websocket-after-close-connection – invariant Jun 10 '16 at 08:52
Your answer is precise and accurate. Thank you! – Brandon Bradley Feb 10 '17 at 16:36

How to listen websocket server close events using akka-http websocket client

1 Answers1