diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 93410f1bb3..e140f5e03e 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -127,6 +127,7 @@ pub struct WsSession { pub redis: MultiplexedConnection, pub id: SessionId, pub hub: HubServiceHandle, + hub_state: Arc>, 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, hub: web::Data, // <-- было Addr + hub_state: web::Data>>, ) -> Result { 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, }; diff --git a/src/hub_service.rs b/src/hub_service.rs index 859f5d5f2a..58975b7458 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -89,6 +89,18 @@ enum Command { // ==== Handle ==== +#[derive(Debug, Default)] +pub struct HubState { + sessions: HashMap>, + subs: HashMap>, +} + +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, diff --git a/src/main.rs b/src/main.rs index d524e68e85..de3d240fb1 100644 --- a/src/main.rs +++ b/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(