WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-09-09 11:17:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

code formatting

Browse Source 
Signed-off-by: Alexey Aristov <aav@acm.org>
This commit is contained in:
Alexey Aristov
2025-08-22 12:24:07 +02:00
parent d7711b837f
commit 8860a10cea
6 changed files with 42 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/handlers_http.rs
+1 -2
View File
@@ -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<PathParams>,
redis: web::Data<MultiplexedConnection>,
) -> Result<HttpResponse, actix_web::error::Error> {
let key: String = path.into_inner().key;
trace!(key, "delete request");
src/handlers_ws.rs
+4 -14
View File
@@ -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<u64>,
// #[serde(rename = "expiresAt", skip_serializing_if = "Option::is_none")]
// expires_at: Option<u64>,
@@ -157,12 +153,12 @@ impl Actor for WsSession {
impl actix::Handler<ServerMessage> 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<Result<ws::Message, ws::ProtocolError>> for WsSession {
fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, 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,
src/hub_service.rs
+9 -8
View File
@@ -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::<std::collections::HashMap<_, _>>();
// 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<String> {
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()
}
src/main.rs
+23 -16
View File
@@ -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<HubServiceHandle>| 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<HubServiceHandle>| 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<HubServiceHandle>| async move {
// let subs = hub.dump_subs().await;
// Ok::<_, actix_web::Error>(HttpResponse::Ok().json(subs))
// }))
// .route("/subs", web::get().to(|hub: web::Data<HubServiceHandle>| async move {
// let subs = hub.dump_subs().await;
// Ok::<_, actix_web::Error>(HttpResponse::Ok().json(subs))
// }))
})
.bind(socket)?
.run();
src/redis.rs
+4 -4
View File
@@ -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<Msg> 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}");
src/workspace_owner.rs
+1 -2
View File
@@ -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(())
}
}