diff --git a/Cargo.lock b/Cargo.lock index 5a7f956..3ef55ce 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 98c410d..04f844d 100644 --- a/crates/server/Cargo.toml +++ b/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" diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 7c6d671..707da77 100644 --- a/crates/server/src/main.rs +++ b/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; diff --git a/crates/server/src/stream_session_2.rs b/crates/server/src/stream_session_2.rs new file mode 100644 index 0000000..888e7a0 --- /dev/null +++ b/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), + } + } +} diff --git a/crates/server/src/webrtc.rs b/crates/server/src/webrtc.rs index 8191c66..9c3a594 100644 --- a/crates/server/src/webrtc.rs +++ b/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 = None; let mut audio_pt = None; + let mut channel_id = None; let mut connected = false; let mut video_stream: Option>> = None; let mut audio_stream: Option>> = 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>, + 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) + } + } }