discours.io/quoter
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixes

Browse Source
This commit is contained in:
Tony Rewin 2023-10-02 14:48:27 +03:00
parent 52a9d2df3a
commit 908a94056b
1 changed files with 6 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/main.rs
View File

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