🏡 index : ~doyle/scrs.git

073311bc40feae2260dcd40e7aeeea5bcd19f722 [patch]

c6bcb05050518305dbb931b2ed731d75e09fcdf9

2607d45e7a61c2427348f10d0115ae713e4e9af3

073311bc40feae2260dcd40e7aeeea5bcd19f722.tar.gz

Add concurrent listener count

Diff

 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());