diff --git a/Cargo.lock b/Cargo.lock index 7a479d2..239b925 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2872,6 +2872,7 @@ dependencies = [ "async-broadcast", "axum", "bytes", + "chrono", "dashmap", "entity", "futures", diff --git a/Dockerfile b/Dockerfile index c942b1d..2d84168 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,6 +43,6 @@ EXPOSE 1935 EXPOSE 3000 EXPOSE 6969/udp -ENV RUST_LOG="warn" +ENV RUST_LOG="info,warn" CMD ["/rtmp-to-whip"] diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index f90fbe0..40dc169 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -34,3 +34,4 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } symphonia = { version = "0.5", features = ["aac"] } opus = "0.3.1" rubato = "3.0.0" +chrono = "0.4.45" diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index e7cc5d2..d67565b 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -18,8 +18,9 @@ use axum::{ response::{IntoResponse, Response}, routing::{get, post}, }; +use chrono::{DateTime, Utc}; use entity::{auth_session, stream_key, stream_session, users}; -use sea_orm::{DatabaseConnection, EntityTrait, QueryFilter}; +use sea_orm::{DatabaseConnection, EntityTrait, QueryFilter, prelude::DateTimeUtc}; use serde::{Deserialize, Serialize}; use tokio::{ net::TcpListener, @@ -108,6 +109,7 @@ struct StreamListing { label: String, id: i32, user: String, + started_at: DateTime, //UNIX TIMESTAMP } async fn catalog_handler(State(state): State>) -> impl IntoResponse { @@ -118,6 +120,7 @@ async fn catalog_handler(State(state): State>) -> impl IntoRespo let catalog: Vec = futures::future::join_all(streams.iter().map(|listing| { let db = state.db.clone(); let stream_key_id = listing.stream_key_id; + let state2 = state.clone(); async move { let key_info = stream_key::Entity::find_by_id(stream_key_id) .one(&db) @@ -129,10 +132,20 @@ async fn catalog_handler(State(state): State>) -> impl IntoRespo .await .unwrap() .unwrap(); + let meow = state2 + .appstate + .clone() + .lock() + .await + .stream_sessions + .get(&key_info.id) + .unwrap() + .started_at; StreamListing { id: stream_key_id, label: key_info.label, user: user.username, + started_at: meow, } } })) diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index bd2b960..a366adc 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -1,3 +1,4 @@ +use ::chrono::{DateTime, Utc}; use std::{env, error::Error, sync::Arc}; use tracing::{info, level_filters::LevelFilter, warn}; @@ -57,6 +58,8 @@ pub struct StreamSession { pub frame_channel: async_broadcast::Sender>, pub audio_channel: async_broadcast::Sender>, pub codec: Option, + // + pub started_at: DateTime, } #[tokio::main] diff --git a/crates/server/src/rtmp.rs b/crates/server/src/rtmp.rs index eadf16e..7edc73b 100644 --- a/crates/server/src/rtmp.rs +++ b/crates/server/src/rtmp.rs @@ -1,6 +1,7 @@ use std::{error::Error, sync::Arc}; use async_broadcast::broadcast; +use chrono::Utc; use dashmap::DashMap; use entity::stream_session; use rml_rtmp::{ @@ -11,6 +12,7 @@ use sea_orm::{DatabaseConnection, IntoActiveModel, sqlx::types::chrono::Local}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream}, + time::Instant, }; use tracing::{debug, info, warn}; @@ -233,6 +235,7 @@ impl Rtmp { frame_channel: video_tx.clone(), audio_channel: audio_tx.clone(), codec: None, + started_at: Utc::now(), }, ); diff --git a/crates/server/src/webrtc.rs b/crates/server/src/webrtc.rs index f25dd1c..95d7538 100644 --- a/crates/server/src/webrtc.rs +++ b/crates/server/src/webrtc.rs @@ -2,23 +2,14 @@ use bytes::Bytes; use dashmap::DashMap; use entity::stream_key; use sea_orm::{DatabaseConnection, EntityTrait}; -use std::{ - error::Error, - net::SocketAddr, - sync::Arc, - time::{Duration, Instant}, -}; -use tokio::{ - net::UdpSocket, - sync::mpsc::{Receiver, Sender}, - time::sleep, -}; +use std::{net::SocketAddr, sync::Arc, time::Instant}; +use tokio::{net::UdpSocket, sync::mpsc::Receiver}; use tracing::{debug, error, info, warn}; use str0m::{ Candidate, Event, Input, Output, Rtc, change::SdpOffer, - media::{Frequency, MediaKind, MediaTime, Mid}, + media::{Frequency, MediaKind, MediaTime, Mid, Pt}, net::{Protocol, Receive}, }; @@ -92,7 +83,15 @@ impl Webrtc { cc.enable_av1(true); } _ => { - cc.add_h264(102.into(), None, true, 0x4d001f); // Main + // Advertise the common H.264 profiles. str0m matches an + // incoming offer's payload by profile-level-id, so we must + // list every profile a browser might offer or negotiation + // fails and the m=video line comes back with an empty PT + // list. Firefox in particular only ever offers Constrained + // Baseline (0x42e01f) — without it Firefox sees no codec. + cc.add_h264(102.into(), None, true, 0x42e01f); // Constrained Baseline + cc.add_h264(104.into(), None, true, 0x4d001f); // Main + cc.add_h264(106.into(), None, true, 0x64001f); // High } } } @@ -216,8 +215,22 @@ impl Webrtc { .map(|p| (p.pt(), p.spec().codec)) .collect(); info!(stream_id, ?all_pts, "video payload params"); + // Rank payload types by their profile/level so we + // pick the highest-capability PT the browser + // negotiated. The relevant field differs per codec: + // H.264 uses profile_level_id, H.265 stores it in + // h265_profile_tier_level, AV1 in level_idx. Keying + // only on profile_level_id (H.264's field) leaves + // every H.265/AV1 PT tied at 0, so the pick becomes + // arbitrary when a browser offers several of them. let best = writer.payload_params().max_by_key(|p| { - p.spec().format.profile_level_id.unwrap_or(0) + let f = &p.spec().format; + f.profile_level_id + .or(f + .h265_profile_tier_level + .map(|ptl| ptl.level_id() as u32)) + .or(f.level_idx.map(|l| l as u32)) + .unwrap_or(0) }); if let Some(params) = best { info!(stream_id, pt = ?params.pt(), codec = ?params.spec().codec, "selected video PT"); @@ -301,105 +314,15 @@ impl Webrtc { } } } - if let Some(ref mut stream) = video_stream { - let mut wrote_any = false; - for _ in 0..8 { - match stream.try_recv() { - Ok(frame) => { - if !saw_keyframe { - if !frame.is_keyframe { - continue; - } - saw_keyframe = true; - info!( - stream_id, - ts = frame.timestamp_ms, - "first keyframe — starting RTP send" - ); - } - let now = Instant::now(); - let rtp_time = - MediaTime::from_90khz(frame.timestamp_ms as u64 * 90); - match (video_pt, video_mid.and_then(|m| rtc.writer(m))) { - (Some(pt), Some(writer)) => { - match writer.write(pt, now, rtp_time, frame.data.to_vec()) { - Ok(_) => wrote_any = true, - Err(e) => { - error!(stream_id, "video RTP write error: {:?}", e) - } - } - } - _ => warn!( - stream_id, - "video_pt or writer not ready, dropping frame (video_pt={:?} video_mid={:?})", - video_pt, - video_mid - ), - } - } - Err(async_broadcast::TryRecvError::Empty) => break, - Err(async_broadcast::TryRecvError::Closed) => { - warn!("video channel closed, stream ended"); - sleep(Duration::from_secs(1)).await; - break; - } - Err(async_broadcast::TryRecvError::Overflowed(_)) => { - // Frames were dropped from the ring buffer; resuming mid-GOP - // would give the decoder frames without their references. - // Wait for the next keyframe. - saw_keyframe = false; - continue; - } - } - } - if let Some(ref mut audio) = audio_stream { - for _ in 0..8 { - match audio.try_recv() { - Ok(frame) => { - let now = Instant::now(); - let rtp_time = MediaTime::new( - frame.timestamp_ms as u64 * 48, - Frequency::FORTY_EIGHT_KHZ, - ); - if let (Some(pt), Some(writer)) = - (audio_pt, audio_mid.and_then(|m| rtc.writer(m))) - { - match writer.write(pt, now, rtp_time, frame.data.to_vec()) { - Ok(_) => wrote_any = true, - Err(e) => warn!("RTP write error: {:?}", e), - } - } - } - Err(async_broadcast::TryRecvError::Empty) => break, - Err(async_broadcast::TryRecvError::Closed) => { - info!("audio channel closed, stream ended"); - break; - } - Err(async_broadcast::TryRecvError::Overflowed(_)) => continue, - } - } - } - // If we queued RTP data, drain poll_output immediately so packets are - // transmitted in this iteration rather than 20ms later. But still drive - // str0m's timeout if the deadline has passed — ICE consent refresh depends on it. - if wrote_any { - let now = Instant::now(); - if now >= deadline { - if let Err(e) = rtc.handle_input(Input::Timeout(now)) { - error!("handle_input(Timeout) error: {:?}", e); - return; - } - } - continue; - } - } } - // debug!("waiting"); - - let wait_until = deadline - .min(Instant::now() + Duration::from_millis(20)) - .max(Instant::now()); + // Wait for whichever happens first: str0m's own deadline, an incoming UDP + // packet, or a fresh video/audio frame. Waking directly on frame arrival + // (instead of polling on a fixed timer) keeps RTP pacing tight to the + // source frame rate — a fixed poll interval coarser than the frame period + // (e.g. 20ms vs. a 16.7ms 60fps cadence) causes frames to be delivered in + // uneven bursts, which reads as judder even though decode timestamps are correct. + let wait_until = deadline.max(Instant::now()); let sleep = tokio::time::sleep_until(wait_until.into()); tokio::select! { @@ -427,6 +350,155 @@ impl Webrtc { } } } + res = Webrtc::recv_video(&mut video_stream), if video_stream.is_some() => { + match res { + Ok(frame) => { + Webrtc::write_video_frame( + frame, &mut saw_keyframe, video_pt, video_mid, &mut rtc, stream_id, + ); + Webrtc::drain_video( + &mut video_stream, &mut saw_keyframe, video_pt, video_mid, &mut rtc, stream_id, + ); + } + Err(async_broadcast::RecvError::Closed) => { + debug!(stream_id, "video channel closed, stream ended"); + } + Err(async_broadcast::RecvError::Overflowed(_)) => { + // Frames were dropped from the ring buffer; resuming mid-GOP + // would give the decoder frames without their references. + // Wait for the next keyframe. + saw_keyframe = false; + } + } + } + res = Webrtc::recv_audio(&mut audio_stream), if audio_stream.is_some() => { + match res { + Ok(frame) => { + Webrtc::write_audio_frame(frame, audio_pt, audio_mid, &mut rtc); + Webrtc::drain_audio(&mut audio_stream, audio_pt, audio_mid, &mut rtc); + } + Err(async_broadcast::RecvError::Closed) => { + debug!("audio channel closed, stream ended"); + } + Err(async_broadcast::RecvError::Overflowed(_)) => {} + } + } + } + } + } + + async fn recv_video( + stream: &mut Option>>, + ) -> Result, async_broadcast::RecvError> { + stream.as_mut().unwrap().recv().await + } + + async fn recv_audio( + stream: &mut Option>>, + ) -> Result, async_broadcast::RecvError> { + stream.as_mut().unwrap().recv().await + } + + fn write_video_frame( + frame: Arc, + saw_keyframe: &mut bool, + video_pt: Option, + video_mid: Option, + rtc: &mut Rtc, + stream_id: i32, + ) { + if !*saw_keyframe { + if !frame.is_keyframe { + return; + } + *saw_keyframe = true; + info!( + stream_id, + ts = frame.timestamp_ms, + "first keyframe — starting RTP send" + ); + } + let now = Instant::now(); + let rtp_time = MediaTime::from_90khz(frame.timestamp_ms as u64 * 90); + match (video_pt, video_mid.and_then(|m| rtc.writer(m))) { + (Some(pt), Some(writer)) => { + if let Err(e) = writer.write(pt, now, rtp_time, frame.data.to_vec()) { + error!(stream_id, "video RTP write error: {:?}", e); + } + } + _ => warn!( + stream_id, + "video_pt or writer not ready, dropping frame (video_pt={:?} video_mid={:?})", + video_pt, + video_mid + ), + } + } + + fn drain_video( + stream: &mut Option>>, + saw_keyframe: &mut bool, + video_pt: Option, + video_mid: Option, + rtc: &mut Rtc, + stream_id: i32, + ) { + let Some(s) = stream.as_mut() else { return }; + for _ in 0..7 { + match s.try_recv() { + Ok(frame) => Webrtc::write_video_frame( + frame, + saw_keyframe, + video_pt, + video_mid, + rtc, + stream_id, + ), + Err(async_broadcast::TryRecvError::Empty) => break, + Err(async_broadcast::TryRecvError::Closed) => { + warn!(stream_id, "video channel closed, stream ended"); + *stream = None; + break; + } + Err(async_broadcast::TryRecvError::Overflowed(_)) => { + *saw_keyframe = false; + } + } + } + } + + fn write_audio_frame( + frame: Arc, + audio_pt: Option, + audio_mid: Option, + rtc: &mut Rtc, + ) { + let now = Instant::now(); + let rtp_time = MediaTime::new(frame.timestamp_ms as u64 * 48, Frequency::FORTY_EIGHT_KHZ); + if let (Some(pt), Some(writer)) = (audio_pt, audio_mid.and_then(|m| rtc.writer(m))) { + if let Err(e) = writer.write(pt, now, rtp_time, frame.data.to_vec()) { + warn!("RTP write error: {:?}", e); + } + } + } + + fn drain_audio( + stream: &mut Option>>, + audio_pt: Option, + audio_mid: Option, + rtc: &mut Rtc, + ) { + let Some(s) = stream.as_mut() else { return }; + for _ in 0..7 { + match s.try_recv() { + Ok(frame) => Webrtc::write_audio_frame(frame, audio_pt, audio_mid, rtc), + Err(async_broadcast::TryRecvError::Empty) => break, + Err(async_broadcast::TryRecvError::Closed) => { + info!("audio channel closed, stream ended"); + *stream = None; + break; + } + Err(async_broadcast::TryRecvError::Overflowed(_)) => {} } } }