如何使用akka-http websocket客户端监听websocket服务器关闭事件

Question

我有websocket客户端连接到akka-http websocket服务器，如何监听服务器上发生的连接关闭事件（即服务器关闭/服务器关闭websocket连接）？

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) 

} 

Answer 1

WebSocket连接终止建模为一个常规流完成，因此你的情况，你可以使用物化Future[Done]产生由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") 
}