use global state for Hub

Signed-off-by: Alexey Aristov <aav@acm.org>

src/handlers_ws.rs

@@ -127,6 +127,7 @@ pub struct WsSession {
     pub redis: MultiplexedConnection,
     pub id: SessionId,
     pub hub: HubServiceHandle,
+    hub_state: Arc<RwLock<HubState>>,
     pub claims: Claims,
 }
 
@@ -400,6 +401,7 @@ impl WsSession {
                 self.fut_send(ctx, fut, base);
             }
 
+            /*
             WsCommand::Sub { key, correlation } => {
                 // LEVENT 3
                 tracing::info!("SUB {}", &key); //  correlation: {:?} , &correlation
@@ -426,6 +428,41 @@ impl WsSession {
                     map.insert("result".into(), json!("OK"));
                 }
                 ctx.text(obj.to_string());
+            }*/
+            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 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());
             }
 
             WsCommand::Unsub { key, correlation } => {
@@ -472,6 +509,14 @@ impl WsSession {
                 let hub = self.hub.clone();
                 let id = self.id;
 
+                let fut = async move {
+                    let hub_state = self.hub_state.read().await;
+
+                    //hub_state.subscribe(self.id, key.clone());
+
+                    //
+                };
+
                 self.fut_send(
                     ctx,
                     async move {
@@ -485,11 +530,16 @@ impl WsSession {
     }
 }
 
+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
         .extensions()
@@ -500,6 +550,7 @@ pub async fn handler(
     let session = WsSession {
         redis: redis.get_ref().clone(),
         hub: hub.get_ref().clone(),
+        hub_state: hub_state.get_ref().clone(),
         id: new_session_id(),
         claims,
     };

src/hub_service.rs

@@ -89,6 +89,18 @@ enum Command {
 
 // ==== Handle  ====
 
+#[derive(Debug, Default)]
+pub struct HubState {
+    sessions: HashMap<SessionId, Recipient<ServerMessage>>,
+    subs: HashMap<String, HashSet<SessionId>>,
+}
+
+impl HubState {
+    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>,

src/main.rs

@@ -34,7 +34,7 @@ mod redis;
 mod workspace_owner;
 
 mod hub_service;
-use hub_service::HubServiceHandle;
+use hub_service::{HubServiceHandle, HubState};
 
 use config::CONFIG;
 
@@ -104,6 +104,9 @@ async fn main() -> anyhow::Result<()> {
     // starting HubService
     let hub = HubServiceHandle::start(redis_connection.clone());
 
+    let hub_state = HubState::default();
+    let hub_state = Arc::new(RwLock::new(hub_state));
+
     // starting Logger
     tokio::spawn(redis::receiver(redis_client, hub.clone()));
 
@@ -115,6 +118,9 @@ async fn main() -> anyhow::Result<()> {
     tracing::info!("WebSocket API: {}/ws", &url);
     tracing::info!("Status: {}/status", &url);
 
+    use std::sync::Arc;
+    use tokio::sync::RwLock;
+
     let server = HttpServer::new(move || {
         let cors = Cors::default()
             .allow_any_origin()
@@ -125,7 +131,8 @@ async fn main() -> anyhow::Result<()> {
 
         App::new()
             .app_data(web::Data::new(redis_connection.clone()))
-            .app_data(web::Data::new(hub.clone()))
+            //.app_data(web::Data::new(hub.clone()))
+            .app_data(web::Data::new(hub_state.clone()))
             .wrap(middleware::Logger::default())
             .wrap(cors)
             .service(