Bump : 0.6.0 : Data channel pipeline, currently used for sending viewers data

Cargo.lock

@@ -2856,9 +2856,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.150"
+version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9"
+checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
  "itoa",
  "memchr",
@@ -2892,7 +2892,7 @@ dependencies = [
 
 [[package]]
 name = "server"
-version = "0.5.4"
+version = "0.6.0"
 dependencies = [
  "argon2",
  "async-broadcast",
@@ -2908,6 +2908,7 @@ dependencies = [
  "rubato",
  "sea-orm",
  "serde",
+ "serde_json",
  "str0m",
  "symphonia",
  "sysinfo",

crates/server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.5.4"
+version = "0.6.0"
 edition = "2024"
 
 [target.x86_64-unknown-linux-gnu]
@@ -36,3 +36,4 @@ time = "0.3"
 thiserror = "2.0.18"
 sysinfo = "0.36"
 axum-extra = { version = "0.12.6", features = ["cookie"] }
+serde_json = "1.0.151"

crates/server/src/main.rs

@@ -27,6 +27,7 @@ mod hash;
 mod http;
 mod http_error;
 mod rtmp;
+mod stream_session_2;
 mod webrtc;
 mod webrtc_ingest;
 mod webrtc_proxy;

crates/server/src/stream_session_2.rs

@@ -0,0 +1,18 @@
+use serde::Serialize;
+
+use crate::StreamSession;
+
+#[derive(Serialize)]
+pub struct StreamUpdateData {
+    viewers: u32,
+}
+
+impl StreamSession {
+    pub fn stream_update_data(&self) -> StreamUpdateData {
+        StreamUpdateData {
+            viewers: self
+                .active_clients
+                .load(std::sync::atomic::Ordering::Relaxed),
+        }
+    }
+}

crates/server/src/webrtc.rs

@@ -2,13 +2,18 @@ use bytes::Bytes;
 use dashmap::DashMap;
 use entity::stream_key;
 use sea_orm::{DatabaseConnection, EntityTrait};
-use std::{net::SocketAddr, sync::Arc, time::Instant};
+use std::{
+    net::SocketAddr,
+    sync::Arc,
+    time::{Duration, Instant},
+};
 use tokio::{net::UdpSocket, sync::mpsc::Receiver};
 use tracing::{debug, error, info, trace, warn};
 
 use str0m::{
     Candidate, Event, Input, Output, Rtc,
     change::SdpOffer,
+    channel::ChannelId,
     media::{Frequency, MediaKind, MediaTime, Mid, Pt},
     net::{Protocol, Receive},
 };
@@ -236,10 +241,34 @@ impl Webrtc {
         let mut video_pt = None;
         let mut audio_mid: Option<Mid> = None;
         let mut audio_pt = None;
+        let mut channel_id = None;
         let mut connected = false;
         let mut video_stream: Option<async_broadcast::Receiver<Arc<VideoFrame>>> = None;
         let mut audio_stream: Option<async_broadcast::Receiver<Arc<OpusAudioFrame>>> = None;
         let mut saw_keyframe = false;
+        // Update client with stream info through webrtc data channel ()
+        let mut tick = tokio::time::interval(Duration::from_millis(2000));
+
+        if let Some(ses) = sessions_ref.get(&stream_id) {
+            ses.active_clients
+                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        }
+        struct ActiveClientGuard {
+            sessions: Arc<DashMap<i32, StreamSession>>,
+            id: i32,
+        }
+        impl Drop for ActiveClientGuard {
+            fn drop(&mut self) {
+                if let Some(ses) = self.sessions.get(&self.id) {
+                    ses.active_clients
+                        .fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
+                }
+            }
+        }
+        let _active_client_guard = ActiveClientGuard {
+            sessions: sessions_ref.clone(),
+            id: stream_id,
+        };
 
         loop {
             let deadline = loop {
@@ -252,6 +281,9 @@ impl Webrtc {
                         }
                     }
                     Ok(Output::Event(e)) => match e {
+                        Event::ChannelOpen(channelId, _name) => {
+                            channel_id = Some(channelId);
+                        }
                         Event::MediaAdded(ma) => {
                             info!(stream_id, kind = ?ma.kind, mid = ?ma.mid, "MediaAdded");
                             if ma.kind == MediaKind::Video {
@@ -431,6 +463,14 @@ impl Webrtc {
                         Err(async_broadcast::RecvError::Overflowed(_)) => {}
                     }
                 }
+                _interval = tick.tick() => {
+               if let Some(id) = channel_id
+                   && let Some(session) = sessions_ref.get(&stream_id)
+                   && let Ok(json) = serde_json::to_vec(&session.stream_update_data())
+                   && let Some(mut ch) = rtc.channel(id) {
+                       let _ = ch.write(false, &json);
+                   }
+                }
             }
         }
     }
@@ -497,4 +537,12 @@ impl Webrtc {
             warn!("RTP write error: {:?}", e);
         }
     }
+
+    fn write_channel_data(rtc: &mut Rtc, channel_id: ChannelId, data: &[u8]) {
+        if let Some(mut channel) = rtc.channel(channel_id)
+            && let Err(e) = channel.write(false, data)
+        {
+            warn!("Channel write error: {:?}", e)
+        }
+    }
 }