From 073311bc40feae2260dcd40e7aeeea5bcd19f722 Mon Sep 17 00:00:00 2001
From: Jordan Doyle <jordan@doyle.la>
Date: Thu, 23 Jul 2020 00:02:19 +0000
Subject: [PATCH] Add concurrent listener count

---
 scrs-serve/src/main.rs | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/scrs-serve/src/main.rs b/scrs-serve/src/main.rs
index d712604..0b97f9a 100644
--- a/scrs-serve/src/main.rs
+++ a/scrs-serve/src/main.rs
@@ -14,7 +14,10 @@
 
 use std::convert::TryFrom;
 use std::net::SocketAddr;
-use std::sync::Arc;
+use std::sync::{
+    atomic::{AtomicU64, Ordering},
+    Arc,
+};
 
 #[derive(Debug, Deref)]
 struct Request(http::Request<()>);
@@ -41,6 +44,7 @@
 struct MetadataContainer {
     stream: ArcSwap<Option<StreamMetadata>>,
     track: ArcSwap<Option<TrackMetadata>>,
+    listeners: AtomicU64,
 }
 
 impl serde::ser::Serialize for MetadataContainer {
@@ -52,6 +56,7 @@
         let mut struc = serializer.serialize_struct("", 2)?;
         struc.serialize_field("stream", &**self.stream.load())?;
         struc.serialize_field("track", &**self.track.load())?;
+        struc.serialize_field("listeners", &self.listeners)?;
         struc.end()
     }
 }
@@ -169,6 +174,8 @@
             let resp = resp.header("Content-Type", "audio/mpeg").body(()).unwrap();
             write_response(&mut stream, resp).await;
 
+            metadata.listeners.fetch_add(1, Ordering::Relaxed);
+
             loop {
                 if let Some(v) = subscriber.next().await {
                     if stream.write_all(v.as_ref()).await.is_err() {
@@ -176,6 +183,8 @@
                     }
                 }
             }
+
+            metadata.listeners.fetch_sub(1, Ordering::Relaxed);
         }
         "/stream" => {
             println!("stream req: {:?}", req.headers());
--
rgit 0.1.3