From 382e4a8762f9cc4941977092c2fe6f47f57c30dc Mon Sep 17 00:00:00 2001 From: Reboot-Codes Date: Wed, 30 Apr 2025 13:07:23 -0700 Subject: [PATCH] Add a reply field --- src/server/listener.rs | 6 ++- src/server/models.rs | 3 ++ src/server/websockets.rs | 4 +- src/user/mod.rs | 97 ++++++++++++++++++++++++++-------------- 4 files changed, 74 insertions(+), 36 deletions(-) diff --git a/src/server/listener.rs b/src/server/listener.rs index 48484a6..4f4ccfc 100644 --- a/src/server/listener.rs +++ b/src/server/listener.rs @@ -444,7 +444,8 @@ pub async fn nexus_listener( author: "ipc://com.reboot-codes.nexus.listener".to_string(), kind, message: "api key removed from store".to_string(), - id: gen_message_id_with_check(&ipc_dispatch_store).await + id: gen_message_id_with_check(&ipc_dispatch_store).await, + replying_to: Some("".to_string()) }; ipc_dispatch_store.messages.lock().await.insert(generated_message.id.clone(), generated_message.clone().into()); @@ -508,7 +509,8 @@ pub async fn nexus_listener( author: author.clone(), kind: msg.kind.clone(), message: msg.message.clone(), - id: gen_message_id_with_check(&internal_ipc_store).await + id: gen_message_id_with_check(&internal_ipc_store).await, + replying_to: msg.replying_to.clone() }) { Ok(_) => {} Err(e) => { diff --git a/src/server/models.rs b/src/server/models.rs index 31c4ca7..308f8bf 100644 --- a/src/server/models.rs +++ b/src/server/models.rs @@ -40,6 +40,7 @@ pub struct IPCMessage { pub author: String, pub kind: String, pub message: String, + pub replying_to: Option, } #[derive(Debug, Deserialize, Serialize, Clone)] @@ -48,6 +49,7 @@ pub struct IPCMessageWithId { pub kind: String, pub message: String, pub id: String, + pub replying_to: Option, } impl Into for IPCMessageWithId { @@ -56,6 +58,7 @@ impl Into for IPCMessageWithId { author: self.author, kind: self.kind, message: self.message, + replying_to: self.replying_to, } } } diff --git a/src/server/websockets.rs b/src/server/websockets.rs index 8f4f89a..ab86780 100644 --- a/src/server/websockets.rs +++ b/src/server/websockets.rs @@ -49,6 +49,7 @@ pub struct WsIn { pub kind: String, pub message: String, pub api_key: Option, + pub replying_to: Option, } pub async fn handle_ws_client( @@ -175,7 +176,7 @@ pub async fn handle_ws_client( } if allowed_to_send { - let generated_message = IPCMessageWithId { id: message_id.clone(), author: format!("ws://{}?client={}", recv_api_key.user_id.clone(), ws_client.id.clone()), kind: msg.kind.clone(), message: msg.message.clone() }; + let generated_message = IPCMessageWithId { id: message_id.clone(), author: format!("ws://{}?client={}", recv_api_key.user_id.clone(), ws_client.id.clone()), kind: msg.kind.clone(), message: msg.message.clone(), replying_to: msg.replying_to.clone() }; recv_store.messages.lock().await.insert(message_id.clone(), generated_message.clone().into()); @@ -246,6 +247,7 @@ pub async fn handle_ws_client( author: msg.author.clone(), kind: msg.kind.clone(), message: msg.message.clone(), + replying_to: msg.replying_to.clone() }) .unwrap(); match sender.send(Message::text(response)).await { diff --git a/src/user/mod.rs b/src/user/mod.rs index 670491e..b7dd5eb 100644 --- a/src/user/mod.rs +++ b/src/user/mod.rs @@ -1,9 +1,27 @@ +use crate::{ + arbiter::models::ApiKeyWithKey, + client::ClientStatus, + server::{ + models::IPCMessageWithId, + websockets::WsIn, + }, +}; use log::error; use regex::Regex; -use tokio::{sync::{broadcast, mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}, Mutex}, task::JoinHandle}; -use tokio_util::sync::CancellationToken; use std::sync::Arc; -use crate::{arbiter::models::ApiKeyWithKey, client::ClientStatus, server::{models::IPCMessageWithId, websockets::WsIn}}; +use tokio::{ + sync::{ + Mutex, + broadcast, + mpsc::{ + UnboundedReceiver, + UnboundedSender, + unbounded_channel, + }, + }, + task::JoinHandle, +}; +use tokio_util::sync::CancellationToken; /// Use in a thread to connect to send messages to/recieve messages from nexus with. /// Requires a [NexusClient](nexus::client::NexusClient) or a tokio IPC channel @@ -17,7 +35,7 @@ pub struct NexusUser { pub cancellation_token: CancellationToken, api_key: ApiKeyWithKey, to_client: broadcast::Sender, - from_server_tx: broadcast::Sender + from_server_tx: broadcast::Sender, } impl NexusUser { @@ -27,7 +45,7 @@ impl NexusUser { cancellation_token: CancellationToken, api_key: ApiKeyWithKey, to_client: broadcast::Sender, - from_server_tx: broadcast::Sender + from_server_tx: broadcast::Sender, ) -> Self { NexusUser { is_child, @@ -35,22 +53,24 @@ impl NexusUser { cancellation_token, api_key, to_client, - from_server_tx + from_server_tx, } } - pub fn send(&self, kind: &String, message: &String) -> Result<(), anyhow::Error> { + pub fn send( + &self, + kind: &String, + message: &String, + replying_to: &Option, + ) -> Result<(), anyhow::Error> { match self.to_client.send(WsIn { kind: kind.clone(), message: message.clone(), - api_key: Some(self.api_key.key.clone()) + api_key: Some(self.api_key.key.clone()), + replying_to: replying_to.clone(), }) { - Ok(_) => { - Ok(()) - }, - Err(e) => { - Err(e.into()) - } + Ok(_) => Ok(()), + Err(e) => Err(e.into()), } } @@ -63,30 +83,41 @@ impl NexusUser { ( rx, tokio::task::spawn(async move { - cancellation_token.run_until_cancelled(async move { - while let Ok(message) = from_client.recv().await { - for allowed_event_regex in &this.api_key.allowed_events_to.clone() { - match Regex::new(&allowed_event_regex) { - Ok(regex) => { - if regex.is_match(&message.kind.clone()) { - match tx.send(message.clone()) { - Ok(_) => {}, - Err(e) => { - error!("Failed to send message to user: {}, due to:\n{}", this.api_key.user_id.clone(), e); - } - }; + cancellation_token + .run_until_cancelled(async move { + while let Ok(message) = from_client.recv().await { + for allowed_event_regex in &this.api_key.allowed_events_to.clone() { + match Regex::new(&allowed_event_regex) { + Ok(regex) => { + if regex.is_match(&message.kind.clone()) { + match tx.send(message.clone()) { + Ok(_) => {} + Err(e) => { + error!( + "Failed to send message to user: {}, due to:\n{}", + this.api_key.user_id.clone(), + e + ); + } + }; - break; + break; + } + } + Err(e) => { + error!( + "Allowed event regular expression: \"{}\", for user id: {}, errored with: {}", + allowed_event_regex.clone(), + this.api_key.user_id.clone(), + e + ); } - }, - Err(e) => { - error!("Allowed event regular expression: \"{}\", for user id: {}, errored with: {}", allowed_event_regex.clone(), this.api_key.user_id.clone(), e); } } } - } - }).await; - }) + }) + .await; + }), ) } } -- 2.51.2