Scala & Play Websockets: Stockage des messages échangés

Question

J'ai commencé à jouer autour de scala et je suis arrivé à cette page d'accueil en particulier de la chatroom de socket web en scala.

Ils utilisent MessageHub.source() et BroadcastHub.sink() comme Source et Sink pour envoyer les messages à tous les clients connectés.

L'exemple fonctionne correctement pour échanger des messages tels quels. Je souhaite stocker les messages échangés dans le salon de discussion dans un DB.

J'ai essayé d'ajouter des fonctions map et fold à la source et à la mémoire pour récupérer les messages envoyés mais je n'ai pas pu.

J'ai essayé d'ajouter une étape d'écoulement entre MergeHub et BroadcastHub comme ci-dessous

val flow = Flow[WSMessage].map(element => println(s"Message: $element")) 
source.via(flow).toMat(sink)(Keep.both).run() 

Mais il génère une erreur de compilation qui ne peut pas faire référence à toMat avec une telle signature. Est-ce que quelqu'un peut m'aider ou me signaler comment puis-je obtenir des messages qui sont envoyés et les stocker dans la base de données.?

Lien pour le modèle complet:

https://github.com/playframework/play-scala-chatroom-example

Answer 1

Regardons votre flux:

val flow = Flow[WSMessage].map(element => println(s"Message: $element")) 

Il prend des éléments de type WSMessage, et ne retourne rien (Unit). Ici, il est à nouveau avec le bon type:

val flow: Flow[Unit] = Flow[WSMessage].map(element => println(s"Message: $element")) 

Cela ne fonctionnera pas clairement que l'évier attend WSMessage et non Unit.

Voici comment vous pouvez résoudre le problème ci-dessus:

val flow = Flow[WSMessage].map { element => 
    println(s"Message: $element") 
    element 
} 

Pas que pour les messages persistants dans la base de données, vous souhaiterez probablement utiliser une étape async, à peu près:

val flow = Flow[WSMessage].mapAsync(parallelism) { element => 
    println(s"Message: $element") 
    // assuming DB.write() returns a Future[Unit] 
    DB.write(element).map(_ => element) 
}