diff --git a/src/main.rs b/src/main.rs index 92ae2b1..0de1795 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,7 +6,7 @@ use serde_json::Value; use std::collections::HashMap; use std::env; use std::error::Error; -use futures::FutureExt; +use futures::{StreamExt, FutureExt}; use tokio::sync::broadcast::{self, Receiver}; #[derive(Debug, Serialize, Deserialize)] @@ -32,7 +32,6 @@ struct Payload { body: String, } -// Получаем id автора из токена async fn get_auth_id(token: &str) -> Result> { let api_base = env::var("API_BASE")?; let gql = match api_base.contains("v2") { @@ -53,7 +52,6 @@ async fn get_auth_id(token: &str) -> Result> { Ok(id) } -// Обработчик SSE async fn sse_handler( token: web::Path, rx: web::Data>, @@ -73,25 +71,23 @@ async fn sse_handler( .await .unwrap(); - // Получаем все чаты авторизованного пользователя let chats: Vec = con .smembers(format!("chats_by_author/{}", author_id)) .await .unwrap(); - // Подписываемся на каждый чат в Redis Pub/Sub let mut pubsub = con.into_pubsub(); for chat_id in chats { pubsub.subscribe(format!("message:{}", chat_id)).await.unwrap(); } - let server_event = rx.get_ref().clone().into_stream().map(|result| { + let server_event = rx.get_ref().into_stream().map(|result| { match result { Ok(payload) => { let payload: Payload = serde_json::from_str(&payload).unwrap(); Ok::<_, actix_web::Error>(web::Bytes::from(format!("data: {:?}\n\n", payload.chat_id))) } - Err(_) => Err(actix_web::Error::from(())), + Err(_) => Err(actix_web::Error::from("An error occurred")), } }); @@ -107,7 +103,6 @@ async fn sse_handler( #[actix_web::main] async fn main() -> std::io::Result<()> { - // Создаем одно соединение с Redis Pub/Sub и канал вещания let (tx, rx) = broadcast::channel(100); let redis_url = env::var("REDIS_URL").unwrap(); let client = redis::Client::open(redis_url).unwrap(); @@ -115,14 +110,11 @@ async fn main() -> std::io::Result<()> { let mut conn = client.get_async_connection().await.unwrap(); let mut pubsub = conn.into_pubsub(); - // новый подписчик автора pubsub.subscribe("new_follower").await.unwrap(); - // в теме, автор, сообщество pubsub.subscribe("new_shout").await.unwrap(); - // оценка, комментарий, одобрение в криках автора и реагировавших криках pubsub.subscribe("new_reaction").await.unwrap(); - while let Some(msg) = pubsub.on_message().next().await { + while let Some(msg) = pubsub.on_message().await { let payload: HashMap = msg.get_payload().unwrap(); tx.send(serde_json::to_string(&payload).unwrap()).unwrap(); } @@ -130,8 +122,8 @@ async fn main() -> std::io::Result<()> { HttpServer::new(move || { App::new() - .data(rx.clone()) - .data(client.clone()) + .app_data(web::Data::new(rx.clone())) + .app_data(web::Data::new(client.clone())) .route("/aware/{token}", web::get().to(sse_handler)) }) .bind("127.0.0.1:8080")?