Merge pull request #6 from tthogho1/websocket_QA

add video chat function

Cargo.toml

@@ -16,4 +16,5 @@ bb8 = "0.8"
 bb8-redis = "0.17.0"
 serde = { version = "1.0.210", features = ["derive"] }
 serde_json = "1.0.128"
-once_cell = "1.20.2"
+once_cell = "1.20.2"
+mockall = "0.13.0"

Dockerfile

@@ -25,5 +25,8 @@ FROM debian:buster-slim
 RUN apt-get update && rm -rf /var/lib/apt/lists/*
 
 COPY --from=builder /app/target/release/websocket_rust /usr/local/bin/websocket_rust
+COPY --from=builder /app/static /usr/local/bin/static
+
+EXPOSE 8000
 
 CMD ["websocket_rust"]

README.md

@@ -1 +1,23 @@
-"# websocket_rust" 
+# websocket_rust
+
+## Introduction
+
+[websocket_rust](https://github.com/taiki-ogawa/websocket_rust) is a websocket server based on axum and redis. It is a sample for how to use websocket in Rust
+You can connect to a Redis server and save username and location information on the Redis server.
+
+## Installation
+git clone https://github.com/taiki-ogawa/websocket_rust
+cd websocket_rust
+cargo build --release
+
+## env variables
+PORT: port number
+REDIS_URL: redis url 
+WS_SERVER= address for websocket  ex: ws://localhost:8000/ws
+
+## Run
+* Place the static folder in the same directory as websocket_rust.exe
+* execute websocket_rust.exe
+
+## Usage
+http://localhost:8080/Chat

src/handlers/websocket.rs

@@ -12,6 +12,7 @@ use std::sync::Arc;
 use websocket_rust::AppState;
 use websocket_rust::ChatMessage;
 use websocket_rust::get_app_state;
+use websocket_rust::MessageContent;
 
 use crate::handlers::position::delete_user;
 
@@ -41,14 +42,22 @@ async fn handle_socket(socket: WebSocket, state: Arc<AppState>, name: String) {
     // ブロードキャストチャンネルの受信機を取得
     let mut rx = state.tx.subscribe();
 
-    // 受信したメッセージをブロードキャストするタスク
     let mut send_task = tokio::spawn(async move {
         while let Ok(msg) = rx.recv().await {
-            if msg.to_id.is_empty() || (msg.to_id == user_id_clone) {
+            if !msg.to_id.is_empty(){ 
+                if msg.to_id == user_id_clone {
+                    println!("{}: {}", msg.user_id, msg.message);
+                    let json_string = serde_json::to_string(&msg).unwrap();
+                    let _ = sender
+                            .send(Message::Text(json_string))
+                            .await;                        
+                }
+            }else{
+                let json_string = serde_json::to_string(&msg).unwrap();
                 let _ = sender
-                .send(Message::Text(format!("{}: {}", msg.user_id, msg.message)))
-                .await;                
-            }
+                .send(Message::Text(json_string))
+                .await; 
+            } 
         }
     });
 

src/lib.rs

@@ -1,6 +1,7 @@
 use bb8_redis::{bb8, RedisConnectionManager};
 
 use serde :: Deserialize;
+use serde :: Serialize;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::Sender;
 use std::sync::Arc;
@@ -14,11 +15,49 @@ pub struct AppState {
     pub tx: broadcast::Sender<ChatMessage>,
 }
 
-#[derive(Clone, Debug,Deserialize)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct ChatMessage {
     pub user_id: String,
     pub to_id: String,
-    pub message: String,
+    pub message: MessageContent,
+}
+
+#[derive(Clone, Debug,Deserialize , Serialize)]
+#[serde(untagged)]
+pub enum MessageContent {
+    Text(String),
+    Sdp(Sdp),
+    Ice(Ice),
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Sdp {
+    pub r#type: String,
+    pub sdp: String,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Ice{
+    pub r#type: String,
+    pub candidate: Candidate,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[allow(non_snake_case)]
+pub struct Candidate {
+    pub candidate: String,
+    pub sdpMid: String,
+    pub sdpMLineIndex: u32,
+}
+
+impl std::fmt::Display for MessageContent {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match self {
+            MessageContent::Text(text) => write!(f, "{}", text),
+            MessageContent::Sdp(sdp) => write!(f, "Sdp(type: {}, sdp: {})", sdp.r#type, sdp.sdp),
+            MessageContent::Ice(ice) => write!(f, "Ice(type: {}, candidate: {})", ice.r#type, ice.candidate.candidate),
+        }
+    }
 }
 
 impl AppState {
@@ -61,4 +100,61 @@ pub async fn create_pool(redis_url: &str) -> Result<bb8::Pool<RedisConnectionMan
 pub async fn get_connection(pool: &bb8::Pool<RedisConnectionManager>) -> Result<bb8::PooledConnection<'_, RedisConnectionManager>, Box<dyn std::error::Error>> {
     let conn = pool.get().await?;
     Ok(conn)
-}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mockall::predicate::*;
+    use mockall::mock;
+    use dotenv::dotenv;
+    use std::env;
+
+    mock! {
+        RedisConnectionManager {}
+        impl Clone for RedisConnectionManager {
+            fn clone(&self) -> Self;
+        }
+    }
+
+    mock! {
+        Pool<RedisConnectionManager> {}
+    }
+
+    #[tokio::test]
+    async fn test_create_pool_success() {
+        dotenv().ok();
+        let redis_url = env::var("REDIS_URL").expect("REDIS_URL must be set in .env file");
+
+        let result = create_pool(&redis_url).await;
+
+        // 結果を検証
+        assert!(result.is_ok());
+
+        let pool = result.unwrap();
+        let result_con  = get_connection(&pool).await;
+
+        assert!(result_con.is_ok());
+
+        let _con = result_con.unwrap();
+
+        let state = pool.state();
+        let count = state.connections;
+
+        // 結果を検証
+        assert_eq!(count, 1);
+    }
+
+    #[tokio::test]
+    async fn test_create_pool_failure() {
+        let redis_url = "invalid_url";
+
+        // テスト実行
+        let result = create_pool(redis_url).await;
+
+        assert!(result.is_err());
+    }
+}
+
+
+

src/main.rs

@@ -32,6 +32,8 @@ async fn main() {
     // redis cloudへ接続
     let redis_url = env::var("REDIS_URL").unwrap().to_string();
     let port = env::var("PORT").expect("PORT environment variable not set").parse::<u16>().expect("PORT is not a number");
+    let ws_server = env::var("WS_SERVER").unwrap().to_string();
+
 
     let pool = create_pool(&redis_url).await.unwrap();
     let (tx, _rx) = broadcast::channel(100);
@@ -41,7 +43,7 @@ async fn main() {
     let app = Router::new()
         .route("/ws", get(ws_handler))
         .with_state(Arc::clone(&app_state));
-    let app = app.route("/Chat", get(move || async move {hello_handler(port).await}));
+    let app = app.route("/Chat", get(move || async move {hello_handler(ws_server).await}));
     let app = app.route("/position",post(position_handler)).with_state(Arc::clone(&app_state));
     let app: Router<Arc<websocket_rust::AppState>> = app.route("/users",post(getallusers_handler)).with_state(Arc::clone(&app_state));
     let app = app.route("/usersinbounds",post(get_users_in_bounds)).with_state(Arc::clone(&app_state));
@@ -63,6 +65,6 @@ async fn main() {
     axum::serve(listener, app).await.unwrap();
 }
 
-async fn hello_handler(port : u16) -> impl IntoResponse {
-    return render_template("test".to_string(), port);
+async fn hello_handler(ws_server: String) -> impl IntoResponse {
+    return render_template("test".to_string(), ws_server);
 }

src/view/mod.rs

@@ -6,12 +6,12 @@ use axum::response::IntoResponse;
 #[template(path = "hello.html")]
 struct HelloTemplate<'a> {
     name: &'a str,
-    port: &'a u16
+    ws_server: &'a str
 }
 
-pub fn render_template(name: String,port:u16) ->  impl IntoResponse {
+pub fn render_template(name: String,ws_server: String) ->  impl IntoResponse {
     // テンプレートのレンダリングロジック
-    let template = HelloTemplate { name: &name,port:&port };
+    let template = HelloTemplate { name: &name, ws_server:&ws_server };
     let rendered = template.render().unwrap();
 
     return Html(rendered);