diff --git a/Cargo.lock b/Cargo.lock index 5ecd2e434c..1668acc159 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.3.7" +version = "0.4.0" dependencies = [ "actix-cors", "actix-web", diff --git a/Cargo.toml b/Cargo.toml index f80f449ad2..0cdcc1233a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "hulypulse" -version = "0.3.7" +version = "0.4.0" edition = "2024" [dependencies] @@ -43,6 +43,7 @@ tokio-tungstenite = { version = "0.21", default-features = false, features = [ ] } [features] -default = ["db-redis","auth"] +default = ["db-redis","auth"] # lopt auth = ["regorus", "uuid", "hulyrs", "secrecy"] +lopt = [] db-redis = ["redis"] \ No newline at end of file diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 05bc8015fc..b836f7254d 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -33,6 +33,9 @@ use crate::{ hub_service::{HubState, SessionId, new_session_id}, }; +#[cfg(feature = "lopt")] +use crate::hub_service::send_to_name; + #[cfg(feature = "auth")] use crate::workspace_owner::check_workspace_core; @@ -44,6 +47,20 @@ use strum::AsRefStr; #[derive(Deserialize, Debug, AsRefStr)] #[serde(rename_all = "lowercase", tag = "type")] pub enum WsCommand { + #[cfg(feature = "lopt")] + Personal { + to: String, + correlation: String, + data: String, + }, + + #[cfg(feature = "lopt")] + Answer { + to: String, + correlation: String, + data: String, + }, + Put { #[serde(default = "default_corr")] correlation: String, @@ -156,8 +173,39 @@ async fn handle_command( hub_state: &Arc>, #[cfg(feature = "auth")] claims: Option, session_id: SessionId, + #[cfg(feature = "lopt")] client_name: &str, ) { match cmd { + #[cfg(feature = "lopt")] + WsCommand::Personal { + to, + correlation, + data, + } => { + use crate::hub_service::send_to_name; + + tracing::debug!("PERSONAL from {} to {}", &client_name, &to); + let payload = + json!({ "personal": client_name, "correlation": correlation, "data": data }); + if !send_to_name(&hub_state, &to, payload).await { + tracing::debug!("PERSONAL send from [{}] to [{}] failed", &client_name, &to); + result_err("failed", &correlation, ws).await; + } + } + + #[cfg(feature = "lopt")] + WsCommand::Answer { + to, + correlation, + data, + } => { + tracing::debug!("ANSWER from {} to {}", &client_name, &to); + let payload = json!({ "correlation": correlation, "data": data }); + if !send_to_name(&hub_state, &to, payload).await { + tracing::debug!("PERSONAL send_to failed: no such session {}", to); + } + } + // INFO WsCommand::Info { correlation } => { tracing::debug!("INFO"); @@ -333,16 +381,26 @@ pub async fn handler( .to_owned(), ); + #[cfg(feature = "lopt")] + let client_name = req + .match_info() + .get("client_name") + .unwrap_or("") + .to_string(); + let (response, mut session, mut msg_stream) = actix_ws::handle(&req, payload)?; let session_id = new_session_id(); let (abort_handle, abort_reg) = AbortHandle::new_pair(); - hub_state - .write() - .await - .connect(session_id, session.clone(), abort_handle); + hub_state.write().await.connect( + session_id, + session.clone(), + abort_handle, + #[cfg(feature = "lopt")] + client_name.clone(), + ); tracing::debug!("WebSocket connected: {}", session_id); actix_web::rt::spawn(Abortable::new( @@ -385,6 +443,7 @@ pub async fn handler( | WsCommand::List { key, .. } | WsCommand::Sub { key, .. } | WsCommand::Unsub { key, .. } => key.as_str(), + // | WsCommand::Personal { key, .. } => key.as_str(), _ => "", }; @@ -404,6 +463,8 @@ pub async fn handler( #[cfg(feature = "auth")] claims.clone(), session_id, + #[cfg(feature = "lopt")] + &client_name, ) .await; } diff --git a/src/hub_service.rs b/src/hub_service.rs index e716634881..137bdf5013 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -75,6 +75,11 @@ pub struct HubState { heartbeats: HashMap, serverping: HashMap, abort_handles: HashMap, + // client_ids: HashMap, + #[cfg(feature = "lopt")] + name_by_session: HashMap, + #[cfg(feature = "lopt")] + session_by_name: HashMap, } use futures::future::AbortHandle; @@ -93,6 +98,7 @@ impl HubState { session_id: SessionId, session: actix_ws::Session, abort_handle: AbortHandle, + #[cfg(feature = "lopt")] client_name: String, ) { self.sessions.insert(session_id, session); self.heartbeats @@ -100,6 +106,11 @@ impl HubState { self.serverping .insert(session_id, std::time::Instant::now()); self.abort_handles.insert(session_id, abort_handle); + + #[cfg(feature = "lopt")] + self.name_by_session.insert(session_id, client_name.clone()); + #[cfg(feature = "lopt")] + self.session_by_name.insert(client_name, session_id); } pub fn disconnect(&mut self, session_id: SessionId) { @@ -111,6 +122,12 @@ impl HubState { ids.remove(&session_id); !ids.is_empty() }); + + #[cfg(feature = "lopt")] + if let Some(client_id) = self.name_by_session.remove(&session_id) { + self.session_by_name.remove(&client_id); + } + tracing::debug!( "hub.disconnected {}, all: {}", session_id, @@ -201,6 +218,23 @@ pub async fn broadcast_event( } } +#[cfg(feature = "lopt")] +pub async fn send_to_name(hub_state: &Arc>, to: &str, payload: Value) -> bool { + let hub = hub_state.read().await; + + let to_sid = if let Some(&sid) = hub.session_by_name.get(to) { + sid + } else { + return false; + }; + + let Some(mut session) = hub.sessions.get(&to_sid).cloned() else { + return false; + }; + + session.text(payload.to_string()).await.is_ok() +} + pub fn check_heartbeat(hub_state: Arc>) { tokio::spawn(async move { let mut ticker = tokio::time::interval(std::time::Duration::from_secs(2)); diff --git a/src/main.rs b/src/main.rs index 596c751fcf..7221b2c9f9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -240,6 +240,7 @@ async fn main() -> anyhow::Result<()> { .wrap(middleware::Logger::default()) .wrap(cors) .service(api_scope) + .route("/ws/{client_name}", web::get().to(handlers_ws::handler)) .route("/ws", ws_route) .route( "/status", diff --git a/src/redis.rs b/src/redis.rs index 78c5775edd..37ea4bcdbd 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -433,7 +433,7 @@ pub async fn receiver( while let Some(message) = messages.next().await { match RedisEvent::try_from(message) { Ok(ev) => { - push_event(&hub_state, &mut redis, ev).await; + push_event(&hub_state, &mut redis, ev); // .await; } Err(e) => { warn!("invalid redis message: {e}");