Merge pull request #8 from waterbustech/feat/redis-cluster

feat: use redis cluster

Author: lambiengcode

Cargo.lock

@@ -35,7 +35,7 @@ socketioxide = { version = "0.17.1", features = [
     "state",
     "msgpack",
 ] }
-socketioxide-redis = "0.2.2"
+socketioxide-redis = { version = "0.2.2", features = ["redis-cluster"] }
 tokio = { version = "1.45.0", features = ["full"] }
 tokio-util = "0.7.15"
 tower = { version = "0.5.2", default-features = false }
@@ -91,7 +91,8 @@ tonic = "0.13.1"
 etcd-client = "0.15.0"
 sysinfo = "0.35.1"
 futures-util = "0.3.31"
-redis = "0.31.0"
+deadpool-redis = "0.22.0"
+redis = { version = "0.32.4", features = ["cluster"] }
 futures = "0.3.31"
 crossbeam = "0.8.4"
 mimalloc = "0.1.46"

Cargo.toml

@@ -15,4 +15,5 @@ async-channel = { workspace = true }
 serde_json = { workspace = true }
 serde = { workspace = true }
 futures-util = { workspace = true }
+deadpool-redis = { workspace = true }
 redis = { workspace = true }

crates/dispatcher/Cargo.toml

@@ -24,7 +24,7 @@ pub struct DispatcherConfigs {
     pub group_id: String,
     pub dispatcher_port: u16,
     pub sfu_port: u16,
-    pub redis_uri: String,
+    pub redis_uris: Vec<String>,
     pub etcd_uri: String,
     pub sender: Sender<DispatcherCallback>,
 }
@@ -51,7 +51,7 @@ impl DispatcherManager {
         .unwrap();
 
         let sfu_grpc_client = SfuGrpcClient::default();
-        let cache_manager = CacheManager::new(configs.redis_uri);
+        let cache_manager = CacheManager::new(configs.redis_uris);
 
         Self {
             sfu_grpc_client,

crates/dispatcher/src/dispatcher_manager.rs

@@ -1,4 +1,4 @@
-use redis::{Client, Commands};
+use redis::{Commands, cluster::ClusterClient};
 use serde::{Deserialize, Serialize};
 use std::sync::{Arc, Mutex};
 
@@ -21,14 +21,14 @@ impl CacheKey {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct CacheManager {
-    client: Arc<Mutex<Client>>,
+    client: Arc<Mutex<ClusterClient>>,
 }
 
 impl CacheManager {
-    pub fn new(redis_uri: String) -> Self {
-        let client = Client::open(redis_uri).unwrap();
+    pub fn new(urls: Vec<String>) -> Self {
+        let client = ClusterClient::new(urls).unwrap();
         Self {
             client: Arc::new(Mutex::new(client)),
         }

crates/dispatcher/src/infrastructure/cache/cache_manager.rs

@@ -1,23 +1,35 @@
-# APP
 APP_PORT=5998
+CLIENT_SECRET_KEY=
+SERVER_SECRET_KEY=
+TLS_ENABLED=false
 
-# DB
-DATABASE_URI=postgresql://postgres:postgres@localhost:5432/mydatabase
-REDIS_URI=redis://localhost:6379
+DATABASE_URI=postgres://postgres:password@localhost:5432/database
 
-# AWS Configure
-AWS_ACCESS_KEY_ID=YOUR_AWS_ACCESS_KEY_ID
-AWS_SECRET_ACCESS_KEY=YOUR_AWS_SECRET_ACCESS_KEY
-AWS_REGION=ap-southeast-1
-AWS_S3_BUCKET_NAME=your-s3-bucket-name
+REDIS_URIS=redis://127.0.0.1:6379?protocol=resp3,redis://127.0.0.1:6380?protocol=resp3,redis://127.0.0.1:6381?protocol=resp3,redis://127.0.0.1:6382?protocol=resp3,redis://127.0.0.1:6383?protocol=resp3,redis://127.0.0.1:6384?protocol=resp3
 
-# JWT Configure
-AUTH_JWT_SECRET=your_jwt_secret
-AUTH_JWT_TOKEN_EXPIRES_IN=24h
-AUTH_REFRESH_SECRET=your_refresh_secret
-AUTH_REFRESH_TOKEN_EXPIRES_IN=3650d
+STORAGE_ACCOUNT_ID=
+STORAGE_ACCESS_KEY_ID=
+STORAGE_SECRET_ACCESS_KEY=
+STORAGE_REGION=auto
+STORAGE_BUCKET_NAME=
+STORAGE_ENDPOINT_URL=
+STORAGE_CUSTOM_DOMAIN=
+
+AUTH_JWT_SECRET=
+AUTH_JWT_TOKEN_EXPIRES_IN=86400
+AUTH_REFRESH_SECRET=
+AUTH_REFRESH_TOKEN_EXPIRES_IN=31536000
 
-# WebRTC
 PUBLIC_IP=
-PORT_MIN_UDP=19200
-PORT_MAX_UDP=19250
+PORT_MIN_UDP=19000
+PORT_MAX_UDP=60000
+
+GROUP_ID=waterbus-group-1
+SFU_HOST=http://0.0.0.0
+SFU_PORT=50051
+DISPATCHER_HOST=http://0.0.0.0
+DISPATCHER_PORT=50052
+ETCD_URI=127.0.0.1:2379
+
+MOQ_URI=http://localhost:4443/waterbus/
+HLS_MODE=LOCAL