diff --git a/README.md b/README.md index 2385a333c9..7cd4cff564 100644 --- a/README.md +++ b/README.md @@ -178,11 +178,11 @@ Size of data is limited to some reasonable size ** Server to Client ** subscribed events: - - `{"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Set","value":"hello"}` + - `{"message":"Set","key":"00000000-0000-0000-0000-000000000001/foo/bar","value":"hello"}` - - `{"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Expired"}` + - `{"message":"Expired","key":"00000000-0000-0000-0000-000000000001/foo/bar"}` - - `{"key":"00000000-0000-0000-0000-000000000001/foo/bar","action":"Del"}` + - `{"message":"Del","key":"00000000-0000-0000-0000-000000000001/foo/bar"}` ## Running diff --git a/scripts/TEST.html b/scripts/TEST.html index 8ecf37a3c8..2649fbd0e3 100644 --- a/scripts/TEST.html +++ b/scripts/TEST.html @@ -63,7 +63,7 @@
- +

@@ -89,9 +89,15 @@

Waiting for server response...
diff --git a/scripts/TEST_HTTP_API.sh b/scripts/TEST_HTTP_API.sh index e0fc3918eb..04383691a1 100755 --- a/scripts/TEST_HTTP_API.sh +++ b/scripts/TEST_HTTP_API.sh @@ -12,14 +12,14 @@ ZP="00000000-0000-0000-0000-000000000001/TESTS" put "00000000-0000-0000-0000-000000000001/TESTS" "Value" -exit +#exit delete "00000000-0000-0000-0000-000000000001/TESTS" put "00000000-0000-0000-0000-000000000001/TESTS" "Value" delete "00000000-0000-0000-0000-000000000001/TESTS" "If-Match: *" put "00000000-0000-0000-0000-000000000001/TESTS" "Value" delete "00000000-0000-0000-0000-000000000001/TESTS" "If-Match: dd358c74cb9cb897424838fbcb69c933" -exit +#exit put "00000000-0000-0000-0000-000000000001/TESTS" "Value" "HULY-TTL: 2" put "00000000-0000-0000-0000-000000000001/TESTS/1" "Value_1" "HULY-TTL: 2" @@ -27,7 +27,7 @@ exit put "00000000-0000-0000-0000-000000000001/TESTS/2/$/secret" "Value_secret" "HULY-TTL: 2" get "00000000-0000-0000-0000-000000000001/TESTS/" -exit +#exit delete "0000000/TESTS" delete ${ZP} diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index e140f5e03e..97385c5e7f 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -13,6 +13,9 @@ // limitations under the License. // +use std::sync::Arc; + +use tokio::sync::RwLock; use actix::{Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut}; use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse, web}; use actix_web_actors::ws; @@ -20,12 +23,11 @@ use redis::aio::MultiplexedConnection; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; -use crate::redis::{ - SaveMode, Ttl, deprecated_symbol, redis_delete, redis_list, redis_read, redis_save, -}; - -use crate::hub_service::{HubServiceHandle, ServerMessage, SessionId, new_session_id}; +use crate::redis::{ SaveMode, Ttl, redis_delete, redis_list, redis_read, redis_save }; +use crate::hub_service::{ServerMessage, SessionId, new_session_id}; use crate::workspace_owner::check_workspace_core; +use crate::hub_service::HubState; + #[derive(Serialize, Default)] struct ReturnBase<'a> { @@ -122,11 +124,9 @@ pub enum WsCommand { use hulyrs::services::jwt::Claims; /// Session condition -#[allow(dead_code)] pub struct WsSession { pub redis: MultiplexedConnection, pub id: SessionId, - pub hub: HubServiceHandle, hub_state: Arc>, pub claims: Claims, } @@ -136,18 +136,24 @@ impl Actor for WsSession { type Context = ws::WebsocketContext; fn started(&mut self, ctx: &mut Self::Context) { - let addr = ctx.address(); - let recipient = addr.recipient::(); - - self.hub.connect(self.id, recipient); - tracing::info!("WebSocket connected: {}", self.id); + let id = self.id; + let recipient = ctx.address().recipient::(); + let hub_state = self.hub_state.clone(); + ctx.spawn( + actix::fut::wrap_future(async move { + hub_state.write().await.connect(id, recipient); + }).map(|_, _, _| ()) + ); + tracing::info!("WebSocket connected: {}", id); } fn stopped(&mut self, _ctx: &mut Self::Context) { - if self.id != 0 { - self.hub.disconnect(self.id); - } - tracing::info!("WebSocket disconnected: {:?}", &self.id); + let id = self.id; + let hub_state = self.hub_state.clone(); + actix::spawn(async move { + hub_state.write().await.disconnect(id); + }); + tracing::info!("WebSocket disconnected: {}", id); } } @@ -178,15 +184,8 @@ impl StreamHandler> for WsSession { } } -/// All logic in one impl +/// All logic impl WsSession { - fn ws_error(&self, ctx: &mut ws::WebsocketContext, msg: &str) { - ctx.text(format!(r#"{{"error":"{}"}}"#, msg)); - } - - fn workspace_check_ws(&self, key: &str) -> Result<(), &'static str> { - check_workspace_core(&self.claims, key) - } fn fut_send( &mut self, @@ -215,6 +214,8 @@ impl WsSession { /// When valid JSON recieved for WsSession fn handle_command(&mut self, cmd: WsCommand, ctx: &mut ws::WebsocketContext) { + + // PUT match cmd { WsCommand::Put { key, @@ -227,14 +228,6 @@ impl WsSession { } => { tracing::info!("PUT {} = {}", &key, &data); // (expires_at: {:?}) (ttl: {:?}) correlation: {:?} &expires_at, &ttl, &correlation - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { - self.ws_error(ctx, e); - return; - } - - let mut redis = self.redis.clone(); - let base = serde_json::json!(ReturnBase { action: "put", // key: Some(key.as_str()), @@ -247,7 +240,13 @@ impl WsSession { ..Default::default() }); + let mut redis = self.redis.clone(); + let claims = self.claims.clone(); + let fut = async move { + // Check workspace + if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); } + // TTL logic let real_ttl = if let Some(secs) = ttl { Some(Ttl::Sec(secs as usize)) @@ -294,13 +293,8 @@ impl WsSession { } => { tracing::info!("DELETE {}", &key); // correlation:{:?} , &correlation - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { - self.ws_error(ctx, e); - return; - } - let mut redis = self.redis.clone(); + let claims = self.claims.clone(); let base = serde_json::json!(ReturnBase { action: "delete", @@ -311,6 +305,9 @@ impl WsSession { }); let fut = async move { + // Check workspace + if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); } + // MODE logic let mut mode = Some(SaveMode::Upsert); if let Some(s) = if_match { @@ -340,14 +337,6 @@ impl WsSession { WsCommand::Get { key, correlation } => { tracing::info!("GET {}", &key); // correlation:{:?} , &correlation - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { - self.ws_error(ctx, e); - return; - } - - let mut redis = self.redis.clone(); - let base = serde_json::json!(ReturnBase { action: "get", // key: Some(key.as_str()), @@ -355,7 +344,13 @@ impl WsSession { ..Default::default() }); + let mut redis = self.redis.clone(); + let claims = self.claims.clone(); + let fut = async move { + // Check workspace + if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); } + let data_opt = redis_read(&mut redis, &key) .await .map_err(|e| e.to_string())?; @@ -376,14 +371,6 @@ impl WsSession { WsCommand::List { key, correlation } => { tracing::info!("LIST {:?}", &key); // correlation: {:?} , &correlation - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { - self.ws_error(ctx, e); - return; - } - - let mut redis = self.redis.clone(); - let base = serde_json::json!(ReturnBase { action: "list", // key: Some(key.as_str()), @@ -391,7 +378,13 @@ impl WsSession { ..Default::default() }); + let mut redis = self.redis.clone(); + let claims = self.claims.clone(); + let fut = async move { + // Check workspace + if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); } + let data = redis_list(&mut redis, &key) .await .map_err(|e| e.to_string())?; @@ -401,100 +394,60 @@ impl WsSession { self.fut_send(ctx, fut, base); } - /* 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 base = serde_json::json!(ReturnBase { + action: "sub", + // key: Some(key.as_str()), + correlation: correlation.as_deref(), + ..Default::default() }); - let map = obj.as_object_mut().unwrap(); + let hub_state = self.hub_state.clone(); + let id = self.id.clone(); + let claims = self.claims.clone(); - if deprecated_symbol(&key) { - map.insert("error".into(), json!("Deprecated symbol in key")); - } else { - self.hub.subscribe(self.id, key.clone()); - map.insert("result".into(), json!("OK")); - } - ctx.text(obj.to_string()); - }*/ - WsCommand::Sub { key, correlation } => { - // LEVENT 3 - tracing::info!("SUB {}", &key); // correlation: {:?} , &correlation + let fut = async move { + // Check workspace + if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); } - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { - self.ws_error(ctx, e); - return; - } + hub_state.write().await.subscribe(id, key); + Ok(json!({ "result": "OK" })) + }; - 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()); + self.fut_send(ctx, fut, base); } WsCommand::Unsub { key, correlation } => { // LEVENT 4 tracing::info!("UNSUB {}", &key); // correlation: {:?} , &correlation - let mut obj = serde_json::json!(ReturnBase { + let base = serde_json::json!(ReturnBase { action: "unsub", // key: Some(key.as_str()), correlation: correlation.as_deref(), ..Default::default() }); - let map = obj.as_object_mut().unwrap(); + let hub_state = self.hub_state.clone(); + let id = self.id.clone(); + let claims = self.claims.clone(); - if key == "*" { - self.hub.unsubscribe_all(self.id); - map.insert("result".into(), json!("OK")); - } else { - // Check workspace - if let Err(e) = self.workspace_check_ws(&key) { - self.ws_error(ctx, e); - return; - } - if deprecated_symbol(&key) { - map.insert("error".into(), json!("Deprecated symbol in key")); + let fut = async move { + if key == "*" { + hub_state.write().await.unsubscribe_all(id); + Ok(json!({ "result": "OK" })) } else { - self.hub.unsubscribe(self.id, key.clone()); - map.insert("result".into(), json!("OK")); + // Check workspace + if let Err(e) = check_workspace_core(&claims, &key) { return Err(e.into()); } + + hub_state.write().await.unsubscribe(id, key); + Ok(json!({ "result": "OK" })) } - } - ctx.text(obj.to_string()); + }; + self.fut_send(ctx, fut, base); } WsCommand::Sublist { correlation } => { @@ -506,39 +459,24 @@ impl WsSession { ..Default::default() }); - let hub = self.hub.clone(); - let id = self.id; + let hub_state = self.hub_state.clone(); + let id = self.id.clone(); let fut = async move { - let hub_state = self.hub_state.read().await; - - //hub_state.subscribe(self.id, key.clone()); - - // + let keys = hub_state.read().await.subscribe_list(id); + Ok(json!({ "result": keys })) }; - - self.fut_send( - ctx, - async move { - let keys = hub.subscribe_list(id).await; - Ok(json!({ "result": keys })) - }, - base, - ); - } // End of commands + self.fut_send(ctx, fut, base); + } + // End of commands } } } -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 @@ -549,7 +487,7 @@ pub async fn handler( let session = WsSession { redis: redis.get_ref().clone(), - hub: hub.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 58975b7458..dc07686484 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -1,11 +1,27 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; use actix::prelude::*; use redis::aio::MultiplexedConnection; use serde::Serialize; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{RwLock}; fn subscription_matches(sub_key: &str, key: &str) -> bool { if sub_key == key { @@ -49,45 +65,11 @@ pub enum RedisEventAction { #[derive(Debug, Clone, Serialize)] pub struct RedisEvent { // pub db: u32, + pub message: RedisEventAction, pub key: String, - pub action: RedisEventAction, } -// ==== Commands for worker ==== - -enum Command { - Connect { - session_id: SessionId, - addr: Recipient, - }, - Disconnect { - session_id: SessionId, - }, - Subscribe { - session_id: SessionId, - key: String, - }, - Unsubscribe { - session_id: SessionId, - key: String, - }, - UnsubscribeAll { - session_id: SessionId, - }, - SubscribeList { - session_id: SessionId, - reply: oneshot::Sender>, - }, - Count { - reply: oneshot::Sender, - }, - // DumpSubs { - // reply: oneshot::Sender>>, - // }, - RedisEvent(RedisEvent), -} - -// ==== Handle ==== +// ==== Handle ==== #[derive(Debug, Default)] pub struct HubState { @@ -96,184 +78,93 @@ pub struct HubState { } impl HubState { + pub fn connect(&mut self, session_id: SessionId, addr: Recipient) { + self.sessions.insert(session_id, addr); + } + pub fn disconnect(&mut self, session_id: SessionId) { + self.sessions.remove(&session_id); + self.subs.retain(|_, ids| { ids.remove(&session_id); !ids.is_empty() }); + } 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, -} - -impl HubServiceHandle { - pub fn start(redis: MultiplexedConnection) -> Self { - let (tx, mut rx) = mpsc::channel::(1024); - - // Владелец состояния живёт внутри задачи - tokio::spawn(async move { - let mut sessions: HashMap> = HashMap::new(); - let mut subs: HashMap> = HashMap::new(); - let mut redis_conn = redis; - - fn subscribers_for( - subs: &HashMap>, - key: &str, - ) -> HashSet { - let mut out = HashSet::::new(); - for (sub_key, set) in subs.iter() { - if subscription_matches(sub_key, key) { - out.extend(set.iter().copied()); - } - } - out - } - - while let Some(cmd) = rx.recv().await { - match cmd { - Command::Connect { session_id, addr } => { - sessions.insert(session_id, addr); - } - - Command::Disconnect { session_id } => { - subs.retain(|_, ids| { - ids.remove(&session_id); - !ids.is_empty() - }); - sessions.remove(&session_id); - } - - Command::Subscribe { session_id, key } => { - subs.entry(key).or_default().insert(session_id); - } - - Command::Unsubscribe { session_id, key } => { - if let Some(set) = subs.get_mut(&key) { - set.remove(&session_id); - if set.is_empty() { - subs.remove(&key); - } - } - } - - Command::UnsubscribeAll { session_id } => { - subs.retain(|_, ids| { - ids.remove(&session_id); - !ids.is_empty() - }); - } - - Command::SubscribeList { session_id, reply } => { - let list = subs - .iter() - .filter_map(|(key, ids)| { - if ids.contains(&session_id) { - Some(key.clone()) - } else { - None - } - }) - .collect::>(); - let _ = reply.send(list); - } - - Command::Count { reply } => { - let _ = reply.send(sessions.len()); - } - - // Command::DumpSubs { reply } => { - // let snapshot = subs - // .iter() - // .map(|(k, set)| (k.clone(), set.iter().copied().collect::>())) - // .collect::>(); - // let _ = reply.send(snapshot); - // } - Command::RedisEvent(event) => { - let targets = subscribers_for(&subs, &event.key); - if targets.is_empty() { - continue; - } - let recipients: Vec> = targets - .into_iter() - .filter_map(|sid| sessions.get(&sid).cloned()) - .collect(); - - // Inside: waiting GET - let need_get = matches!(event.action, RedisEventAction::Set); - let mut value: Option = None; - if need_get { - match redis::cmd("GET") - .arg(&event.key) - .query_async::>(&mut redis_conn) - .await - { - Ok(v) => value = v, - Err(e) => { - tracing::warn!("redis GET {} failed: {}", &event.key, e); - } - } - } - - let payload = ServerMessage { event, value }; - - for rcpt in recipients { - let _ = rcpt.do_send(payload.clone()); - } - } - } + pub fn unsubscribe(&mut self, session_id: SessionId, key: String) { + if let Some(set) = self.subs.get_mut(&key) { + set.remove(&session_id); + if set.is_empty() { + self.subs.remove(&key); } + } + } + pub fn unsubscribe_all(&mut self, session_id: SessionId) { + self.subs.retain(|_, ids| { + ids.remove(&session_id); + !ids.is_empty() }); - - Self { tx } } - - // ---- API, ничего не выполняет параллельно внутри worker'а ---- - - pub fn connect(&self, session_id: SessionId, addr: Recipient) { - let _ = self.tx.try_send(Command::Connect { session_id, addr }); - } - - pub fn disconnect(&self, session_id: SessionId) { - let _ = self.tx.try_send(Command::Disconnect { session_id }); - } - - pub fn subscribe(&self, session_id: SessionId, key: String) { - let _ = self.tx.try_send(Command::Subscribe { session_id, key }); - } - - pub fn unsubscribe(&self, session_id: SessionId, key: String) { - let _ = self.tx.try_send(Command::Unsubscribe { session_id, key }); - } - - pub fn unsubscribe_all(&self, session_id: SessionId) { - let _ = self.tx.try_send(Command::UnsubscribeAll { session_id }); - } - - pub async fn subscribe_list(&self, session_id: SessionId) -> Vec { - let (tx, rx) = oneshot::channel(); - let _ = self - .tx - .send(Command::SubscribeList { - session_id, - reply: tx, + pub fn subscribe_list(&self, session_id: SessionId) -> Vec { + self.subs + .iter() + .filter_map(|(key, ids)| { + if ids.contains(&session_id) { + Some(key.clone()) + } else { + None + } }) - .await; - rx.await.unwrap_or_default() + .collect() + } + pub fn count(&self) -> usize { + self.sessions.len() + } + pub fn recipients_for_key(&self, key: &str) -> Vec> { + let mut out = Vec::new(); + for (sub_key, set) in &self.subs { + if subscription_matches(sub_key, key) { + for sid in set { + if let Some(r) = self.sessions.get(sid) { + out.push(r.clone()); + } + } + } + } + out } - pub async fn count(&self) -> usize { - let (tx, rx) = oneshot::channel(); - let _ = self.tx.send(Command::Count { reply: tx }).await; - rx.await.unwrap_or_default() +} + + + +// Send messages about new Redis events +pub async fn push_event( + hub_state: &Arc>, + redis: &mut MultiplexedConnection, + ev: RedisEvent, +) { + // Collect Addresses + let recipients: Vec> = { + hub_state.read().await.recipients_for_key(&ev.key) + }; + if recipients.is_empty() { + return; } - // pub async fn dump_subs(&self) -> std::collections::HashMap> { - // let (tx, rx) = oneshot::channel(); - // let _ = self.tx.send(Command::DumpSubs { reply: tx }).await; - // rx.await.unwrap_or_default() - // } + // Get value from Redis (only for `Set` event, not for `Delete`, `Expire`) + let mut value: Option = None; + if matches!(ev.message, RedisEventAction::Set) { + match redis::cmd("GET") + .arg(&ev.key) + .query_async::>(redis) + .await + { + Ok(v) => value = v, + Err(e) => tracing::warn!("redis GET {} failed: {}", &ev.key, e), + } + } - pub fn push_event(&self, ev: RedisEvent) { - let _ = self.tx.try_send(Command::RedisEvent(ev)); + // Sending + let payload = ServerMessage { event: ev, value }; + for rcpt in recipients { + let _ = rcpt.do_send(payload.clone()); } } diff --git a/src/main.rs b/src/main.rs index de3d240fb1..6a1ed21a39 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, HubState}; +use hub_service::{HubState}; use config::CONFIG; @@ -102,13 +102,10 @@ async fn main() -> anyhow::Result<()> { let redis_connection = redis_client.get_multiplexed_async_connection().await?; // starting HubService - let hub = HubServiceHandle::start(redis_connection.clone()); - - let hub_state = HubState::default(); - let hub_state = Arc::new(RwLock::new(hub_state)); + let hub_state = Arc::new(RwLock::new(HubState::default())); // starting Logger - tokio::spawn(redis::receiver(redis_client, hub.clone())); + tokio::spawn(redis::receiver(redis_client, hub_state.clone())); let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port); @@ -144,27 +141,20 @@ async fn main() -> anyhow::Result<()> { .route("/{key:.+}", web::put().to(handlers_http::put)) .route("/{key:.+}", web::delete().to(handlers_http::delete)), ) - .route( - "/ws", - web::get() - .to(handlers_ws::handler) + .route("/ws",web::get().to(handlers_ws::handler) .wrap(middleware::from_fn(extract_claims)), ) // WebSocket - // .route("/status", web::get().to(async || "ok")) - .route( - "/status", - web::get().to(|hub: web::Data| async move { - let count = hub.count().await; - Ok::<_, actix_web::Error>( - HttpResponse::Ok().json(json!({ "websockets": count, "status": "OK" })), - ) - }), - ) - - // .route("/subs", web::get().to(|hub: web::Data| async move { - // let subs = hub.dump_subs().await; - // Ok::<_, actix_web::Error>(HttpResponse::Ok().json(subs)) - // })) + .route("/status", web::get().to({ + move |hub_state: web::Data>>| { + let hub_state = hub_state.clone(); + async move { + let count = hub_state.read().await.count(); + Ok::<_, actix_web::Error>( + HttpResponse::Ok().json(json!({ "websockets": count, "status": "OK" })), + ) + } + } + })) }) .bind(socket)? .run(); diff --git a/src/redis.rs b/src/redis.rs index f030b7f1be..40c1564b1f 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -13,15 +13,16 @@ // limitations under the License. // -use std::time::{SystemTime, UNIX_EPOCH}; +use std::{sync::Arc, time::{SystemTime, UNIX_EPOCH}}; use ::redis::Msg; +use tokio::sync::RwLock; use tokio_stream::StreamExt; use tracing::*; use crate::{ - config::{CONFIG, RedisMode}, - hub_service::{HubServiceHandle, RedisEvent, RedisEventAction}, + config::{RedisMode, CONFIG}, + hub_service::{push_event, HubState, RedisEvent, RedisEventAction}, }; #[derive(serde::Serialize)] @@ -336,7 +337,7 @@ impl TryFrom for RedisEvent { // "__keyevent@0__:set" → event="set", db=0; payload = key let event = channel.rsplit(':').next().unwrap_or(""); - let action = match event { + let message = match event { "set" => RedisEventAction::Set, "del" => RedisEventAction::Del, "unlink" => RedisEventAction::Unlink, @@ -354,12 +355,15 @@ impl TryFrom for RedisEvent { Ok(RedisEvent { // db, key: payload.clone(), - action, + message, }) } } -pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Result<()> { +pub async fn receiver(redis_client: Client, + // hub: HubServiceHandle + hub_state: Arc>, +) -> anyhow::Result<()> { let mut redis = redis_client.get_multiplexed_async_connection().await?; let mut pubsub = redis_client.get_async_pubsub().await?; @@ -386,7 +390,8 @@ pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Re Ok(ev) => { // debug!("redis event: {ev:#?}"); - hub.push_event(ev); + push_event(&hub_state, &mut redis, ev).await; + } Err(e) => { warn!("invalid redis message: {e}"); diff --git a/src/workspace_owner.rs b/src/workspace_owner.rs index 0b427d9eda..2014953921 100644 --- a/src/workspace_owner.rs +++ b/src/workspace_owner.rs @@ -16,8 +16,15 @@ use hulyrs::services::jwt::Claims; use uuid::Uuid; +use crate::redis::deprecated_symbol; + // common checker pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> { + + if deprecated_symbol(key) { + return Err("Invalid key: deprecated symbols"); + } + if claims.is_system() { return Ok(()); }