Akka websocket - 如何通過服務器關閉連接？

Question

所以這裏是我的websocket服務器實現。

val route = get { 
    pathEndOrSingleSlash { 
    handleWebSocketMessages(websocketFlow) 
    } 
} 

def websocketFlow: Flow[Message, Message, Any] = 
    Flow[Message] 
    .collect { case TextMessage.Strict(textMessage) => protocol.hydrate(textMessage) } 
    .via(chatActorFlow(UUID.randomUUID())) 
    .map(event => TextMessage.Strict(protocol.serialize(event))) 


def chatActorFlow(connectionId: UUID) : Flow[Protocol.Message, Protocol.Event, Any] = { 

    val sink = Flow[Protocol.Message] 
    .map(msg => Protocol.SignedMessage(connectionId, msg)) 
    .to(Sink.actorRef(chatRef, Protocol.CloseConnection(connectionId))) 

    val source = Source 
     .mapMaterializedValue { 
     actor : ActorRef => { 
      chatRef ! Protocol.OpenConnection(actor, connectionId) 
     } 
     } 

    Flow.fromSinkAndSource(sink, source) 
} 

我不知道是否有什麼辦法一旦ConnectionClosed類型的消息是由chatRef發送關閉連接？

Answer 1

下面的解決方案允許通過終止由Source.actorRef階段實現的Actor來從服務器端刪除連接。這隻需發送PoisonPill即可完成。

現在，我還不清楚你想在連接時識別一個「被禁止」的客戶端，所以這個例子 - 故意 - 非常簡單：服務器在最大數量的客戶連接。如果您想要隨時使用任何其他策略來啓動客戶端，則仍然可以應用相同的邏輯並將PoisonPill發送給其自己的源演員。

object ChatApp extends App { 

    implicit val system = ActorSystem("chat") 
    implicit val executor: ExecutionContextExecutor = system.dispatcher 
    implicit val materializer = ActorMaterializer() 

    val route = get { 
    pathEndOrSingleSlash { 
     handleWebSocketMessages(websocketFlow) 
    } 
    } 

    val maximumClients = 1 

    class ChatRef extends Actor { 
    override def receive: Receive = withClients(Map.empty[UUID, ActorRef]) 

    def withClients(clients: Map[UUID, ActorRef]): Receive = { 
     case SignedMessage(uuid, msg) => clients.collect{ 
     case (id, ar) if id != uuid => ar ! msg 
     } 
     case OpenConnection(ar, uuid) if clients.size == maximumClients => ar ! PoisonPill 
     case OpenConnection(ar, uuid) => context.become(withClients(clients.updated(uuid, ar))) 
     case CloseConnection(uuid) => context.become(withClients(clients - uuid)) 
    } 
    } 

    object Protocol { 
    case class SignedMessage(uuid: UUID, msg: String) 
    case class OpenConnection(actor: ActorRef, uuid: UUID) 
    case class CloseConnection(uuid: UUID) 
    } 

    val chatRef = system.actorOf(Props[ChatRef]) 

    def websocketFlow: Flow[Message, Message, Any] = 
    Flow[Message] 
     .mapAsync(1) { 
     case TextMessage.Strict(s) => Future.successful(s) 
     case TextMessage.Streamed(s) => s.runFold("")(_ + _) 
     case b: BinaryMessage => throw new Exception("Binary message cannot be handled") 
     }.via(chatActorFlow(UUID.randomUUID())) 
     .map(TextMessage(_)) 

    def chatActorFlow(connectionId: UUID) : Flow[String, String, Any] = { 

    val sink = Flow[String] 
     .map(msg => Protocol.SignedMessage(connectionId, msg)) 
     .to(Sink.actorRef(chatRef, Protocol.CloseConnection(connectionId))) 

    val source = Source.actorRef(16, OverflowStrategy.fail) 
     .mapMaterializedValue { 
     actor : ActorRef => { 
      chatRef ! Protocol.OpenConnection(actor, connectionId) 
     } 
     } 

    Flow.fromSinkAndSource(sink, source) 
    } 

    Http().bindAndHandle(route, "0.0.0.0", 8080) 
    .map(_ => println(s"Started server...")) 

}