Init

src/routes/gateway/handlers.rs

@@ -42,38 +42,39 @@ pub async fn ws_handler(
 
 async fn handle_socket(socket: WebSocket, state: AppState, user: User) {
     let (mut sender, mut receiver) = socket.split();
-    let (tx, mut rx) = mpsc::unbounded_channel();
+    let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
     let event_bus = state.event_bus.clone();
 
-    let client = GatewayClient::new(user);
+    let client = GatewayClient::new(user, tx);
     client.on_connect().await;
-    state.gateway.add_user(client);
+    state.gateway.add_client(client.clone());
 
     // // Enregistrement du client (Connect)
     // on_connect(user_id, tx, &state).await;
     //
-    // // Task pour envoyer les messages du canal mpsc vers le WebSocket
-    // let mut send_task = tokio::spawn(async move {
-    //     while let Some(message) = rx.recv().await {
-    //         if sender.send(message).await.is_err() {
-    //             break;
-    //         }
-    //     }
-    // });
+    // Task pour envoyer les messages du canal mpsc vers le WebSocket
+    let mut send_task = tokio::spawn(async move {
+        while let Some(message) = rx.recv().await {
+            if sender.send(message).await.is_err() {
+                break;
+            }
+        }
+    });
     //
-    // // Task pour recevoir les messages du WebSocket
-    // let state_clone = state.clone();
-    // let mut recv_task = tokio::spawn(async move {
-    //     while let Some(Ok(message)) = receiver.next().await {
-    //         on_message(user_id, message, &state_clone).await;
-    //     }
-    // });
+    // Task pour recevoir les messages du WebSocket
+    let client_clone = client.clone();
+    let state_clone = state.clone();
+    let mut recv_task = tokio::spawn(async move {
+        while let Some(Ok(message)) = receiver.next().await {
+            client_clone.on_message(message).await;
+        }
+    });
     //
-    // // Attente de la fin d'une des tâches (déconnexion)
-    // tokio::select! {
-    //     _ = (&mut send_task) => recv_task.abort(),
-    //     _ = (&mut recv_task) => send_task.abort(),
-    // };
+    // Attente de la fin d'une des tâches (déconnexion)
+    tokio::select! {
+        _ = (&mut send_task) => recv_task.abort(),
+        _ = (&mut recv_task) => send_task.abort(),
+    };
     //
     // // Déconnexion (Disconnect)
     // on_disconnect(user_id, &state).await;

src/routes/gateway/mod.rs

@@ -1,6 +1,8 @@
 use crate::models::user::Model as User;
+use axum::extract::ws::Message;
 use parking_lot::RwLock;
 use std::collections::HashMap;
+use tokio::sync::mpsc;
 use uuid::Uuid;
 
 pub mod handlers;
@@ -16,6 +18,7 @@ pub struct GatewayManager {
 pub struct GatewayClient {
     user: User,
     connection_id: Uuid,
+    sender: mpsc::UnboundedSender<Message>,
 }
 
 impl GatewayManager {
@@ -37,11 +40,12 @@ impl GatewayManager {
 }
 
 impl GatewayClient {
-    pub fn new(user: User) -> Self {
+    pub fn new(user: User, sender: mpsc::UnboundedSender<Message>) -> Self {
         let connection_id = Uuid::new_v4();
         Self {
             user,
             connection_id,
+            sender,
         }
     }
 
@@ -49,5 +53,15 @@ impl GatewayClient {
 
     async fn on_disconnect(&self) {}
 
-    async fn on_message(&self) {}
+    async fn on_message(&self, message: Message) {
+        match message {
+            Message::Binary(content) => {}
+            Message::Text(content) => {
+                tracing::info!("Received text message: {}", content);
+            }
+            Message::Ping(_) => {}
+            Message::Pong(_) => {}
+            Message::Close(_) => {}
+        }
+    }
 }