fixed-broadcast

Cargo.lock

@@ -487,6 +487,7 @@ name = "discoursio-presense"
 version = "0.2.8"
 dependencies = [
  "actix-web",
+ "bytes",
  "chrono",
  "futures",
  "redis",

Cargo.toml

@@ -7,10 +7,11 @@ edition = "2021"
 
 [dependencies]
 actix-web = "4.4.0"
+tokio = { version = "1", features = ["full"]}
 redis = { version = "0.23", features = ["tokio-comp"]}
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-tokio = { version = "1", features = ["full"] }
+bytes = "1.0"
 reqwest = { version = "0.11", features = ["json"] }
 futures = "0.3.28"
 uuid = { version = "1.4.1", features = ["v4"] }

src/main.rs

@@ -3,15 +3,15 @@ use redis::{Client, AsyncCommands};
 use std::collections::HashMap;
 use std::env;
 use futures::StreamExt;
-use tokio::sync::broadcast::{self, Receiver};
+use tokio::sync::broadcast;
 
-pub mod data;
+mod data;
 
 async fn sse_handler(
     token: web::Path<String>,
-    rx: web::Data<Receiver<String>>,
     redis: web::Data<Client>,
 ) -> impl Responder {
+
     let author_id = match data::get_auth_id(&token).await {
         Ok(id) => id,
         Err(e) => {
@@ -62,34 +62,30 @@ async fn sse_handler(
         }
     };
 
-    let mut pubsub = con.into_pubsub();
+    let (tx, mut rx) = broadcast::channel(100);
-    for chat_id in &chats {
+    let _handle = tokio::spawn(async move {
-        if let Err(e) = pubsub.subscribe(format!("chat:{}", chat_id)).await {
+        let conn = redis.get_async_connection().await.expect("Failed to get async connection");
-            eprintln!("Failed to subscribe to chat: {}", e);
+        let mut pubsub = conn.into_pubsub();
-            return HttpResponse::InternalServerError().finish();
-        }
-    }
 
-    let mut con = match redis.get_async_connection().await {
+        pubsub.subscribe("new_follower").await.expect("Failed to subscribe to new_follower");
-        Ok(con) => con,
+        pubsub.subscribe("new_shout").await.expect("Failed to subscribe to new_shout");
-        Err(e) => {
+        pubsub.subscribe("new_reaction").await.expect("Failed to subscribe to new_reaction");
-            eprintln!("Failed to get async connection: {}", e);
+        for chat_id in &chats {
-            return HttpResponse::InternalServerError().finish();
+            let channel_name = format!("chat:{}", chat_id);
+            pubsub
+                .subscribe(channel_name.clone())
+                .await
+                .expect(&format!("Failed to subscribe to {}", channel_name));
         }
-    };
 
-    let _ = match con.srem::<&str, &i32, usize>("authors-online", &author_id).await {
+        while let Some(msg) = pubsub.on_message().next().await {
-        Ok(_) => (),
+            let payload: HashMap<String, String> = msg.get_payload().expect("Failed to get payload");
-        Err(e) => {
+            tx.clone().send(serde_json::to_string(&payload).expect("Failed to serialize payload")).expect("Failed to send payload");
-            eprintln!("Failed to remove author from online list: {}", e);
-            return HttpResponse::InternalServerError().finish();
         }
-    };
+    });
+    
 
-    // FIXME: cannot borrow data in an `Arc` as mutable
+    let server_event = match rx.recv().await {
-    // trait `DerefMut` is required to modify through a dereference, 
-    // but it is not implemented for `Arc<tokio::sync::broadcast::Receiver<std::string::String>>`
-    let server_event = match rx.recv().await { 
         Ok(event) => event,
         Err(e) => {
             eprintln!("Failed to receive server event: {}", e);
@@ -105,28 +101,11 @@ async fn sse_handler(
 
 #[actix_web::main]
 async fn main() -> std::io::Result<()> {
-    let (tx, _rx) = broadcast::channel(100);
     let redis_url = env::var("REDIS_URL").expect("REDIS_URL must be set");
     let client = redis::Client::open(redis_url).expect("Failed to open Redis client");
 
-    let _handle = tokio::spawn(async move {
-        let mut conn = client.get_async_connection().await.expect("Failed to get async connection");
-        let mut pubsub = conn.into_pubsub();
-
-        pubsub.subscribe("new_follower").await.expect("Failed to subscribe to new_follower");
-        pubsub.subscribe("new_shout").await.expect("Failed to subscribe to new_shout");
-        pubsub.subscribe("new_reaction").await.expect("Failed to subscribe to new_reaction");
-
-        while let Some(msg) = pubsub.on_message().next().await {
-            let payload: HashMap<String, String> = msg.get_payload().expect("Failed to get payload");
-            tx.send(serde_json::to_string(&payload).expect("Failed to serialize payload")).expect("Failed to send payload");
-        }
-    });
-
     HttpServer::new(move || {
-        let rx = tx.subscribe();
         App::new()
-            .app_data(web::Data::new(rx))
             .app_data(web::Data::new(client.clone()))
             .route("/presence/{token}", web::get().to(sse_handler))
     })