features lopt: direct personal messages between websockets by username

Signed-off-by: Leonid Kaganov <lleo@lleo.me>

Cargo.lock

@@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
 
 [[package]]
 name = "hulypulse"
-version = "0.3.7"
+version = "0.4.0"
 dependencies = [
  "actix-cors",
  "actix-web",

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"]

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<RwLock<HubState>>,
     #[cfg(feature = "auth")] claims: Option<Claims>,
     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;
                         }

src/hub_service.rs

@@ -75,6 +75,11 @@ pub struct HubState {
     heartbeats: HashMap<SessionId, std::time::Instant>,
     serverping: HashMap<SessionId, std::time::Instant>,
     abort_handles: HashMap<SessionId, AbortHandle>,
+    // client_ids: HashMap<SessionId, String>,
+    #[cfg(feature = "lopt")]
+    name_by_session: HashMap<SessionId, String>,
+    #[cfg(feature = "lopt")]
+    session_by_name: HashMap<String, SessionId>,
 }
 
 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<RwLock<HubState>>, 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<RwLock<HubState>>) {
     tokio::spawn(async move {
         let mut ticker = tokio::time::interval(std::time::Duration::from_secs(2));

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",

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}");