Refactoring modules: Actor-2 removed

README.md

@@ -178,11 +178,11 @@ Size of data is limited to some reasonable size
 
 ** Server to Client ** subscribed events:
 
-    - `{"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Set","value":"hello"}`
+    - `{"message":"Set","key":"00000000-0000-0000-0000-000000000001/foo/bar","value":"hello"}`
 
-    - `{"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Expired"}`
+    - `{"message":"Expired","key":"00000000-0000-0000-0000-000000000001/foo/bar"}`
 
-    - `{"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Del"}`
+    - `{"message":"Del","key":"00000000-0000-0000-0000-000000000001/foo/bar"}`
 
 
 ## Running

scripts/TEST.html

@@ -63,7 +63,7 @@
 
   <br />
   <button onclick="sendMessage()">Send JSON</button>
-  <button onclick="clean()">❌clear</button>
+  <button onclick="pr('clear')">❌clear</button>
 
   <p>
 
@@ -89,9 +89,15 @@
   <div id="output">Waiting for server response...</div>
 
   <script>
+
     const output = document.getElementById("output");
     const textarea = document.getElementById("jsonInput");
 
+    function pr(s) {
+	if(s=="clear") return output.textContent = "";
+	output.textContent = s+"\n\n"+output.textContent;
+    }
+
 //    const workspace="00000000-0000-0000-0000-000000000001";
 
     let token="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2NvdW50IjoiYWFhYWFhYWEtYmJiYi1jY2NjLWRkZGQtZWVlZWVlZWVlZWVlIiwiZXh0cmEiOnsic2VydmljZSI6ImFjY291bnQifSwid29ya3NwYWNlIjoiMDAwMDAwMDAtMDAwMC0wMDAwLTAwMDAtMDAwMDAwMDAwMDAxIn0.ZrwMvv_0CjuKeF2CkmHyMK2vHd9Ro4M3kHZcZBrBxZQ";
@@ -101,15 +107,16 @@
 
 
     ws.onopen = () => {
-      output.textContent = "✅ WebSocket connected.";
+	pr("clear");
+	pr("✅ WebSocket connected.");
     };
 
     ws.onmessage = (e) => {
-      output.textContent += "\n\n📥 From server:\n" + e.data;
+	pr(e.data);
     };
 
     ws.onerror = (e) => {
-      output.textContent += "\n\n❌ WebSocket error!";
+      pr("❌ WebSocket error!");
       console.error(e);
     };
 
@@ -119,17 +126,15 @@
       try {
         const parsed = JSON.parse(json); // validate JSON
         ws.send(JSON.stringify(parsed));
-        output.textContent += "\n\n📤 Sent:\n" + JSON.stringify(parsed, null, 2);
+        pr("📤 Sent:\n" + JSON.stringify(parsed, null, 2));
       } catch (e) {
-        output.textContent += "\n\n⚠️ Invalid JSON:\n" + e.message;
+        pr("\n\n⚠️ Invalid JSON:\n" + e.message);
       }
 
     }
 
     function place(event) { textarea.value = (event || window.event).target.getAttribute("data"); sendMessage(); }
 
-    function clean() { output.textContent = ""; }
-
   </script>
 
 </body>

scripts/TEST_HTTP_API.sh

@@ -12,14 +12,14 @@ ZP="00000000-0000-0000-0000-000000000001/TESTS"
 
 put "00000000-0000-0000-0000-000000000001/TESTS" "Value"
 
-exit
+#exit
     delete "00000000-0000-0000-0000-000000000001/TESTS"
 put "00000000-0000-0000-0000-000000000001/TESTS" "Value"
     delete "00000000-0000-0000-0000-000000000001/TESTS" "If-Match: *"
 put "00000000-0000-0000-0000-000000000001/TESTS" "Value"
     delete "00000000-0000-0000-0000-000000000001/TESTS" "If-Match: dd358c74cb9cb897424838fbcb69c933"
 
-exit
+#exit
 
     put "00000000-0000-0000-0000-000000000001/TESTS" "Value" "HULY-TTL: 2"
     put "00000000-0000-0000-0000-000000000001/TESTS/1" "Value_1" "HULY-TTL: 2"
@@ -27,7 +27,7 @@ exit
     put "00000000-0000-0000-0000-000000000001/TESTS/2/$/secret" "Value_secret" "HULY-TTL: 2"
     get "00000000-0000-0000-0000-000000000001/TESTS/"
 
-exit
+#exit
 
     delete "0000000/TESTS"
     delete ${ZP}

src/handlers_ws.rs

@@ -13,6 +13,9 @@
 // limitations under the License.
 //
 
+use std::sync::Arc;
+
+use tokio::sync::RwLock;
 use actix::{Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut};
 use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse, web};
 use actix_web_actors::ws;
@@ -20,12 +23,11 @@ use redis::aio::MultiplexedConnection;
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 
-use crate::redis::{
-    SaveMode, Ttl, deprecated_symbol, redis_delete, redis_list, redis_read, redis_save,
-};
-
-use crate::hub_service::{HubServiceHandle, ServerMessage, SessionId, new_session_id};
+use crate::redis::{ SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save };
+use crate::hub_service::{ServerMessage, SessionId, new_session_id};
 use crate::workspace_owner::check_workspace_core;
+use crate::hub_service::HubState;
+
 
 #[derive(Serialize, Default)]
 struct ReturnBase<'a> {
@@ -122,11 +124,9 @@ pub enum WsCommand {
 use hulyrs::services::jwt::Claims;
 
 /// Session condition
-#[allow(dead_code)]
 pub struct WsSession {
     pub redis: MultiplexedConnection,
     pub id: SessionId,
-    pub hub: HubServiceHandle,
     hub_state: Arc<RwLock<HubState>>,
     pub claims: Claims,
 }
@@ -136,18 +136,24 @@ impl Actor for WsSession {
     type Context = ws::WebsocketContext<Self>;
 
     fn started(&mut self, ctx: &mut Self::Context) {
-        let addr = ctx.address();
-        let recipient = addr.recipient::<ServerMessage>();
-
-        self.hub.connect(self.id, recipient);
-        tracing::info!("WebSocket connected: {}", self.id);
+        let id = self.id;
+        let recipient = ctx.address().recipient::<ServerMessage>();
+        let hub_state = self.hub_state.clone();
+        ctx.spawn(
+            actix::fut::wrap_future(async move {
+                hub_state.write().await.connect(id, recipient);
+            }).map(|_, _, _| ())
+        );
+        tracing::info!("WebSocket connected: {}", id);
     }
 
     fn stopped(&mut self, _ctx: &mut Self::Context) {
-        if self.id != 0 {
-            self.hub.disconnect(self.id);
-        }
-        tracing::info!("WebSocket disconnected: {:?}", &self.id);
+        let id = self.id;
+        let hub_state = self.hub_state.clone();
+        actix::spawn(async move {
+            hub_state.write().await.disconnect(id);
+        });
+        tracing::info!("WebSocket disconnected: {}", id);
     }
 }
 
@@ -178,15 +184,8 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
     }
 }
 
-/// All logic in one impl
+/// All logic
 impl WsSession {
-    fn ws_error(&self, ctx: &mut ws::WebsocketContext<Self>, msg: &str) {
-        ctx.text(format!(r#"{{"error":"{}"}}"#, msg));
-    }
-
-    fn workspace_check_ws(&self, key: &str) -> Result<(), &'static str> {
-        check_workspace_core(&self.claims, key)
-    }
 
     fn fut_send(
         &mut self,
@@ -215,6 +214,8 @@ impl WsSession {
 
     /// When valid JSON recieved for WsSession
     fn handle_command(&mut self, cmd: WsCommand, ctx: &mut ws::WebsocketContext<Self>) {
+        
+        // PUT
         match cmd {
             WsCommand::Put {
                 key,
@@ -227,14 +228,6 @@ impl WsSession {
             } => {
                 tracing::info!("PUT {} = {}", &key, &data); //  (expires_at: {:?}) (ttl: {:?}) correlation: {:?} &expires_at, &ttl, &correlation
 
-                // Check workspace
-                if let Err(e) = self.workspace_check_ws(&key) {
-                    self.ws_error(ctx, e);
-                    return;
-                }
-
-                let mut redis = self.redis.clone();
-
                 let base = serde_json::json!(ReturnBase {
                     action: "put",
                     // key: Some(key.as_str()),
@@ -247,7 +240,13 @@ impl WsSession {
                     ..Default::default()
                 });
 
+                let mut redis = self.redis.clone();
+                let claims = self.claims.clone();
+
                 let fut = async move {
+                    // Check workspace
+                    if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+
                     // TTL logic
                     let real_ttl = if let Some(secs) = ttl {
                         Some(Ttl::Sec(secs as usize))
@@ -294,13 +293,8 @@ impl WsSession {
             } => {
                 tracing::info!("DELETE {}", &key); //  correlation:{:?} , &correlation
 
-                // Check workspace
-                if let Err(e) = self.workspace_check_ws(&key) {
-                    self.ws_error(ctx, e);
-                    return;
-                }
-
                 let mut redis = self.redis.clone();
+                let claims = self.claims.clone();
 
                 let base = serde_json::json!(ReturnBase {
                     action: "delete",
@@ -311,6 +305,9 @@ impl WsSession {
                 });
 
                 let fut = async move {
+                    // Check workspace
+                    if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+
                     // MODE logic
                     let mut mode = Some(SaveMode::Upsert);
                     if let Some(s) = if_match {
@@ -340,14 +337,6 @@ impl WsSession {
             WsCommand::Get { key, correlation } => {
                 tracing::info!("GET {}", &key); //  correlation:{:?} , &correlation
 
-                // Check workspace
-                if let Err(e) = self.workspace_check_ws(&key) {
-                    self.ws_error(ctx, e);
-                    return;
-                }
-
-                let mut redis = self.redis.clone();
-
                 let base = serde_json::json!(ReturnBase {
                     action: "get",
                     // key: Some(key.as_str()),
@@ -355,7 +344,13 @@ impl WsSession {
                     ..Default::default()
                 });
 
+                let mut redis = self.redis.clone();
+                let claims = self.claims.clone();
+
                 let fut = async move {
+                    // Check workspace
+                    if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+
                     let data_opt = redis_read(&mut redis, &key)
                         .await
                         .map_err(|e| e.to_string())?;
@@ -376,14 +371,6 @@ impl WsSession {
             WsCommand::List { key, correlation } => {
                 tracing::info!("LIST {:?}", &key); //  correlation: {:?} , &correlation
 
-                // Check workspace
-                if let Err(e) = self.workspace_check_ws(&key) {
-                    self.ws_error(ctx, e);
-                    return;
-                }
-
-                let mut redis = self.redis.clone();
-
                 let base = serde_json::json!(ReturnBase {
                     action: "list",
                     // key: Some(key.as_str()),
@@ -391,7 +378,13 @@ impl WsSession {
                     ..Default::default()
                 });
 
+                let mut redis = self.redis.clone();
+                let claims = self.claims.clone();
+
                 let fut = async move {
+                    // Check workspace
+                    if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+
                     let data = redis_list(&mut redis, &key)
                         .await
                         .map_err(|e| e.to_string())?;
@@ -401,100 +394,60 @@ impl WsSession {
                 self.fut_send(ctx, fut, base);
             }
 
-            /*
             WsCommand::Sub { key, correlation } => {
                 // LEVENT 3
                 tracing::info!("SUB {}", &key); //  correlation: {:?} , &correlation
 
-                // Check workspace
-                if let Err(e) = self.workspace_check_ws(&key) {
-                    self.ws_error(ctx, e);
-                    return;
-                }
-
-                let mut obj = serde_json::json!(ReturnBase {
-                    action: "sub",
-                    // key: Some(key.as_str()),
-                    correlation: correlation.as_deref(),
-                    ..Default::default()
+                let base = serde_json::json!(ReturnBase {
+                     action: "sub",
+                     // key: Some(key.as_str()),
+                     correlation: correlation.as_deref(),
+                     ..Default::default()
                 });
 
-                let map = obj.as_object_mut().unwrap();
+                let hub_state = self.hub_state.clone();
+                let id = self.id.clone();
+                let claims = self.claims.clone();
 
-                if deprecated_symbol(&key) {
-                    map.insert("error".into(), json!("Deprecated symbol in key"));
-                } else {
-                    self.hub.subscribe(self.id, key.clone());
-                    map.insert("result".into(), json!("OK"));
-                }
-                ctx.text(obj.to_string());
-            }*/
-            WsCommand::Sub { key, correlation } => {
-                // LEVENT 3
-                tracing::info!("SUB {}", &key); //  correlation: {:?} , &correlation
+                let fut = async move {
+                    // Check workspace
+                    if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
 
-                // Check workspace
-                if let Err(e) = self.workspace_check_ws(&key) {
-                    self.ws_error(ctx, e);
-                    return;
-                }
+                    hub_state.write().await.subscribe(id, key);
+                    Ok(json!({ "result": "OK" }))
+                };
 
-                let mut obj = serde_json::json!(ReturnBase {
-                    action: "sub",
-                    // key: Some(key.as_str()),
-                    correlation: correlation.as_deref(),
-                    ..Default::default()
-                });
-
-                let map = obj.as_object_mut().unwrap();
-
-                if deprecated_symbol(&key) {
-                    map.insert("error".into(), json!("Deprecated symbol in key"));
-                } else {
-                    let fut = async move {
-                        let mut hub_state = self.hub_state.write().await;
-
-                        hub_state.subscribe(self.id, key.clone());
-                    };
-
-                    // spawn and respond when done
-                    //ctx.spawn(fut);
-
-                    map.insert("result".into(), json!("OK"));
-                }
-                ctx.text(obj.to_string());
+                self.fut_send(ctx, fut, base);
             }
 
             WsCommand::Unsub { key, correlation } => {
                 // LEVENT 4
                 tracing::info!("UNSUB {}", &key); //  correlation: {:?} , &correlation
 
-                let mut obj = serde_json::json!(ReturnBase {
+                let base = serde_json::json!(ReturnBase {
                     action: "unsub",
                     // key: Some(key.as_str()),
                     correlation: correlation.as_deref(),
                     ..Default::default()
                 });
 
-                let map = obj.as_object_mut().unwrap();
+                let hub_state = self.hub_state.clone();
+                let id = self.id.clone();
+                let claims = self.claims.clone();
 
-                if key == "*" {
-                    self.hub.unsubscribe_all(self.id);
-                    map.insert("result".into(), json!("OK"));
-                } else {
-                    // Check workspace
-                    if let Err(e) = self.workspace_check_ws(&key) {
-                        self.ws_error(ctx, e);
-                        return;
-                    }
-                    if deprecated_symbol(&key) {
-                        map.insert("error".into(), json!("Deprecated symbol in key"));
+                let fut = async move {
+                    if key == "*" {
+                        hub_state.write().await.unsubscribe_all(id);
+                        Ok(json!({ "result": "OK" }))
                     } else {
-                        self.hub.unsubscribe(self.id, key.clone());
-                        map.insert("result".into(), json!("OK"));
+                        // Check workspace
+                        if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); }
+
+                        hub_state.write().await.unsubscribe(id, key);
+                        Ok(json!({ "result": "OK" }))
                     }
-                }
-                ctx.text(obj.to_string());
+                };
+                self.fut_send(ctx, fut, base);
             }
 
             WsCommand::Sublist { correlation } => {
@@ -506,39 +459,24 @@ impl WsSession {
                     ..Default::default()
                 });
 
-                let hub = self.hub.clone();
-                let id = self.id;
+                let hub_state = self.hub_state.clone();
+                let id = self.id.clone();
 
                 let fut = async move {
-                    let hub_state = self.hub_state.read().await;
-
-                    //hub_state.subscribe(self.id, key.clone());
-
-                    //
+                    let keys = hub_state.read().await.subscribe_list(id);
+                    Ok(json!({ "result": keys }))
                 };
-
-                self.fut_send(
-                    ctx,
-                    async move {
-                        let keys = hub.subscribe_list(id).await;
-                        Ok(json!({ "result": keys }))
-                    },
-                    base,
-                );
-            } // End of commands
+                self.fut_send(ctx, fut, base);
+            }
+            // End of commands
         }
     }
 }
 
-use crate::hub_service::HubState;
-use std::sync::Arc;
-use tokio::sync::RwLock;
-
 pub async fn handler(
     req: HttpRequest,
     payload: web::Payload,
     redis: web::Data<MultiplexedConnection>,
-    hub: web::Data<HubServiceHandle>, // <-- было Addr<WsHub>
     hub_state: web::Data<Arc<RwLock<HubState>>>,
 ) -> Result<HttpResponse, Error> {
     let claims = req
@@ -549,7 +487,7 @@ pub async fn handler(
 
     let session = WsSession {
         redis: redis.get_ref().clone(),
-        hub: hub.get_ref().clone(),
+        // hub: hub.get_ref().clone(),
         hub_state: hub_state.get_ref().clone(),
         id: new_session_id(),
         claims,

src/hub_service.rs

@@ -1,11 +1,27 @@
+//
+// 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};
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
 
 use actix::prelude::*;
 
 use redis::aio::MultiplexedConnection;
 use serde::Serialize;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::{RwLock};
 
 fn subscription_matches(sub_key: &str, key: &str) -> bool {
     if sub_key == key {
@@ -49,45 +65,11 @@ pub enum RedisEventAction {
 #[derive(Debug, Clone, Serialize)]
 pub struct RedisEvent {
     //    pub db: u32,
+    pub message: RedisEventAction,
     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  ====
+// ==== Handle ====
 
 #[derive(Debug, Default)]
 pub struct HubState {
@@ -96,184 +78,93 @@ pub struct HubState {
 }
 
 impl HubState {
+    pub fn connect(&mut self, session_id: SessionId, addr: Recipient<ServerMessage>) {
+        self.sessions.insert(session_id, addr);
+    }
+    pub fn disconnect(&mut self, session_id: SessionId) {
+        self.sessions.remove(&session_id);
+        self.subs.retain(|_, ids| { ids.remove(&session_id); !ids.is_empty() });
+    }
     pub fn subscribe(&mut self, session_id: SessionId, key: String) {
         self.subs.entry(key).or_default().insert(session_id);
     }
-}
-
-#[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());
-                        }
-                    }
-                }
+    pub fn unsubscribe(&mut self, session_id: SessionId, key: String) {
+        if let Some(set) = self.subs.get_mut(&key) {
+            set.remove(&session_id);
+            if set.is_empty() {
+                self.subs.remove(&key);
             }
+        }
+    }
+    pub fn unsubscribe_all(&mut self, session_id: SessionId) {
+        self.subs.retain(|_, ids| {
+            ids.remove(&session_id);
+            !ids.is_empty()
         });
-
-        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,
+    pub fn subscribe_list(&self, session_id: SessionId) -> Vec<String> {
+        self.subs   
+            .iter()
+            .filter_map(|(key, ids)| {
+                if ids.contains(&session_id) {
+                    Some(key.clone())
+                } else {
+                    None
+                }
             })
-            .await;
-        rx.await.unwrap_or_default()
+            .collect()
+    }
+    pub fn count(&self) -> usize {
+        self.sessions.len()
+    }
+    pub fn recipients_for_key(&self, key: &str) -> Vec<Recipient<ServerMessage>> {
+        let mut out = Vec::new();
+        for (sub_key, set) in &self.subs {
+            if subscription_matches(sub_key, key) {
+                for sid in set {
+                    if let Some(r) = self.sessions.get(sid) {
+                        out.push(r.clone());
+                    }
+                }
+            }
+        }
+        out
     }
 
-    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()
+}
+
+
+
+// Send messages about new Redis events
+pub async fn push_event(
+    hub_state: &Arc<RwLock<HubState>>,
+    redis: &mut MultiplexedConnection,
+    ev: RedisEvent,
+) {
+    // Collect Addresses
+    let recipients: Vec<Recipient<ServerMessage>> = {
+        hub_state.read().await.recipients_for_key(&ev.key)
+    };
+    if recipients.is_empty() {
+        return;
     }
 
-    // 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()
-    // }
+    // Get value from Redis (only for `Set` event, not for `Delete`, `Expire`)
+    let mut value: Option<String> = None;
+    if matches!(ev.message, RedisEventAction::Set) {
+        match redis::cmd("GET")
+            .arg(&ev.key)
+            .query_async::<Option<String>>(redis)
+            .await
+        {
+            Ok(v) => value = v,
+            Err(e) => tracing::warn!("redis GET {} failed: {}", &ev.key, e),
+        }
+    }
 
-    pub fn push_event(&self, ev: RedisEvent) {
-        let _ = self.tx.try_send(Command::RedisEvent(ev));
+    // Sending
+    let payload = ServerMessage { event: ev, value };
+    for rcpt in recipients {
+        let _ = rcpt.do_send(payload.clone());
     }
 }

src/main.rs

@@ -34,7 +34,7 @@ mod redis;
 mod workspace_owner;
 
 mod hub_service;
-use hub_service::{HubServiceHandle, HubState};
+use hub_service::{HubState};
 
 use config::CONFIG;
 
@@ -102,13 +102,10 @@ async fn main() -> anyhow::Result<()> {
     let redis_connection = redis_client.get_multiplexed_async_connection().await?;
 
     // starting HubService
-    let hub = HubServiceHandle::start(redis_connection.clone());
-
-    let hub_state = HubState::default();
-    let hub_state = Arc::new(RwLock::new(hub_state));
+    let hub_state = Arc::new(RwLock::new(HubState::default()));
 
     // starting Logger
-    tokio::spawn(redis::receiver(redis_client, hub.clone()));
+    tokio::spawn(redis::receiver(redis_client, hub_state.clone()));
 
     let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port);
 
@@ -144,27 +141,20 @@ 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(
-                "/status",
-                web::get().to(|hub: web::Data<HubServiceHandle>| async move {
-                    let count = hub.count().await;
-                    Ok::<_, actix_web::Error>(
-                        HttpResponse::Ok().json(json!({ "websockets": count, "status": "OK" })),
-                    )
-                }),
-            )
-
-        // .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))
-        // }))
+            .route("/status", web::get().to({
+                move |hub_state: web::Data<Arc<RwLock<HubState>>>| {
+                    let hub_state = hub_state.clone();
+                    async move {
+                        let count = hub_state.read().await.count();
+                        Ok::<_, actix_web::Error>(
+                            HttpResponse::Ok().json(json!({ "websockets": count, "status": "OK" })),
+                        )
+                    }
+                }
+            }))
     })
     .bind(socket)?
     .run();

src/redis.rs

@@ -13,15 +13,16 @@
 // limitations under the License.
 //
 
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::{sync::Arc, time::{SystemTime, UNIX_EPOCH}};
 
 use ::redis::Msg;
+use tokio::sync::RwLock;
 use tokio_stream::StreamExt;
 use tracing::*;
 
 use crate::{
-    config::{CONFIG, RedisMode},
-    hub_service::{HubServiceHandle, RedisEvent, RedisEventAction},
+    config::{RedisMode, CONFIG},
+    hub_service::{push_event, HubState, RedisEvent, RedisEventAction},
 };
 
 #[derive(serde::Serialize)]
@@ -336,7 +337,7 @@ impl TryFrom<Msg> for RedisEvent {
 
         // "__keyevent@0__:set" → event="set", db=0; payload = key
         let event = channel.rsplit(':').next().unwrap_or("");
-        let action = match event {
+        let message = match event {
             "set" => RedisEventAction::Set,
             "del" => RedisEventAction::Del,
             "unlink" => RedisEventAction::Unlink,
@@ -354,12 +355,15 @@ impl TryFrom<Msg> for RedisEvent {
         Ok(RedisEvent {
             // db,
             key: payload.clone(),
-            action,
+            message,
         })
     }
 }
 
-pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Result<()> {
+pub async fn receiver(redis_client: Client,
+    // hub: HubServiceHandle
+    hub_state: Arc<RwLock<HubState>>,
+) -> anyhow::Result<()> {
     let mut redis = redis_client.get_multiplexed_async_connection().await?;
     let mut pubsub = redis_client.get_async_pubsub().await?;
 
@@ -386,7 +390,8 @@ pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Re
             Ok(ev) => {
                 // debug!("redis event: {ev:#?}");
 
-                hub.push_event(ev);
+                push_event(&hub_state, &mut redis, ev).await;
+
             }
             Err(e) => {
                 warn!("invalid redis message: {e}");

src/workspace_owner.rs

@@ -16,8 +16,15 @@
 use hulyrs::services::jwt::Claims;
 use uuid::Uuid;
 
+use crate::redis::deprecated_symbol;
+
 // common checker
 pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> {
+
+    if deprecated_symbol(key) {
+        return Err("Invalid key: deprecated symbols");
+    }
+
     if claims.is_system() {
         return Ok(());
     }