From 8860a10cea976e73ffc0fceb8a751d00d9641597 Mon Sep 17 00:00:00 2001 From: Alexey Aristov Date: Fri, 22 Aug 2025 12:23:26 +0200 Subject: [PATCH] code formatting Signed-off-by: Alexey Aristov --- src/handlers_http.rs | 3 +-- src/handlers_ws.rs | 18 ++++-------------- src/hub_service.rs | 17 +++++++++-------- src/main.rs | 39 +++++++++++++++++++++++---------------- src/redis.rs | 8 ++++---- src/workspace_owner.rs | 3 +-- 6 files changed, 42 insertions(+), 46 deletions(-) diff --git a/src/handlers_http.rs b/src/handlers_http.rs index dd31bed486..74766e218e 100644 --- a/src/handlers_http.rs +++ b/src/handlers_http.rs @@ -15,7 +15,7 @@ use anyhow::anyhow; use redis::aio::MultiplexedConnection; -use serde::{Deserialize}; +use serde::Deserialize; use tracing::*; use actix_web::{ @@ -162,7 +162,6 @@ pub async fn delete( path: web::Path, redis: web::Data, ) -> Result { - let key: String = path.into_inner().key; trace!(key, "delete request"); diff --git a/src/handlers_ws.rs b/src/handlers_ws.rs index 6eff578291..93410f1bb3 100644 --- a/src/handlers_ws.rs +++ b/src/handlers_ws.rs @@ -13,9 +13,7 @@ // limitations under the License. // -use actix::{ - Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut, -}; +use actix::{Actor, ActorContext, ActorFutureExt, AsyncContext, StreamHandler, fut}; use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse, web}; use actix_web_actors::ws; use redis::aio::MultiplexedConnection; @@ -29,7 +27,6 @@ use crate::redis::{ use crate::hub_service::{HubServiceHandle, ServerMessage, SessionId, new_session_id}; use crate::workspace_owner::check_workspace_core; - #[derive(Serialize, Default)] struct ReturnBase<'a> { action: &'a str, @@ -45,7 +42,6 @@ struct ReturnBase<'a> { #[serde(rename = "TTL", skip_serializing_if = "Option::is_none")] ttl: Option, - // #[serde(rename = "expiresAt", skip_serializing_if = "Option::is_none")] // expires_at: Option, @@ -157,12 +153,12 @@ impl Actor for WsSession { impl actix::Handler for WsSession { type Result = (); fn handle(&mut self, msg: ServerMessage, ctx: &mut Self::Context) { - let json = serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into()); + let json = + serde_json::to_string(&msg).unwrap_or_else(|_| "{\"error\":\"serialization\"}".into()); ctx.text(json); } } - /// StreamHandler External trait: must be in separate impl block impl StreamHandler> for WsSession { fn handle(&mut self, msg: Result, ctx: &mut Self::Context) { @@ -404,7 +400,6 @@ impl WsSession { self.fut_send(ctx, fut, base); } - WsCommand::Sub { key, correlation } => { // LEVENT 3 tracing::info!("SUB {}", &key); // correlation: {:?} , &correlation @@ -433,7 +428,6 @@ impl WsSession { ctx.text(obj.to_string()); } - WsCommand::Unsub { key, correlation } => { // LEVENT 4 tracing::info!("UNSUB {}", &key); // correlation: {:?} , &correlation @@ -466,7 +460,6 @@ impl WsSession { ctx.text(obj.to_string()); } - WsCommand::Sublist { correlation } => { tracing::info!("SUBLIST"); // correlation: {:?} , &correlation // w/o Check workspace! @@ -487,14 +480,11 @@ impl WsSession { }, base, ); - } - - // End of commands + } // End of commands } } } - pub async fn handler( req: HttpRequest, payload: web::Payload, diff --git a/src/hub_service.rs b/src/hub_service.rs index 39e0b71fff..859f5d5f2a 100644 --- a/src/hub_service.rs +++ b/src/hub_service.rs @@ -48,7 +48,7 @@ pub enum RedisEventAction { #[derive(Debug, Clone, Serialize)] pub struct RedisEvent { -// pub db: u32, + // pub db: u32, pub key: String, pub action: RedisEventAction, } @@ -119,7 +119,6 @@ impl HubServiceHandle { while let Some(cmd) = rx.recv().await { match cmd { - Command::Connect { session_id, addr } => { sessions.insert(session_id, addr); } @@ -177,7 +176,6 @@ impl HubServiceHandle { // .collect::>(); // let _ = reply.send(snapshot); // } - Command::RedisEvent(event) => { let targets = subscribers_for(&subs, &event.key); if targets.is_empty() { @@ -204,10 +202,7 @@ impl HubServiceHandle { } } - let payload = ServerMessage { - event, - value, - }; + let payload = ServerMessage { event, value }; for rcpt in recipients { let _ = rcpt.do_send(payload.clone()); @@ -244,7 +239,13 @@ impl HubServiceHandle { 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 }).await; + let _ = self + .tx + .send(Command::SubscribeList { + session_id, + reply: tx, + }) + .await; rx.await.unwrap_or_default() } diff --git a/src/main.rs b/src/main.rs index 13b8c9f204..d524e68e85 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,11 +13,13 @@ // limitations under the License. // -// https://github.com/hcengineering/hulypulse/ - use actix_cors::Cors; use actix_web::{ - body::MessageBody, dev::{ServiceRequest, ServiceResponse}, middleware::{self, Next}, web::{self, Path, Query}, App, Error, HttpMessage, HttpResponse, HttpServer + App, Error, HttpMessage, HttpResponse, HttpServer, + body::MessageBody, + dev::{ServiceRequest, ServiceResponse}, + middleware::{self, Next}, + web::{self, Path, Query}, }; use hulyrs::services::jwt::{Claims, actix::ServiceRequestExt}; use secrecy::ExposeSecret; @@ -109,7 +111,7 @@ async fn main() -> anyhow::Result<()> { let url = format!("http://{}:{}", &CONFIG.bind_host, &CONFIG.bind_port); tracing::info!("Server running at {}", &url); - tracing::info!("HTTP API: {}/api", &url); + tracing::info!("HTTP API: {}/api", &url); tracing::info!("WebSocket API: {}/ws", &url); tracing::info!("Status: {}/status", &url); @@ -135,22 +137,27 @@ 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("/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("/subs", web::get().to(|hub: web::Data| async move { + // let subs = hub.dump_subs().await; + // Ok::<_, actix_web::Error>(HttpResponse::Ok().json(subs)) + // })) }) .bind(socket)? .run(); diff --git a/src/redis.rs b/src/redis.rs index bd090d6d4f..f030b7f1be 100644 --- a/src/redis.rs +++ b/src/redis.rs @@ -19,7 +19,10 @@ use ::redis::Msg; use tokio_stream::StreamExt; use tracing::*; -use crate::{config::{RedisMode, CONFIG}, hub_service::{HubServiceHandle, RedisEvent, RedisEventAction}}; +use crate::{ + config::{CONFIG, RedisMode}, + hub_service::{HubServiceHandle, RedisEvent, RedisEventAction}, +}; #[derive(serde::Serialize)] pub enum Ttl { @@ -356,7 +359,6 @@ impl TryFrom for RedisEvent { } } - pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Result<()> { let mut redis = redis_client.get_multiplexed_async_connection().await?; let mut pubsub = redis_client.get_async_pubsub().await?; @@ -382,11 +384,9 @@ pub async fn receiver(redis_client: Client, hub: HubServiceHandle) -> anyhow::Re while let Some(message) = messages.next().await { match RedisEvent::try_from(message) { Ok(ev) => { - // debug!("redis event: {ev:#?}"); hub.push_event(ev); - } Err(e) => { warn!("invalid redis message: {e}"); diff --git a/src/workspace_owner.rs b/src/workspace_owner.rs index 148b06740c..0b427d9eda 100644 --- a/src/workspace_owner.rs +++ b/src/workspace_owner.rs @@ -18,7 +18,6 @@ use uuid::Uuid; // common checker pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static str> { - if claims.is_system() { return Ok(()); } @@ -41,4 +40,4 @@ pub fn check_workspace_core(claims: &Claims, key: &str) -> Result<(), &'static s } Ok(()) -} \ No newline at end of file +}