W/o Actor

Cargo.lock

@@ -1181,7 +1181,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
 
 [[package]]
 name = "hulypulse"
-version = "0.1.0"
+version = "0.1.7"
 dependencies = [
  "actix",
  "actix-cors",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hulypulse"
-version = "0.1.0"
+version = "0.1.7"
 edition = "2024"
 
 [dependencies]

scripts/TEST_HTTP_API.sh

@@ -10,6 +10,11 @@ TOKEN=$(./token.sh claims.json)
 
 ZP="00000000-0000-0000-0000-000000000001/TESTS"
 
+    delete "0000000/TESTS"
+    delete ${ZP}
+    put ${ZP} "Value_1" "HULY-TTL: 2"
+    delete ${ZP}
+
 
 echo "--------- authorization_test ----------"
 TOKEN=""
@@ -20,7 +25,6 @@ TOKEN=$(./token.sh claims_wrong_ws.json)
     put "00000000-0000-0000-0000-000000000001/TESTS" "Value_1" "HULY-TTL: 2"
 TOKEN=$(./token.sh claims.json)
     put "00000000-0000-0000-0000-000000000002/TESTS" "Value_1" "HULY-TTL: 2"
-exit
 
 
 
@@ -36,9 +40,6 @@ echo "--------- if-match ----------"
     get "00000000-0000-0000-0000-000000000001/TESTS/3/secret$/"
 
 
-
-exit
-
 echo "--------- Deprecated symbols ----------"
 
     put "00000000-0000-0000-0000-000000000001/'TESTS" "Value_1" "HULY-TTL: 2"

scripts/pulse_lib.sh

@@ -72,5 +72,6 @@ delete() {
   local tmpfile
   tmpfile=$(mktemp)
   curl -i -s -X DELETE "$URL/$1" -H "Authorization: Bearer ${TOKEN}" | tr -d '\r' > "$tmpfile"
+#  curl -v -i -s -X DELETE "$URL/$1" -H "Authorization: Bearer ${TOKEN}" | tr -d '\r' > "$tmpfile"
   api ${tmpfile}
 }

src/handlers_http.rs

@@ -24,7 +24,6 @@ use actix_web::{
 };
 
 use crate::redis::{SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save};
-use crate::workspace_owner::workspace_check;
 
 pub fn map_handler_error(err: impl std::fmt::Display) -> Error {
     let msg = err.to_string();
@@ -46,7 +45,6 @@ pub fn map_handler_error(err: impl std::fmt::Display) -> Error {
 
 #[derive(Deserialize, Debug)]
 pub struct PathParams {
-    //workspace: Uuid,
     key: String,
 }
 
@@ -95,14 +93,6 @@ pub async fn get(
     .map_err(map_handler_error)
 }
 
-/*
-#[derive(serde::Deserialize)]
-struct MyHeaders {
-    #[serde(rename = "HULY-TTL")]
-    ttl: Option<u64>,
-}
-*/
-
 /// put
 pub async fn put(
     req: HttpRequest,
@@ -172,7 +162,6 @@ pub async fn delete(
     path: web::Path<PathParams>,
     redis: web::Data<MultiplexedConnection>,
 ) -> Result<HttpResponse, actix_web::error::Error> {
-    workspace_check(&req)?; // Check workspace
 
     let key: String = path.into_inner().key;
 

src/handlers_ws.rs

@@ -13,8 +13,10 @@
 // limitations under the License.
 //
 
+// https://github.com/hcengineering/hulypulse/
+
 use actix::{
-    Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, WrapFuture, fut, prelude::*,
+    Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut,
 };
 use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse, web};
 use actix_web_actors::ws;
@@ -25,10 +27,10 @@ use serde_json::{Value, json};
 use crate::redis::{
     SaveMode, Ttl, deprecated_symbol, redis_delete, redis_list, redis_read, redis_save,
 };
-use crate::ws_hub::{
-    Connect, Disconnect, ServerMessage, SessionId, Subscribe, SubscribeList, Unsubscribe,
-    UnsubscribeAll, WsHub,
-};
+
+use crate::hub_service::{HubServiceHandle, ServerMessage, SessionId, new_session_id};
+use crate::workspace_owner::check_workspace_core;
+
 
 #[derive(Serialize, Default)]
 struct ReturnBase<'a> {
@@ -130,7 +132,7 @@ use hulyrs::services::jwt::Claims;
 pub struct WsSession {
     pub redis: MultiplexedConnection,
     pub id: SessionId,
-    pub hub: Addr<WsHub>,
+    pub hub: HubServiceHandle,
     pub claims: Claims,
 }
 
@@ -139,35 +141,16 @@ impl Actor for WsSession {
     type Context = ws::WebsocketContext<Self>;
 
     fn started(&mut self, ctx: &mut Self::Context) {
-        // ask ID from Hub
         let addr = ctx.address();
-
         let recipient = addr.recipient::<ServerMessage>();
-        // println!("WebSocket connected");
-        self.hub
-            .send(Connect {
-                addr: recipient,
-                session_id: self.id,
-            })
-            .into_actor(self)
-            .map(|res, act, _ctx| match res {
-                Ok(id) => {
-                    act.id = id;
-                    tracing::info!("WebSocket connected: {id}");
-                }
-                Err(e) => {
-                    tracing::error!("WebSocket failed connect to hub: {e}");
-                    _ctx.stop();
-                }
-            })
-            .wait(ctx); // waiting for ID
+
+        self.hub.connect(self.id, recipient);
+        tracing::info!("WebSocket connected: {}", self.id);
     }
 
     fn stopped(&mut self, _ctx: &mut Self::Context) {
         if self.id != 0 {
-            self.hub.do_send(Disconnect {
-                session_id: self.id,
-            });
+            self.hub.disconnect(self.id);
         }
         tracing::info!("WebSocket disconnected: {:?}", &self.id);
     }
@@ -175,14 +158,13 @@ impl Actor for WsSession {
 
 impl actix::Handler<ServerMessage> for WsSession {
     type Result = ();
-
     fn handle(&mut self, msg: ServerMessage, ctx: &mut Self::Context) {
-        let json =
-            serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into());
+        let json = serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into());
         ctx.text(json);
     }
 }
 
+
 /// StreamHandler External trait: must be in separate impl block
 impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
     fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
@@ -201,8 +183,6 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
     }
 }
 
-use crate::workspace_owner::check_workspace_core;
-
 /// All logic in one impl
 impl WsSession {
     fn ws_error(&self, ctx: &mut ws::WebsocketContext<Self>, msg: &str) {
@@ -221,7 +201,6 @@ impl WsSession {
     ) {
         ctx.wait(
             fut::wrap_future(fut).map(move |res, _actor: &mut Self, ctx| {
-                // if !base.is_object() { base = json!({ "base": base }); }
                 let obj = base.as_object_mut().unwrap();
                 match res {
                     Ok(Value::Object(extra)) => {
@@ -251,14 +230,7 @@ impl WsSession {
                 if_none_match,
                 correlation,
             } => {
-                tracing::info!(
-                    "PUT {} = {} (expires_at: {:?}) (ttl: {:?}) correlation: {:?}",
-                    &key,
-                    &data,
-                    &expires_at,
-                    &ttl,
-                    &correlation
-                );
+                tracing::info!("PUT {} = {}", &key, &data); //  (expires_at: {:?}) (ttl: {:?}) correlation: {:?} &expires_at, &ttl, &correlation
 
                 // Check workspace
                 if let Err(e) = self.workspace_check_ws(&key) {
@@ -324,7 +296,7 @@ impl WsSession {
                 correlation,
                 if_match,
             } => {
-                tracing::info!("DELETE {} correlation:{:?}", &key, &correlation);
+                tracing::info!("DELETE {}", &key); //  correlation:{:?} , &correlation
 
                 // Check workspace
                 if let Err(e) = self.workspace_check_ws(&key) {
@@ -332,6 +304,8 @@ impl WsSession {
                     return;
                 }
 
+                tracing::info!("DELETE!!! {}", &key);
+
                 let mut redis = self.redis.clone();
 
                 let base = serde_json::json!(ReturnBase {
@@ -371,7 +345,7 @@ impl WsSession {
             }
 
             WsCommand::Get { key, correlation } => {
-                tracing::info!("GET {} correlation:{:?}", &key, &correlation);
+                tracing::info!("GET {}", &key); //  correlation:{:?} , &correlation
 
                 // Check workspace
                 if let Err(e) = self.workspace_check_ws(&key) {
@@ -407,7 +381,7 @@ impl WsSession {
             }
 
             WsCommand::List { key, correlation } => {
-                tracing::info!("LIST {:?} correlation: {:?}", &key, &correlation);
+                tracing::info!("LIST {:?}", &key); //  correlation: {:?} , &correlation
 
                 // Check workspace
                 if let Err(e) = self.workspace_check_ws(&key) {
@@ -434,9 +408,10 @@ impl WsSession {
                 self.fut_send(ctx, fut, base);
             }
 
+
             WsCommand::Sub { key, correlation } => {
                 // LEVENT 3
-                tracing::info!("SUB {} correlation: {:?}", &key, &correlation);
+                tracing::info!("SUB {}", &key); //  correlation: {:?} , &correlation
 
                 // Check workspace
                 if let Err(e) = self.workspace_check_ws(&key) {
@@ -456,19 +431,16 @@ impl WsSession {
                 if deprecated_symbol(&key) {
                     map.insert("error".into(), json!("Deprecated symbol in key"));
                 } else {
-                    self.hub.do_send(Subscribe {
-                        session_id: self.id,
-                        key: key.clone(),
-                    });
+                    self.hub.subscribe(self.id, key.clone());
                     map.insert("result".into(), json!("OK"));
                 }
-
                 ctx.text(obj.to_string());
             }
 
+
             WsCommand::Unsub { key, correlation } => {
                 // LEVENT 4
-                tracing::info!("UNSUB {} correlation: {:?}", &key, &correlation);
+                tracing::info!("UNSUB {}", &key); //  correlation: {:?} , &correlation
 
                 let mut obj = serde_json::json!(ReturnBase {
                     action: "unsub",
@@ -480,9 +452,7 @@ impl WsSession {
                 let map = obj.as_object_mut().unwrap();
 
                 if key == "*" {
-                    self.hub.do_send(UnsubscribeAll {
-                        session_id: self.id,
-                    });
+                    self.hub.unsubscribe_all(self.id);
                     map.insert("result".into(), json!("OK"));
                 } else {
                     // Check workspace
@@ -490,26 +460,20 @@ impl WsSession {
                         self.ws_error(ctx, e);
                         return;
                     }
-
                     if deprecated_symbol(&key) {
                         map.insert("error".into(), json!("Deprecated symbol in key"));
                     } else {
+                        self.hub.unsubscribe(self.id, key.clone());
                         map.insert("result".into(), json!("OK"));
-                        self.hub.do_send(Unsubscribe {
-                            session_id: self.id,
-                            key: key.clone(),
-                        });
                     }
-                };
-
+                }
                 ctx.text(obj.to_string());
             }
 
+
             WsCommand::Sublist { correlation } => {
-                tracing::info!("SUBLIST correlation: {:?}", &correlation);
-
+                tracing::info!("SUBLIST"); //  correlation: {:?} , &correlation
                 // w/o Check workspace!
-
                 let base = serde_json::json!(ReturnBase {
                     action: "list",
                     correlation: correlation.as_deref(),
@@ -519,27 +483,27 @@ impl WsSession {
                 let hub = self.hub.clone();
                 let id = self.id;
 
-                let fut = async move {
-                    let keys = hub
-                        .send(SubscribeList { session_id: id })
-                        .await
-                        .unwrap_or_default();
-                    Ok(json!({ "result": keys }))
-                };
+                self.fut_send(
+                    ctx,
+                    async move {
+                        let keys = hub.subscribe_list(id).await;
+                        Ok(json!({ "result": keys }))
+                    },
+                    base,
+                );
+            }
 
-                self.fut_send(ctx, fut, base);
-            } // End of commands
+        // End of commands
         }
     }
 }
 
-// ---- auth
 
 pub async fn handler(
     req: HttpRequest,
     payload: web::Payload,
     redis: web::Data<MultiplexedConnection>,
-    hub: web::Data<Addr<WsHub>>,
+    hub: web::Data<HubServiceHandle>, // <-- было Addr<WsHub>
 ) -> Result<HttpResponse, Error> {
     let claims = req
         .extensions()
@@ -550,7 +514,7 @@ pub async fn handler(
     let session = WsSession {
         redis: redis.get_ref().clone(),
         hub: hub.get_ref().clone(),
-        id: crate::ws_hub::new_session_id(),
+        id: new_session_id(),
         claims,
     };
 

src/hub_service.rs

@@ -0,0 +1,266 @@
+use std::collections::{HashMap, HashSet};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use actix::prelude::*;
+
+use redis::aio::MultiplexedConnection;
+use serde::Serialize;
+use tokio::sync::{mpsc, oneshot};
+
+fn subscription_matches(sub_key: &str, key: &str) -> bool {
+    if sub_key == key {
+        return true;
+    }
+    if sub_key.ends_with('/') && key.starts_with(sub_key) {
+        let rest = &key[sub_key.len()..];
+        return !rest.contains('$');
+    }
+    false
+}
+
+#[derive(Clone, Serialize, Debug, Message)]
+#[rtype(result = "()")]
+pub struct ServerMessage {
+    #[serde(flatten)]
+    pub event: RedisEvent,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub value: Option<String>,
+}
+
+// ==== ID ====
+
+pub type SessionId = u64;
+static NEXT_ID: AtomicU64 = AtomicU64::new(1);
+pub fn new_session_id() -> SessionId {
+    NEXT_ID.fetch_add(1, Ordering::SeqCst)
+}
+
+// ==== Redis events ====
+
+#[derive(Debug, Clone, Serialize)]
+pub enum RedisEventAction {
+    Set,
+    Del,
+    Unlink,
+    Expired,
+    Other(String),
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct RedisEvent {
+    pub db: u32,
+    pub key: String,
+    pub action: RedisEventAction,
+}
+
+// ==== Commands for worker ====
+
+enum Command {
+    Connect {
+        session_id: SessionId,
+        addr: Recipient<ServerMessage>,
+    },
+    Disconnect {
+        session_id: SessionId,
+    },
+    Subscribe {
+        session_id: SessionId,
+        key: String,
+    },
+    Unsubscribe {
+        session_id: SessionId,
+        key: String,
+    },
+    UnsubscribeAll {
+        session_id: SessionId,
+    },
+    SubscribeList {
+        session_id: SessionId,
+        reply: oneshot::Sender<Vec<String>>,
+    },
+    Count {
+        reply: oneshot::Sender<usize>,
+    },
+    DumpSubs {
+        reply: oneshot::Sender<std::collections::HashMap<String, Vec<SessionId>>>,
+    },
+    RedisEvent(RedisEvent),
+}
+
+// ==== Handle  ====
+
+#[derive(Clone)]
+pub struct HubServiceHandle {
+    tx: mpsc::Sender<Command>,
+}
+
+impl HubServiceHandle {
+    pub fn start(redis: MultiplexedConnection) -> Self {
+        let (tx, mut rx) = mpsc::channel::<Command>(1024);
+
+        // Владелец состояния живёт внутри задачи
+        tokio::spawn(async move {
+            let mut sessions: HashMap<SessionId, Recipient<ServerMessage>> = HashMap::new();
+            let mut subs: HashMap<String, HashSet<SessionId>> = HashMap::new();
+            let mut redis_conn = redis;
+
+            fn subscribers_for(
+                subs: &HashMap<String, HashSet<SessionId>>,
+                key: &str,
+            ) -> HashSet<SessionId> {
+                let mut out = HashSet::<SessionId>::new();
+                for (sub_key, set) in subs.iter() {
+                    if subscription_matches(sub_key, key) {
+                        out.extend(set.iter().copied());
+                    }
+                }
+                out
+            }
+
+            while let Some(cmd) = rx.recv().await {
+                match cmd {
+
+                    Command::Connect { session_id, addr } => {
+                        sessions.insert(session_id, addr);
+                    }
+
+                    Command::Disconnect { session_id } => {
+                        subs.retain(|_, ids| {
+                            ids.remove(&session_id);
+                            !ids.is_empty()
+                        });
+                        sessions.remove(&session_id);
+                    }
+
+                    Command::Subscribe { session_id, key } => {
+                        subs.entry(key).or_default().insert(session_id);
+                    }
+
+                    Command::Unsubscribe { session_id, key } => {
+                        if let Some(set) = subs.get_mut(&key) {
+                            set.remove(&session_id);
+                            if set.is_empty() {
+                                subs.remove(&key);
+                            }
+                        }
+                    }
+
+                    Command::UnsubscribeAll { session_id } => {
+                        subs.retain(|_, ids| {
+                            ids.remove(&session_id);
+                            !ids.is_empty()
+                        });
+                    }
+
+                    Command::SubscribeList { session_id, reply } => {
+                        let list = subs
+                            .iter()
+                            .filter_map(|(key, ids)| {
+                                if ids.contains(&session_id) {
+                                    Some(key.clone())
+                                } else {
+                                    None
+                                }
+                            })
+                            .collect::<Vec<_>>();
+                        let _ = reply.send(list);
+                    }
+
+                    Command::Count { reply } => {
+                        let _ = reply.send(sessions.len());
+                    }
+
+                    Command::DumpSubs { reply } => {
+                        let snapshot = subs
+                            .iter()
+                            .map(|(k, set)| (k.clone(), set.iter().copied().collect::<Vec<_>>()))
+                            .collect::<std::collections::HashMap<_, _>>();
+                        let _ = reply.send(snapshot);
+                    }
+
+                    Command::RedisEvent(event) => {
+                        let targets = subscribers_for(&subs, &event.key);
+                        if targets.is_empty() {
+                            continue;
+                        }
+                        let recipients: Vec<Recipient<ServerMessage>> = targets
+                            .into_iter()
+                            .filter_map(|sid| sessions.get(&sid).cloned())
+                            .collect();
+
+                        // Inside: waiting GET
+                        let need_get = matches!(event.action, RedisEventAction::Set);
+                        let mut value: Option<String> = None;
+                        if need_get {
+                            match redis::cmd("GET")
+                                .arg(&event.key)
+                                .query_async::<Option<String>>(&mut redis_conn)
+                                .await
+                            {
+                                Ok(v) => value = v,
+                                Err(e) => {
+                                    tracing::warn!("redis GET {} failed: {}", &event.key, e);
+                                }
+                            }
+                        }
+
+                        let payload = ServerMessage {
+                            event,
+                            value,
+                        };
+
+                        for rcpt in recipients {
+                            let _ = rcpt.do_send(payload.clone());
+                        }
+                    }
+                }
+            }
+        });
+
+        Self { tx }
+    }
+
+    // ---- API, ничего не выполняет параллельно внутри worker'а ----
+
+    pub fn connect(&self, session_id: SessionId, addr: Recipient<ServerMessage>) {
+        let _ = self.tx.try_send(Command::Connect { session_id, addr });
+    }
+
+    pub fn disconnect(&self, session_id: SessionId) {
+        let _ = self.tx.try_send(Command::Disconnect { session_id });
+    }
+
+    pub fn subscribe(&self, session_id: SessionId, key: String) {
+        let _ = self.tx.try_send(Command::Subscribe { session_id, key });
+    }
+
+    pub fn unsubscribe(&self, session_id: SessionId, key: String) {
+        let _ = self.tx.try_send(Command::Unsubscribe { session_id, key });
+    }
+
+    pub fn unsubscribe_all(&self, session_id: SessionId) {
+        let _ = self.tx.try_send(Command::UnsubscribeAll { session_id });
+    }
+
+    pub async fn subscribe_list(&self, session_id: SessionId) -> Vec<String> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.tx.send(Command::SubscribeList { session_id, reply: tx }).await;
+        rx.await.unwrap_or_default()
+    }
+
+    pub async fn count(&self) -> usize {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.tx.send(Command::Count { reply: tx }).await;
+        rx.await.unwrap_or_default()
+    }
+
+    pub async fn dump_subs(&self) -> std::collections::HashMap<String, Vec<SessionId>> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.tx.send(Command::DumpSubs { reply: tx }).await;
+        rx.await.unwrap_or_default()
+    }
+
+    pub fn push_event(&self, ev: RedisEvent) {
+        let _ = self.tx.try_send(Command::RedisEvent(ev));
+    }
+}

src/main.rs

@@ -13,29 +13,26 @@
 // limitations under the License.
 //
 
-use actix::prelude::*;
 use actix_cors::Cors;
 use actix_web::{
-    App, Error, HttpMessage, HttpResponse, HttpServer,
-    body::MessageBody,
-    dev::{ServiceRequest, ServiceResponse},
-    middleware::{self, Next},
-    web::{self, Path, Query},
+    body::MessageBody, dev::{ServiceRequest, ServiceResponse}, middleware::{self, Next}, web::{self, Path, Query}, App, Error, HttpMessage, HttpResponse, HttpServer
 };
 use hulyrs::services::jwt::{Claims, actix::ServiceRequestExt};
 use secrecy::ExposeSecret;
+use serde_json::json;
 use tracing::*;
+use uuid::Uuid;
 
 mod config;
 mod handlers_http;
 mod handlers_ws;
 mod redis;
 mod workspace_owner;
-mod ws_hub;
+
+mod hub_service;
+use hub_service::HubServiceHandle;
 
 use config::CONFIG;
-use uuid::Uuid;
-use ws_hub::{TestGetSubs, WsHub};
 
 fn initialize_tracing(level: tracing::Level) {
     use tracing_subscriber::{filter::targets::Targets, prelude::*};
@@ -100,14 +97,22 @@ async fn main() -> anyhow::Result<()> {
     let redis_client = redis::client().await?;
     let redis_connection = redis_client.get_multiplexed_async_connection().await?;
 
-    // starting Hub
-    let hub = WsHub::new(redis_connection.clone()).start();
+    // starting HubService
+    let hub = HubServiceHandle::start(redis_connection.clone());
 
     // starting Logger
     tokio::spawn(redis::receiver(redis_client, hub.clone()));
 
     let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port);
 
+    let url = format!("http://{}:{}", &CONFIG.bind_host, &CONFIG.bind_port);
+    tracing::info!("Server running at {}", &url);
+    tracing::info!("HTTP API: {}/api",  &url);
+    tracing::info!("WebSocket API: {}/ws", &url);
+    tracing::info!("Status: {}/status", &url);
+    tracing::info!("Stats: {}/stat", &url);
+    tracing::info!("Subscriptions: {}/subs", &url);
+
     let server = HttpServer::new(move || {
         let cors = Cors::default()
             .allow_any_origin()
@@ -130,32 +135,21 @@ async fn main() -> anyhow::Result<()> {
                     .route("/{key:.+}", web::put().to(handlers_http::put))
                     .route("/{key:.+}", web::delete().to(handlers_http::delete)),
             )
-            .route(
-                "/ws",
-                web::get()
-                    .to(handlers_ws::handler)
+            .route("/ws", web::get().to(handlers_ws::handler)
                     .wrap(middleware::from_fn(extract_claims)),
             ) // WebSocket
             .route("/status", web::get().to(async || "ok"))
-            //
-            .route(
-                "/stat2",
-                web::get().to(|hub: web::Data<Addr<WsHub>>| async move {
-                    let count = hub.send(crate::ws_hub::Count).await.unwrap_or(0);
-                    HttpResponse::Ok().json(serde_json::json!({ "connections": count }))
-                }),
-            )
-            .route(
-                "/subs",
-                web::get().to(|hub: web::Data<Addr<WsHub>>| async move {
-                    match hub.send(TestGetSubs).await {
-                        Ok(subs) => HttpResponse::Ok().json(subs),
-                        Err(_) => {
-                            HttpResponse::InternalServerError().body("Failed to get subscriptions")
-                        }
-                    }
-                }),
-            )
+
+            .route("/stat", web::get().to(|hub: web::Data<HubServiceHandle>| async move {
+                let count = hub.count().await;
+                Ok::<_, actix_web::Error>(HttpResponse::Ok().json(json!({ "connections": count })))
+            }))
+
+            .route("/subs", web::get().to(|hub: web::Data<HubServiceHandle>| async move {
+                let subs = hub.dump_subs().await;
+                Ok::<_, actix_web::Error>(HttpResponse::Ok().json(subs))
+            }))
+            
     })
     .bind(socket)?
     .run();

src/redis.rs

@@ -15,13 +15,11 @@
 
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use actix::Addr;
 use ::redis::Msg;
 use tokio_stream::StreamExt;
 use tracing::*;
 
-use crate::config::{CONFIG, RedisMode};
-use crate::ws_hub::{RedisEvent, RedisEventAction, WsHub};
+use crate::{config::{RedisMode, CONFIG}, hub_service::{HubServiceHandle, RedisEvent, RedisEventAction}};
 
 #[derive(serde::Serialize)]
 pub enum Ttl {
@@ -358,7 +356,8 @@ impl TryFrom<Msg> for RedisEvent {
     }
 }
 
-pub async fn receiver(redis_client: Client, hub: Addr<WsHub>) -> anyhow::Result<()> {
+
+pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Result<()> {
     let mut redis = redis_client.get_multiplexed_async_connection().await?;
     let mut pubsub = redis_client.get_async_pubsub().await?;
 
@@ -383,10 +382,10 @@ pub async fn receiver(redis_client: Client, hub: Addr<WsHub>) -> anyhow::Result<
     while let Some(message) = messages.next().await {
         match RedisEvent::try_from(message) {
             Ok(ev) => {
-                
-                debug!("redis event: {ev:#?}");
 
-                hub.do_send(ev);
+                // debug!("redis event: {ev:#?}");
+
+                hub.push_event(ev);
                 
             }
             Err(e) => {

src/workspace_owner.rs

@@ -13,12 +13,12 @@
 // limitations under the License.
 //
 
-use actix_web::{HttpMessage, HttpRequest};
 use hulyrs::services::jwt::Claims;
 use uuid::Uuid;
 
 // common checker
 pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> {
+
     if claims.is_system() {
         return Ok(());
     }
@@ -41,22 +41,4 @@ pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static s
     }
 
     Ok(())
-}
-
-/// HTTP API
-pub fn workspace_check(req: &HttpRequest) -> Result<(), actix_web::Error> {
-    let key = req
-        .match_info()
-        .get("key")
-        .ok_or_else(|| actix_web::error::ErrorBadRequest("Missing key in URL path"))?;
-    let claims = req
-        .extensions()
-        .get::<Claims>()
-        .cloned()
-        .ok_or_else(|| actix_web::error::ErrorUnauthorized("Missing auth claims"))?;
-
-    match check_workspace_core(&claims, key) {
-        Ok(()) => Ok(()),
-        Err(msg) => Err(actix_web::error::ErrorUnauthorized(msg)),
-    }
-}
+}

src/ws_hub.rs

@@ -1,309 +0,0 @@
-//
-// Copyright © 2025 Hardcore Engineering Inc.
-//
-// Licensed under the Eclipse Public License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License. You may
-// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-//
-// See the License for the specific language governing permissions and
-// limitations under the License.
-//
-
-use std::{
-    collections::{HashMap, HashSet},
-    sync::atomic::AtomicU64,
-};
-
-use actix::prelude::*;
-use redis::aio::MultiplexedConnection;
-use serde::Serialize;
-
-fn subscription_matches(sub_key: &str, key: &str) -> bool {
-    if sub_key == key {
-        return true;
-    }
-    if sub_key.ends_with('/') && key.starts_with(sub_key) {
-        let rest = &key[sub_key.len()..];
-        return !rest.contains('$');
-    }
-    false
-}
-
-#[derive(Message, Clone, Serialize, Debug)]
-#[rtype(result = "()")]
-pub struct ServerMessage {
-    #[serde(flatten)]
-    pub event: RedisEvent,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub value: Option<String>,
-}
-
-/// Count of active sessions
-#[derive(Message)]
-#[rtype(result = "usize")]
-pub struct Count;
-
-pub type SessionId = u64;
-
-static NEXT_ID: AtomicU64 = AtomicU64::new(1);
-
-pub struct WsHub {
-    sessions: HashMap<SessionId, Recipient<ServerMessage>>,
-    subs: HashMap<String, HashSet<SessionId>>, // Subscriptions array: key -> {id, id, id ...}
-    redis: MultiplexedConnection,
-}
-
-impl WsHub {
-    pub fn new(redis: MultiplexedConnection) -> Self {
-        Self {
-            sessions: HashMap::new(),
-            subs: HashMap::new(),
-            redis,
-        }
-    }
-}
-
-impl Actor for WsHub {
-    type Context = Context<Self>;
-}
-
-/// Connect
-#[derive(Message)]
-#[rtype(result = "SessionId")]
-pub struct Connect {
-    pub session_id: SessionId,
-    pub addr: Recipient<ServerMessage>,
-}
-
-pub fn new_session_id() -> SessionId {
-    NEXT_ID.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
-}
-
-impl Handler<Connect> for WsHub {
-    type Result = SessionId;
-
-    fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
-        // LEVENT 1
-        //let id = self.next_id;
-        //self.next_id = self.next_id.wrapping_add(1);
-        self.sessions.insert(msg.session_id, msg.addr);
-        // tracing::info!("session connected: id={id} (total={})", self.sessions.len());
-        msg.session_id
-    }
-}
-
-/// Disconnect
-#[derive(Message)]
-#[rtype(result = "()")]
-pub struct Disconnect {
-    pub session_id: SessionId,
-}
-
-impl Handler<Disconnect> for WsHub {
-    type Result = ();
-
-    fn handle(&mut self, msg: Disconnect, _ctx: &mut Context<Self>) {
-        // LEVENT 2
-
-        // Delete all subscribes
-        self.subs.retain(|_key, session_ids| {
-            session_ids.remove(&msg.session_id);
-            !session_ids.is_empty()
-        });
-
-        let existed = self.sessions.remove(&msg.session_id).is_some();
-        if existed {
-            // tracing::info!("session disconnected: id={} (total={})", msg.session_id, self.sessions.len());
-        } else {
-            tracing::warn!("disconnect for unknown id={}", msg.session_id);
-        }
-    }
-}
-
-/// SubscribeList
-#[derive(Message)]
-#[rtype(result = "Vec<String>")]
-pub struct SubscribeList {
-    pub session_id: SessionId,
-}
-
-impl Handler<SubscribeList> for WsHub {
-    type Result = MessageResult<SubscribeList>;
-
-    fn handle(&mut self, msg: SubscribeList, _ctx: &mut Context<Self>) -> Self::Result {
-        // Collect all keys with my session_id
-        let list = self
-            .subs
-            .iter()
-            .filter_map(|(key, sessions)| {
-                if sessions.contains(&msg.session_id) {
-                    Some(key.clone())
-                } else {
-                    None
-                }
-            })
-            .collect::<Vec<_>>();
-
-        MessageResult(list)
-    }
-}
-
-/// Count of IDs
-impl Handler<Count> for WsHub {
-    type Result = usize;
-
-    fn handle(&mut self, _: Count, _: &mut Context<Self>) -> Self::Result {
-        self.sessions.len()
-    }
-}
-
-/// Subscribe
-#[derive(Message)]
-#[rtype(result = "()")]
-pub struct Subscribe {
-    pub session_id: SessionId,
-    pub key: String,
-}
-
-impl Handler<Subscribe> for WsHub {
-    type Result = ();
-    fn handle(&mut self, msg: Subscribe, _ctx: &mut Context<Self>) {
-        self.subs.entry(msg.key).or_default().insert(msg.session_id);
-    }
-}
-
-/// Unsubscribe
-#[derive(Message)]
-#[rtype(result = "()")]
-pub struct Unsubscribe {
-    pub session_id: SessionId,
-    pub key: String,
-}
-
-impl Handler<Unsubscribe> for WsHub {
-    type Result = ();
-    fn handle(&mut self, msg: Unsubscribe, _ctx: &mut Context<Self>) {
-        if let Some(set) = self.subs.get_mut(&msg.key) {
-            set.remove(&msg.session_id);
-            if set.is_empty() {
-                self.subs.remove(&msg.key);
-            }
-        }
-    }
-}
-
-#[derive(Message)]
-#[rtype(result = "()")]
-pub struct UnsubscribeAll {
-    pub session_id: SessionId,
-}
-
-impl Handler<UnsubscribeAll> for WsHub {
-    type Result = ();
-    fn handle(&mut self, msg: UnsubscribeAll, _ctx: &mut Context<Self>) {
-        self.subs.retain(|_key, session_ids| {
-            session_ids.remove(&msg.session_id);
-            !session_ids.is_empty()
-        });
-    }
-}
-
-#[derive(Message)]
-#[rtype(result = "HashMap<String, Vec<SessionId>>")]
-pub struct TestGetSubs;
-
-impl Handler<TestGetSubs> for WsHub {
-    type Result = MessageResult<TestGetSubs>;
-
-    fn handle(&mut self, _msg: TestGetSubs, _ctx: &mut Context<Self>) -> Self::Result {
-        let s: HashMap<String, Vec<SessionId>> = self
-            .subs
-            .iter()
-            .map(|(key, ids)| (key.clone(), ids.iter().copied().collect()))
-            .collect();
-        MessageResult(s)
-    }
-}
-
-// List of subscribers
-impl WsHub {
-    fn subscribers_for(&self, key: &str) -> HashSet<SessionId> {
-        let mut out = HashSet::new();
-        for (sub_key, set) in &self.subs {
-            if subscription_matches(sub_key, key) {
-                out.extend(set.iter().copied());
-            }
-        }
-        out
-    }
-}
-
-#[derive(Debug, Clone, Serialize)]
-pub enum RedisEventAction {
-    Set,     // Insert or Update
-    Del,     // Delete
-    Unlink,  // async Delete
-    Expired, // TTL Delete
-    Other(String),
-}
-
-use actix::Message;
-
-#[derive(Debug, Clone, Serialize, Message)]
-#[rtype(result = "()")]
-pub struct RedisEvent {
-    pub db: u32,
-    pub key: String,
-    pub action: RedisEventAction,
-}
-
-impl Handler<RedisEvent> for WsHub {
-    type Result = ResponseActFuture<Self, ()>;
-
-    fn handle(&mut self, msg: RedisEvent, _ctx: &mut Context<Self>) -> Self::Result {
-        let targets = self.subscribers_for(&msg.key);
-        if targets.is_empty() {
-            return Box::pin(actix::fut::ready(()).into_actor(self));
-        }
-
-        let recipients: Vec<Recipient<ServerMessage>> = targets
-            .into_iter()
-            .filter_map(|sid| self.sessions.get(&sid).cloned())
-            .collect();
-
-        let mut redis = self.redis.clone();
-        let event = msg.clone();
-        let need_get = matches!(msg.action, RedisEventAction::Set);
-
-        Box::pin(
-            async move {
-                let value = if need_get {
-                    match redis::cmd("GET")
-                        .arg(&event.key)
-                        .query_async::<Option<String>>(&mut redis)
-                        .await
-                    {
-                        Ok(v) => v,
-                        Err(e) => {
-                            tracing::warn!("redis GET {} failed: {}", &event.key, e);
-                            None
-                        }
-                    }
-                } else {
-                    None
-                };
-
-                let payload = ServerMessage { event, value };
-
-                for rcpt in recipients {
-                    let _ = rcpt.do_send(payload.clone());
-                }
-            }
-            .into_actor(self),
-        )
-    }
-}