From c1435a70fb30613f9b2e03f5bad21e5fe6e655d2 Mon Sep 17 00:00:00 2001 From: Doloro1978 Date: Fri, 31 Jul 2026 20:20:48 +0100 Subject: [PATCH] slop --- Cargo.lock | 150 +++++++- crates/server/Cargo.toml | 1 + crates/server/src/codec/av1.rs | 5 +- crates/server/src/codec/h264.rs | 5 +- crates/server/src/codec/h265.rs | 5 +- crates/server/src/http.rs | 122 ++++++- crates/server/src/main.rs | 59 +-- crates/server/src/rtmp.rs | 47 ++- crates/server/src/webrtc.rs | 135 ++++--- crates/server/src/webrtc_ingest.rs | 562 ++++++++++++++++++++++++++++- crates/server/src/webrtc_proxy.rs | 74 ++-- flake.lock | 12 +- 12 files changed, 1009 insertions(+), 168 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e9fc128..8516648 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -618,7 +618,7 @@ dependencies = [ "num-traits", "serde", "wasm-bindgen", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -1481,7 +1481,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core", + "windows-core 0.62.2", ] [[package]] @@ -1857,6 +1857,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "ntapi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" +dependencies = [ + "winapi", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -1936,6 +1945,25 @@ dependencies = [ "libm", ] +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags 2.13.1", +] + +[[package]] +name = "objc2-io-kit" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15" +dependencies = [ + "libc", + "objc2-core-foundation", +] + [[package]] name = "oid-registry" version = "0.8.1" @@ -2031,7 +2059,7 @@ dependencies = [ "libc", "redox_syscall 0.5.18", "smallvec", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -2900,6 +2928,7 @@ dependencies = [ "serde_json", "str0m", "symphonia", + "sysinfo", "thiserror 2.0.18", "tokio", "tower-http", @@ -3515,6 +3544,20 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "sysinfo" +version = "0.36.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "252800745060e7b9ffb7b2badbd8b31cfa4aa2e61af879d0a3bf2a317c20217d" +dependencies = [ + "libc", + "memchr", + "ntapi", + "objc2-core-foundation", + "objc2-io-kit", + "windows", +] + [[package]] name = "tap" version = "1.0.1" @@ -4050,6 +4093,41 @@ dependencies = [ "num-traits", ] +[[package]] +name = "windows" +version = "0.61.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" +dependencies = [ + "windows-collections", + "windows-core 0.61.2", + "windows-future", + "windows-link 0.1.3", + "windows-numerics", +] + +[[package]] +name = "windows-collections" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8" +dependencies = [ + "windows-core 0.61.2", +] + +[[package]] +name = "windows-core" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link 0.1.3", + "windows-result 0.3.4", + "windows-strings 0.4.2", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -4058,9 +4136,20 @@ checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" dependencies = [ "windows-implement", "windows-interface", - "windows-link", - "windows-result", - "windows-strings", + "windows-link 0.2.1", + "windows-result 0.4.1", + "windows-strings 0.5.1", +] + +[[package]] +name = "windows-future" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", + "windows-threading", ] [[package]] @@ -4085,19 +4174,53 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "windows-link" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-numerics" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", +] + +[[package]] +name = "windows-result" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows-result" version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" dependencies = [ - "windows-link", + "windows-link 0.2.1", +] + +[[package]] +name = "windows-strings" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" +dependencies = [ + "windows-link 0.1.3", ] [[package]] @@ -4106,7 +4229,7 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -4133,7 +4256,7 @@ version = "0.61.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -4167,6 +4290,15 @@ dependencies = [ "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows-threading" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.48.5" diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 5b86232..f45c091 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -37,4 +37,5 @@ rubato = "3.0.0" chrono = "0.4.45" regex = "1" thiserror = "2.0.18" +sysinfo = "0.36" axum-extra = { version = "0.12.6", features = ["cookie"] } diff --git a/crates/server/src/codec/av1.rs b/crates/server/src/codec/av1.rs index 62cbc70..158fa73 100644 --- a/crates/server/src/codec/av1.rs +++ b/crates/server/src/codec/av1.rs @@ -95,14 +95,13 @@ impl Av1CodecParser { self.first_coded_frame = false; - if is_keyframe { - if let Some(config) = &self.config_obus { + if is_keyframe + && let Some(config) = &self.config_obus { let mut out = Vec::with_capacity(config.len() + payload.len()); out.extend_from_slice(config); out.extend_from_slice(payload); return Some((out, true)); } - } Some((payload.to_vec(), is_keyframe)) } diff --git a/crates/server/src/codec/h264.rs b/crates/server/src/codec/h264.rs index ddc8a57..8088e93 100644 --- a/crates/server/src/codec/h264.rs +++ b/crates/server/src/codec/h264.rs @@ -148,14 +148,13 @@ impl H264CodecParser { // Prepend SPS+PPS before every keyframe so str0m's packetizer // can bundle them into a STAP-A alongside the IDR NALU. - if is_keyframe { - if let (Some(sps), Some(pps)) = (&self.sps, &self.pps) { + if is_keyframe + && let (Some(sps), Some(pps)) = (&self.sps, &self.pps) { out.extend_from_slice(&[0, 0, 0, 1]); out.extend_from_slice(sps); out.extend_from_slice(&[0, 0, 0, 1]); out.extend_from_slice(pps); } - } // Convert each length-prefixed NALU to an Annex B start-code NALU. let mut i = 0; diff --git a/crates/server/src/codec/h265.rs b/crates/server/src/codec/h265.rs index a475d20..1dab529 100644 --- a/crates/server/src/codec/h265.rs +++ b/crates/server/src/codec/h265.rs @@ -74,8 +74,8 @@ impl H265CodecParser { fn hvcc_to_annexb(&self, payload: &[u8], is_keyframe: bool) -> Option> { let mut out = Vec::with_capacity(payload.len()); - if is_keyframe { - if let (Some(vps), Some(sps), Some(pps)) = (&self.vps, &self.sps, &self.pps) { + if is_keyframe + && let (Some(vps), Some(sps), Some(pps)) = (&self.vps, &self.sps, &self.pps) { out.extend_from_slice(&[0, 0, 0, 1]); out.extend_from_slice(vps); out.extend_from_slice(&[0, 0, 0, 1]); @@ -83,7 +83,6 @@ impl H265CodecParser { out.extend_from_slice(&[0, 0, 0, 1]); out.extend_from_slice(pps); } - } let mut i = 0; while i + 4 <= payload.len() { diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index 9cb33ef..352b67e 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -19,28 +19,25 @@ static KEY_RE: LazyLock = LazyLock::new(|| Regex::new(r"^[A-Za-z0-9 _-]{1 use axum::{ Json, Router, - body::{Body, Bytes}, - extract::{Form, FromRequestParts, Path, Query, State}, + extract::{FromRequestParts, Path, State}, http::{ - HeaderMap, HeaderName, Method, StatusCode, - header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, SET_COOKIE}, + HeaderName, Method, StatusCode, + header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE}, request::Parts, }, response::{IntoResponse, Response}, - routing::{get, post}, + routing::{delete, get, post}, }; use chrono::{DateTime, Utc}; use entity::{auth_session, stream_key, stream_session, users}; -use sea_orm::{ - DatabaseConnection, EntityTrait, IntoActiveModel, QueryFilter, prelude::DateTimeUtc, -}; +use sea_orm::{DatabaseConnection, EntityTrait, IntoActiveModel, QueryFilter}; use serde::{Deserialize, Serialize}; +use sysinfo::System; use tokio::{ net::TcpListener, sync::{Mutex, mpsc::Sender}, }; use tower_http::cors::CorsLayer; -use uuid::Uuid; use tracing::{debug, info, warn}; @@ -48,7 +45,7 @@ use crate::{ AppState, hash::{hash_password, verify_password}, http_error::HttpError, - webrtc_ingest::handle_whip_injest, + webrtc_ingest::{handle_whip_injest, handle_whip_injest_delete, handle_whip_injest_patch}, }; const MAX_USERNAME_LEN: usize = 32; @@ -58,6 +55,11 @@ pub struct HttpServerConfig { pub signup_code: String, } +pub struct ServerInfo { + pub version: &'static str, + pub start_time: std::time::Instant, +} + pub struct HttpServer { pub offer_tx: Sender<(i32, i32, String)>, pub accept_rx: async_broadcast::InactiveReceiver<(i32, Option)>, @@ -65,6 +67,7 @@ pub struct HttpServer { pub request_count: AtomicI32, pub db: DatabaseConnection, pub config: Arc, + pub info: Arc, } impl HttpServer { @@ -77,7 +80,7 @@ impl HttpServer { ]; let cors = CorsLayer::new() - .allow_methods([Method::GET, Method::POST, Method::OPTIONS]) + .allow_methods([Method::GET, Method::POST, Method::PATCH, Method::DELETE, Method::OPTIONS]) .allow_headers([ AUTHORIZATION, ACCEPT, @@ -97,16 +100,28 @@ impl HttpServer { .patch(edit_stream_key), ) // .route("/api/admin/server_stats", get(todo!())) - .route("/api/whip", post(handle_whip_injest)) + // .route("/api/whip", post(handle_whip_injest)) .route("/api/login", post(login_handler)) .route("/api/stream/{slug}", post(stream_handler)) + .route("/stream", post(handle_whip_injest)) + .route("/stream/{slug}", delete(handle_whip_injest_delete).patch(handle_whip_injest_patch)) + .route("/api/whip/{slug}", delete(handle_whip_injest_delete).patch(handle_whip_injest_patch)) .route("/api/meow", get(meow_handler)) + .route("/api/health", get(health_handler)) + .route("/api/uptime", get(uptime_handler)) + .route("/api/version", get(version_handler)) + .route("/api/stats", get(stats_handler)) .layer(cors) .with_state(state); let addr = SocketAddr::from(([0, 0, 0, 0], 3000)); let listener = TcpListener::bind(addr).await.unwrap(); - axum::serve(listener, app).await.unwrap(); + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); } } @@ -336,6 +351,7 @@ fn cookie_for_token(token: &str, dev: bool) -> Cookie<'static> { false => Cookie::build(("session", token)) .path("/") .http_only(true) + // .max_age(Duration::from_hours(999999)) .same_site(axum_extra::extract::cookie::SameSite::Lax) .build(), } @@ -488,3 +504,83 @@ async fn stream_handler( async fn meow_handler() -> &'static str { "meow" } + +#[derive(Serialize)] +struct HealthResponse { + status: &'static str, + uptime_seconds: u64, + version: &'static str, +} + +async fn health_handler( + State(state): State>, +) -> Result { + let uptime = state.info.start_time.elapsed().as_secs(); + Ok(( + StatusCode::OK, + Json(HealthResponse { + status: "ok", + uptime_seconds: uptime, + version: state.info.version, + }), + )) +} + +#[derive(Serialize)] +struct UptimeResponse { + uptime_seconds: u64, +} + +async fn uptime_handler( + State(state): State>, +) -> Result { + let uptime = state.info.start_time.elapsed().as_secs(); + Ok(( + StatusCode::OK, + Json(UptimeResponse { + uptime_seconds: uptime, + }), + )) +} + +async fn version_handler( + State(state): State>, +) -> Result { + Ok(( + StatusCode::OK, + Json(serde_json::json!({ "version": state.info.version })), + )) +} + +#[derive(Serialize)] +struct StatsResponse { + version: &'static str, + uptime_seconds: u64, + cpu_usage_percent: f32, + active_streams: usize, + request_count: i32, +} + +async fn stats_handler( + State(state): State>, +) -> Result { + let uptime = state.info.start_time.elapsed().as_secs(); + let mut system = System::new(); + system.refresh_cpu_all(); + let cpu_usage = system.global_cpu_usage(); + let active_streams = { + let app = state.appstate.lock().await; + app.stream_sessions.len() + }; + let request_count = state.request_count.load(Ordering::Relaxed); + Ok(( + StatusCode::OK, + Json(StatsResponse { + version: state.info.version, + uptime_seconds: uptime, + cpu_usage_percent: cpu_usage, + active_streams, + request_count, + }), + )) +} diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 7a52502..1b06069 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -1,36 +1,32 @@ use ::chrono::{DateTime, Utc}; -use std::{env, error::Error, sync::Arc}; +use std::{ + env, + error::Error, + sync::{Arc, atomic::AtomicU32}, +}; use tracing::{info, warn}; use async_broadcast::broadcast; use dashmap::DashMap; use entity::stream_session; use migration::{Migrator, MigratorTrait}; -use rml_rtmp::{ - handshake::{Handshake, HandshakeProcessResult, PeerType}, - sessions::{ServerSession, ServerSessionConfig, ServerSessionEvent, ServerSessionResult}, -}; -use sea_orm::{ - Database, IntoActiveModel, - sqlx::types::chrono::{self, Local}, -}; +use sea_orm::Database; use tokio::{ - fs, - io::{AsyncReadExt, AsyncWriteExt}, net::TcpListener, sync::Mutex, task::JoinSet, - time::Instant, }; use tracing_subscriber::EnvFilter; use crate::{ audio::OpusAudioFrame, codec::VideoFrame, - http::{HttpServer, HttpServerConfig}, - webrtc_proxy::WebRtcProxyConfig, + http::{HttpServer, HttpServerConfig, ServerInfo}, + webrtc_proxy::{WebRtcProxyConfig, WebrtcProxy}, }; +const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION"); + mod audio; mod codec; mod hash; @@ -44,6 +40,7 @@ mod webrtc_proxy; // #[derive(Debug)] pub struct AppState { pub stream_sessions: Arc>, + pub webrtc_proxy: WebrtcProxy, } #[derive(Clone, Debug)] @@ -53,14 +50,24 @@ pub enum StreamCodec { AV1, } + + pub struct StreamSession { pub stream_key_id: i32, pub stream_key_label: String, pub frame_channel: async_broadcast::Sender>, pub audio_channel: async_broadcast::Sender>, pub codec: Option, - // + // ---- WHIP-negotiated video codec details (populated by whip ingest) ---- + /// Payload type from WHIP SDP negotiation. + pub video_pt: Option, + /// H.264 profile_level_id from WHIP SDP negotiation (e.g. 0x42e01f for CBP). + /// When set alongside `video_pt`, the WHEP answer only advertises this exact + /// profile instead of all three H.264 profiles. + pub video_profile_level_id: Option, + // ---- pub started_at: DateTime, + pub active_clients: AtomicU32, } #[tokio::main] @@ -83,9 +90,17 @@ async fn main() -> Result<(), Box> { stream_session::Model::clean_unended_streams(&db) .await .unwrap(); + let proxyconfig = WebRtcProxyConfig { + proxy_port: env::var("RTC_PORT") + .unwrap_or("6969".into()) + .parse() + .expect("RTC_PORT needs to be a number (i32)"), + }; + let proxy = webrtc_proxy::WebrtcProxy::new(proxyconfig).await.unwrap(); let appstate = Arc::new(Mutex::new(AppState { stream_sessions: Arc::new(DashMap::new()), + webrtc_proxy: proxy.clone(), })); let (offer_tx, offer_rx) = tokio::sync::mpsc::channel::<(i32, i32, String)>(64); // Request_Id, @@ -109,14 +124,11 @@ async fn main() -> Result<(), Box> { }), } .into(), + info: Arc::new(ServerInfo { + version: SERVER_VERSION, + start_time: std::time::Instant::now(), + }), }; - let proxyconfig = WebRtcProxyConfig { - proxy_port: env::var("RTC_PORT") - .unwrap_or("6969".into()) - .parse() - .expect("RTC_PORT needs to be a number (i32)"), - }; - let proxy = webrtc_proxy::WebrtcProxy::new(proxyconfig).await.unwrap(); let app = appstate.lock().await; let webrtc = webrtc::Webrtc { @@ -124,7 +136,8 @@ async fn main() -> Result<(), Box> { accept_tx: answer_tx, sessions_ref: app.stream_sessions.clone(), db: db.clone(), - proxy: proxy.clone().into(), + proxy: proxy.clone().into(), // The entire of the proxy's internal are within an arc, cheap + // to clone }; let rtmp = rtmp::Rtmp { diff --git a/crates/server/src/rtmp.rs b/crates/server/src/rtmp.rs index 81f9ee3..85f4fa3 100644 --- a/crates/server/src/rtmp.rs +++ b/crates/server/src/rtmp.rs @@ -34,12 +34,11 @@ pub struct Rtmp { async fn write_outbound(socket: &mut TcpStream, results: Vec) { for r in results { - if let ServerSessionResult::OutboundResponse(p) = r { - if let Err(e) = socket.write_all(&p.bytes).await { + if let ServerSessionResult::OutboundResponse(p) = r + && let Err(e) = socket.write_all(&p.bytes).await { warn!("RTMP write error: {e}"); return; } - } } } @@ -96,8 +95,8 @@ impl Rtmp { x => return Err(format!("unexpected handshake state: {:?}", x).into()), } - let (rtmp_session, init_bytes) = - ServerSession::new(ServerSessionConfig::new()).map_err(|e| format!("ServerSession::new failed: {e}"))?; + let (rtmp_session, init_bytes) = ServerSession::new(ServerSessionConfig::new()) + .map_err(|e| format!("ServerSession::new failed: {e}"))?; write_outbound(&mut socket, init_bytes).await; Ok((rtmp_session, socket)) @@ -127,8 +126,8 @@ impl Rtmp { } }; info!(%peer_addr, "RTMP handshake complete"); - let (mut video_tx, mut video_rx) = broadcast::>(32); - let (mut audio_tx, mut audio_rx) = broadcast::>(32); + let (mut video_tx, video_rx) = broadcast::>(32); + let (mut audio_tx, audio_rx) = broadcast::>(32); // video_rx.cycle video_tx.set_overflow(true); @@ -215,7 +214,9 @@ impl Rtmp { let reply = match session.accept_request(request_id) { Ok(r) => r, Err(e) => { - warn!("Failed to accept connection request {request_id}: {e}"); + warn!( + "Failed to accept connection request {request_id}: {e}" + ); return; } }; @@ -235,7 +236,11 @@ impl Rtmp { key } else { warn!(stream_key = %stream_key, "stream key not found, rejecting"); - let reply = match session.reject_request(request_id, "", "Stream key invalid") { + let reply = match session.reject_request( + request_id, + "", + "Stream key invalid", + ) { Ok(r) => r, Err(e) => { warn!("Failed to reject stream key request: {e}"); @@ -255,10 +260,16 @@ impl Rtmp { Ok(s) => s, Err(e) => { warn!("DB error checking active stream: {e}"); - let reply = match session.reject_request(request_id, "", "Internal error") { + let reply = match session.reject_request( + request_id, + "", + "Internal error", + ) { Ok(r) => r, Err(e2) => { - warn!("Failed to reject request on DB error: {e2}"); + warn!( + "Failed to reject request on DB error: {e2}" + ); return; } }; @@ -294,6 +305,9 @@ impl Rtmp { audio_channel: audio_tx.clone(), codec: None, started_at: Utc::now(), + active_clients: 0.into(), + video_pt: None, + video_profile_level_id: None, }, ); @@ -312,7 +326,11 @@ impl Rtmp { .await { warn!("Failed to create stream session record: {e}"); - let _ = session.reject_request(request_id, "", "Internal error"); + let _ = session.reject_request( + request_id, + "", + "Internal error", + ); return; } stream_id = Some(key.id); @@ -343,13 +361,12 @@ impl Rtmp { } => match Self::parse_video_codec(&data) { Ok(codec) => { if !codec_stamped { - if let Some(id) = current_stream_key_id { - if let Some(mut session) = + if let Some(id) = current_stream_key_id + && let Some(mut session) = stream_sessions.get_mut(&id) { session.codec = Some(codec.clone()); } - } codec_stamped = true; } match codec { diff --git a/crates/server/src/webrtc.rs b/crates/server/src/webrtc.rs index 95d7538..b182d55 100644 --- a/crates/server/src/webrtc.rs +++ b/crates/server/src/webrtc.rs @@ -59,39 +59,68 @@ impl Webrtc { let local_addr = self.proxy.public_addr(); - let stream_codec = self - .sessions_ref - .get(&stream_id) - .and_then(|s| s.codec.clone()); + let session = self.sessions_ref.get(&stream_id); + let stream_codec = session.as_ref().and_then(|s| s.codec.clone()); + let video_pt = session.as_ref().and_then(|s| s.video_pt); + let video_profile = session.as_ref().and_then(|s| s.video_profile_level_id); + drop(session); + info!(request_id, stream_id, codec = ?match &stream_codec { Some(StreamCodec::H264) => "H264", Some(StreamCodec::H265) => "H265", Some(StreamCodec::AV1) => "AV1", None => "unknown", - }, "configuring RTC codec"); + }, video_pt, video_profile, "configuring RTC codec"); let mut builder = Rtc::builder(); { + use str0m::format::{Codec, FormatParams}; + use str0m::media::Frequency; + let cc = builder.codec_config(); cc.clear(); cc.enable_opus(true); - match stream_codec { - Some(StreamCodec::H265) => { + + // When video_pt is set (WHIP-negotiated), enable only that + // exact codec+PT+profile. Otherwise fall back to enabling all + // common profiles for the codec family (RTMP ingest doesn't + // carry profile info, so we cast a wide net for browser compat). + match (stream_codec.clone(), video_pt, video_profile) { + // --- Specific PT + profile (WHIP path) --- + (Some(StreamCodec::H264), Some(pt), Some(profile)) => { + cc.add_h264(pt.into(), None, true, profile); + } + (Some(StreamCodec::H264), Some(pt), None) => { + // PT known, profile unknown — default to Constrained + // Baseline (works everywhere but may mismatch for + // Main/High streams from OBS). + cc.add_h264(pt.into(), None, true, 0x42e01f); + } + (Some(StreamCodec::H265), Some(pt), _) => { + cc.add_h265(pt.into(), None, 1, 0, 180); + } + (Some(StreamCodec::AV1), Some(pt), _) => { + cc.add_config( + pt.into(), + None, + Codec::Av1, + Frequency::NINETY_KHZ, + None, + FormatParams::default(), + ); + } + + // --- Broad codec family (RTMP path, no WHIP info) --- + (Some(StreamCodec::H265), None, _) => { cc.enable_h265(true); } - Some(StreamCodec::AV1) => { + (Some(StreamCodec::AV1), None, _) => { cc.enable_av1(true); } _ => { - // 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 + cc.add_h264(102.into(), None, true, 0x42e01f); + cc.add_h264(104.into(), None, true, 0x4d001f); + cc.add_h264(106.into(), None, true, 0x64001f); } } } @@ -246,16 +275,16 @@ impl Webrtc { warn!(stream_id, mid = ?ma.mid, "no writer for video mid"); } } - if ma.kind == MediaKind::Audio { - if let Some(writer) = rtc.writer(ma.mid) { - let best = writer.payload_params().max_by_key(|p| { - p.spec().format.profile_level_id.unwrap_or(0) - }); - if let Some(params) = best { - info!(stream_id, pt = ?params.pt(), "selected audio PT"); - audio_pt = Some(params.pt()); - audio_mid = Some(ma.mid); - } + if ma.kind == MediaKind::Audio + && let Some(writer) = rtc.writer(ma.mid) + { + let best = writer + .payload_params() + .max_by_key(|p| p.spec().format.profile_level_id.unwrap_or(0)); + if let Some(params) = best { + info!(stream_id, pt = ?params.pt(), "selected audio PT"); + audio_pt = Some(params.pt()); + audio_mid = Some(ma.mid); } } } @@ -292,27 +321,25 @@ impl Webrtc { ); } } - if let Some(video) = &video_stream { - if video.is_closed() { - if let Some(session) = sessions_ref.get(&stream_id) { - video_stream = Some(session.frame_channel.new_receiver()); - debug!(stream_id, "subscribed to video channel"); - } - } + if let Some(video) = &video_stream + && video.is_closed() + && let Some(session) = sessions_ref.get(&stream_id) + { + video_stream = Some(session.frame_channel.new_receiver()); + debug!(stream_id, "subscribed to video channel"); } - if audio_stream.is_none() { - if let Some(session) = sessions_ref.get(&stream_id) { - audio_stream = Some(session.audio_channel.new_receiver()); - debug!(stream_id, "subscribed to audio channel"); - } + if audio_stream.is_none() + && let Some(session) = sessions_ref.get(&stream_id) + { + audio_stream = Some(session.audio_channel.new_receiver()); + debug!(stream_id, "subscribed to audio channel"); } - if let Some(audio) = &audio_stream { - if audio.is_closed() { - if let Some(session) = sessions_ref.get(&stream_id) { - audio_stream = Some(session.audio_channel.new_receiver()); - debug!(stream_id, "subscribed to audio channel"); - } - } + if let Some(audio) = &audio_stream + && audio.is_closed() + && let Some(session) = sessions_ref.get(&stream_id) + { + audio_stream = Some(session.audio_channel.new_receiver()); + debug!(stream_id, "subscribed to audio channel"); } } @@ -333,9 +360,9 @@ impl Webrtc { } } result = rx.recv() => { - if let Some((data, from)) = result { - if let Ok(contents) = (&data[..]).try_into() { - if let Err(e) = rtc.handle_input(Input::Receive( + if let Some((data, from)) = result + && let Ok(contents) = (&data[..]).try_into() + && let Err(e) = rtc.handle_input(Input::Receive( Instant::now(), Receive { proto: Protocol::Udp, @@ -347,8 +374,6 @@ impl Webrtc { error!("handle_input(Receive) error: {:?}", e); return; } - } - } } res = Webrtc::recv_video(&mut video_stream), if video_stream.is_some() => { match res { @@ -475,10 +500,10 @@ impl Webrtc { ) { 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); - } + if let (Some(pt), Some(writer)) = (audio_pt, audio_mid.and_then(|m| rtc.writer(m))) + && let Err(e) = writer.write(pt, now, rtp_time, frame.data.to_vec()) + { + warn!("RTP write error: {:?}", e); } } diff --git a/crates/server/src/webrtc_ingest.rs b/crates/server/src/webrtc_ingest.rs index 69e27b1..627426a 100644 --- a/crates/server/src/webrtc_ingest.rs +++ b/crates/server/src/webrtc_ingest.rs @@ -1,17 +1,561 @@ -use std::sync::Arc; +use std::{net::SocketAddr, sync::Arc, time::Instant}; -use axum::{extract::State, response::IntoResponse}; -use str0m::change::SdpOffer; -use tracing::info; +use async_broadcast::broadcast; +use axum::{ + extract::{ConnectInfo, Path, State}, + http::{HeaderMap, StatusCode, header}, + response::IntoResponse, +}; +use bytes::Bytes; +use chrono::{Local, Utc}; +use dashmap::DashMap; +use entity::{stream_key, stream_session}; +use sea_orm::{DatabaseConnection, IntoActiveModel}; +use str0m::{ + Candidate, Event, Input, Output, Rtc, + change::SdpOffer, + media::{MediaKind, Mid}, + net::{Protocol, Receive}, +}; +use tokio::{net::UdpSocket, sync::mpsc::Receiver}; +use tracing::{debug, error, info, warn}; -use crate::http::HttpServer; +use crate::{ + StreamSession, audio::OpusAudioFrame, codec::VideoFrame, http::HttpServer, + http_error::HttpError, +}; + +pub async fn handle_whip_injest_delete( + State(state): State>, + ConnectInfo(remote): ConnectInfo, + headers: HeaderMap, +) -> Result { + let token = headers + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(|s| s.trim().to_string()); + let Some(token) = token else { + warn!("Whip: missing bearer token from {}", remote); + return Err(HttpError::Unauthorized); + }; + + let key = stream_key::Entity::find_by_key(&state.db, &token) + .await? + .ok_or(HttpError::Unauthorized)?; + + let active_session = stream_session::Model::get_active_by_stream_key_id(&state.db, key.id) + .await? + .ok_or(HttpError::NotFound)?; + + active_session + .into_active_model() + .finish_stream_session(&state.db, Utc::now()) + .await?; + + Ok(StatusCode::OK) +} + +/// PATCH /api/whip/{id} — Trickle ICE candidate delivery. +pub async fn handle_whip_injest_patch( + State(state): State>, + Path(slug): Path, + headers: HeaderMap, + body: String, +) -> Result { + // The slug is the stream_key_id (set by the POST handler's Location header). + let stream_key_id: i32 = slug + .parse() + .map_err(|e| { +warn!(%slug, "Whip PATCH: bad slug: {:?}", e); +HttpError::NotFound + })?; + + info!(stream_key_id, ct = ?headers.get(header::CONTENT_TYPE), "Whip PATCH: trickle candidate"); + + // Look up the trickle sender for this session. + let trickle_map = state + .appstate + .lock() + .await + .webrtc_proxy + .trickle_tx + .clone(); + let tx = trickle_map + .get(&stream_key_id) + .ok_or_else(|| { +warn!(stream_key_id, "Whip PATCH: no trickle channel for session"); +HttpError::NotFound + })?; + + debug!(stream_key_id, %body, "Whip PATCH: forwarding trickle candidate"); + tx.send(body).ok(); + + Ok(StatusCode::NO_CONTENT) +} pub async fn handle_whip_injest( State(state): State>, + ConnectInfo(remote): ConnectInfo, + headers: HeaderMap, offer: String, -) -> impl IntoResponse { - let sdp_offer = SdpOffer::from_sdp_string(&offer).unwrap(); - for x in &sdp_offer.media_lines { - info!("{}", x); +) -> axum::response::Response { + let err = |status, msg: &str| -> axum::response::Response { + ( + status, + [(header::CONTENT_TYPE, "text/plain")], + msg.to_string(), + ) + .into_response() + }; + let public_addr = state.appstate.lock().await.webrtc_proxy.public_addr(); + + let token = headers + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(|s| s.trim().to_string()); + let Some(token) = token else { + warn!("Whip: missing bearer token from {}", remote); + return err(StatusCode::UNAUTHORIZED, "missing bearer token"); + }; + + let key = match stream_key::Entity::find_by_key(&state.db, &token).await { + Ok(Some(key)) => key, + Ok(None) => { + warn!("Whip: stream key not found, rejecting {}", remote); + return err(StatusCode::UNAUTHORIZED, "invalid stream key"); + } + Err(e) => { + error!("Whip: DB error looking up stream key: {:?}", e); + return err(StatusCode::INTERNAL_SERVER_ERROR, "db error"); + } + }; + + match stream_session::Model::get_active_by_stream_key_id(&state.db, key.id).await { + Ok(Some(_)) => { + warn!(stream_key_id = key.id, label = %key.label, "Whip: stream key already live, rejecting duplicate publish from {}", remote); + return err(StatusCode::CONFLICT, "stream already live"); + } + Ok(None) => {} + Err(e) => { + error!("Whip: DB error checking active stream: {:?}", e); + return err(StatusCode::INTERNAL_SERVER_ERROR, "db error"); + } + }; + info!(stream_key_id = key.id, label = %key.label, "Whip authenticated stream key from {}", remote); + + // Parse the SDP offer first so we can detect the codec and + // configure the Rtc before accepting. + let sdp_offer = match SdpOffer::from_sdp_string(&offer) { + Ok(o) => o, + Err(e) => { + error!("Whip: cant parse offer from {}: {:?}", remote, e); + return err(StatusCode::BAD_REQUEST, "invalid SDP offer"); + } + }; + info!( + "Whip offer from {} — {} media line(s):", + remote, + sdp_offer.media_lines.len() + ); + for x in &sdp_offer.media_lines { + info!(" {}", x); + } + + // Detect codec from the offer so we can enable matching codecs. + let stream_codec = video_codec_from_sdp_offer(&sdp_offer); + info!(?stream_codec, "detected codec from WHIP offer"); + + // Build Rtc with ICE-Lite (required by WHIP RFC 9728 §4.1) and + // matching codecs enabled. + // Not using ICE-Lite: full ICE lets the server initiate checks + // when OBS hasn't sent its candidates yet (Trickle ICE without PATCH). + let mut builder = Rtc::builder(); + { + let cc = builder.codec_config(); + cc.clear(); + cc.enable_opus(true); + match &stream_codec { + Some(crate::StreamCodec::H264) => { + info!("Whip: enabling H.264 codec"); + cc.enable_h264(true); + } + Some(crate::StreamCodec::H265) => { + info!("Whip: enabling H.265 codec"); + cc.enable_h265(true); + } + Some(crate::StreamCodec::AV1) => { + info!("Whip: enabling AV1 codec"); + cc.enable_av1(true); + } + None => { + warn!("Whip: no video codec detected in offer, enabling H.264 as fallback"); + cc.enable_h264(true); + } + } + } + let mut rtc = builder.build(Instant::now()); + let candidate = Candidate::host(public_addr, Protocol::Udp).unwrap(); + rtc.add_local_candidate(candidate); + info!(%public_addr, "Whip: added local ICE candidate, accepting offer…"); + + let offer_answe = match rtc.sdp_api().accept_offer(sdp_offer) { + Ok(a) => a, + Err(e) => { + error!("cant accept inject offer: {:?}", e); + return err(StatusCode::BAD_REQUEST, "could not accept offer"); + } + }; + + // OBS sends no a=candidate: lines (disableAutoGathering). Derive a + // remote host candidate from the HTTP source address so the server + // has somewhere to send STUN checks. + if let Ok(c) = Candidate::host(remote, Protocol::Udp) { + info!(%remote, "Whip: no candidates in offer, adding HTTP-derived remote host candidate"); + rtc.add_remote_candidate(c); + } + + // Extract negotiated codec, PT, and profile from the answer + // so WHEP viewers can use the exact same codec config. + let (negotiated_codec, video_pt, video_profile) = + extract_negotiated_codec_info(&offer_answe); + info!( + ?negotiated_codec, + video_pt, + ?video_profile, + "negotiated codec from WHIP answer" + ); + if negotiated_codec.is_none() || video_pt.is_none() { + warn!("Whip: no common video codec negotiated, rejecting"); + return err(StatusCode::NOT_ACCEPTABLE, "no common video codec"); + } + + let answer_sdp = offer_answe.to_sdp_string() + // Strip a=ice-options:trickle so OBS starts ICE immediately. + .replace("a=ice-options:trickle\r\n", "") + .replace("a=ice-options:trickle\n", ""); + // Fix up the answer for libdatachannel (OBS WHIP): + // 1. Strip a=group:BUNDLE — OBS doesn't negotiate it. + // 2. Add the host candidate to the video m= line. + let answer_sdp = answer_sdp + .replace("a=group:BUNDLE 0 1\r\n", "") + .replace("a=group:BUNDLE 0 1\n", ""); + let answer_sdp = if let Some(cand_line) = answer_sdp + .lines() + .find(|l| l.starts_with("a=candidate:")) + { + // Insert the candidate line after the video m= line. + let cand_replacement = format!("m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 127.0.0.1\r\n{}\r\n", cand_line); + answer_sdp.replace("m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 0.0.0.0\r\n", &cand_replacement) + } else { + answer_sdp + }; + info!("Serving Whip SDP answer to {}:\n{}", remote, answer_sdp); + + let ufrag = answer_sdp + .lines() + .find(|l| l.starts_with("a=ice-ufrag:")) + .and_then(|l| l.strip_prefix("a=ice-ufrag:")) + .map(|s| s.trim().to_string()); + let Some(ufrag) = ufrag else { + error!("Whip: answer has no a=ice-ufrag"); + return err(StatusCode::INTERNAL_SERVER_ERROR, "no ice-ufrag in answer"); + }; + let ice_pwd = answer_sdp + .lines() + .find(|l| l.starts_with("a=ice-pwd:")) + .and_then(|l| l.strip_prefix("a=ice-pwd:")) + .map(|s| s.trim().to_string()); + info!(?ufrag, ?ice_pwd, "Whip: registering ICE credentials with proxy"); + let (socket, rx) = state.appstate.lock().await.webrtc_proxy.add_client(ufrag); + + let stream_sessions = state.appstate.lock().await.stream_sessions.clone(); + let (video_tx, video_rx) = broadcast::>(4); + let (audio_tx, audio_rx) = broadcast::>(4); + + stream_sessions.insert( + key.id, + StreamSession { + stream_key_id: key.id, + stream_key_label: key.label, + frame_channel: video_tx, + audio_channel: audio_tx, + codec: negotiated_codec, + started_at: Utc::now(), + active_clients: 0.into(), + video_pt, + video_profile_level_id: video_profile, + }, + ); + info!(stream_key_id = key.id, "Whip: StreamSession inserted, spawning detach task"); + + // Trickle-ICE channel: OBS can send candidates via PATCH after the + // initial offer. We forward them to the Rtc task for add_remote_candidate. + let (trickle_tx, trickle_rx) = tokio::sync::mpsc::unbounded_channel(); + let trickle_map = state.appstate.lock().await.webrtc_proxy.trickle_tx.clone(); + trickle_map.insert(key.id, trickle_tx); + + let db = state.db.clone(); + tokio::spawn(async move { + detach_inject_rtc( + stream_sessions, + key.id, + db, + socket, + rx, + trickle_rx, + rtc, + public_addr, + video_rx, + audio_rx, + ) + .await; + // Clean up trickle channel when done. + trickle_map.remove(&key.id); + }); + + stream_session::Model::create_stream_session(&state.db, key.id, Local::now().into()) + .await + .ok(); + + let location = format!("/api/whip/{}", key.id); + axum::response::Response::builder() + .status(StatusCode::CREATED) + .header(header::CONTENT_TYPE, "application/sdp") + .header(header::LOCATION, &location) + // Provide a STUN server via Link header so OBS can gather + // ICE candidates even without explicit STUN configuration. + .header(header::LINK, "; rel=\"ice-server\"") + .body(axum::body::Body::from(answer_sdp)) + .unwrap() +} + +async fn detach_inject_rtc( + sessions_ref: Arc>, + stream_key_id: i32, + db: DatabaseConnection, + socket: Arc, + mut rx: Receiver<(Bytes, SocketAddr)>, + mut trickle_rx: tokio::sync::mpsc::UnboundedReceiver, + mut rtc: Rtc, + local_addr: SocketAddr, + video_rx: async_broadcast::Receiver>, + audio_rx: async_broadcast::Receiver>, +) { + let cleanup = || { + let sessions_ref = sessions_ref.clone(); + let db = db.clone(); + async move { + sessions_ref.remove(&stream_key_id); + if let Ok(Some(s)) = + stream_session::Model::get_active_by_stream_key_id(&db, stream_key_id).await + { + s.into_active_model() + .finish_stream_session(&db, Local::now().into()) + .await + .ok(); + } + } + }; + let mut video_mid: Option = None; + let mut audio_mid: Option = None; + let mut video_tx: Option>> = None; + let mut audio_tx: Option>> = None; + let mut _connected = false; + + loop { + // Blankly using them so they don't drop (like RTMP). + let _ = video_rx.is_closed(); + let _ = audio_rx.is_closed(); + + let deadline = loop { + match rtc.poll_output() { + Ok(Output::Timeout(t)) => break t, + Ok(Output::Transmit(t)) => { + debug!( + "Whip TX: {} bytes → {}:{}", + t.contents.len(), + t.destination.ip(), + t.destination.port() + ); + if let Err(e) = socket.send_to(&t.contents, t.destination).await { + warn!("Whip UDP send error: {:?}", e); + cleanup().await; + return; + } + } + Ok(Output::Event(e)) => match e { + Event::MediaAdded(ma) => { + info!(stream_key_id, kind = ?ma.kind, mid = ?ma.mid, "Whip MediaAdded"); + if ma.kind == MediaKind::Video { + video_mid = Some(ma.mid); + } + if ma.kind == MediaKind::Audio { + audio_mid = Some(ma.mid); + } + } + Event::MediaData(md) => { + // str0m depacketizes RTP into full codec frames (no decode). + // H.264/H.265 → Annex-B, AV1 → OBU, Opus → raw Opus packets. + // Same format as our RTMP codec parsers produce — push directly. + if Some(md.mid) == video_mid { + let ts_ms = (md.time.as_seconds() * 1000.0) as u32; + let frame = VideoFrame { + data: Bytes::copy_from_slice(&md.data), + is_keyframe: md.is_keyframe(), + timestamp_ms: ts_ms, + }; + if let Some(tx) = &video_tx { + tx.broadcast(Arc::new(frame)).await.ok(); + } + } + if Some(md.mid) == audio_mid { + let ts_ms = (md.time.as_seconds() * 1000.0) as u32; + let frame = OpusAudioFrame { + data: Bytes::copy_from_slice(&md.data), + timestamp_ms: ts_ms, + }; + if let Some(tx) = &audio_tx { + tx.broadcast(Arc::new(frame)).await.ok(); + } + } + } + Event::IceConnectionStateChange(state) => { + info!(stream_key_id, ?state, "Whip ICE state change"); + if matches!(state, str0m::IceConnectionState::Disconnected) { + info!("Whip ICE disconnected, closing connection"); + cleanup().await; + return; + } + } + Event::Connected => { + info!(stream_key_id, "Whip DTLS+ICE connected, wiring broadcast channels"); + if let Some(session) = sessions_ref.get(&stream_key_id) { + video_tx = Some(session.frame_channel.clone()); + audio_tx = Some(session.audio_channel.clone()); + } + _connected = true; + } + _ => {} + }, + Err(e) => { + error!("Whip poll_output error (closing): {:?}", e); + cleanup().await; + return; + } + } + }; + + let sleep = tokio::time::sleep_until(deadline.max(Instant::now()).into()); + tokio::select! { + _ = sleep => { + if let Err(e) = rtc.handle_input(Input::Timeout(Instant::now())) { + error!(stream_key_id, "Whip handle_input(Timeout) error: {:?}", e); + cleanup().await; + return; + } + } + // Trickle-ICE candidates from PATCH /api/whip/{id} + Some(candidate_line) = trickle_rx.recv() => { + info!(stream_key_id, %candidate_line, "Whip: received trickle candidate"); + // Parse the candidate string (without "a=candidate:" prefix if present). + let cand_str = candidate_line + .strip_prefix("a=candidate:") + .unwrap_or(&candidate_line); + match str0m::Candidate::from_sdp_string(cand_str) { + Ok(c) => { + info!(stream_key_id, "Whip: adding remote candidate"); + rtc.add_remote_candidate(c); + } + Err(e) => { + warn!(stream_key_id, %cand_str, "Whip: bad trickle candidate: {:?}", e); + } + } + } + result = rx.recv() => { + let Some((data, from)) = result else { + info!(stream_key_id, "Whip proxy channel closed, cleaning up"); + cleanup().await; + return; + }; + debug!( + stream_key_id, + len = data.len(), + %from, + "Whip RX: {} bytes", + data.len() + ); + if let Ok(contents) = (&data[..]).try_into() + && let Err(e) = rtc.handle_input(Input::Receive( + Instant::now(), + Receive { + proto: Protocol::Udp, + source: from, + destination: local_addr, + contents, + }, + )) { + error!("Whip handle_input(Receive) error: {:?}", e); + cleanup().await; + return; + } + } + } } } + +/// Extract the negotiated video codec, PT, and profile_level_id from an SDP answer. +/// +/// Walks the answer's media lines, finds the first video m-line with +/// negotiated rtp_params, and returns the codec + PT + H.264 profile. +pub fn extract_negotiated_codec_info( +answer: &str0m::change::SdpAnswer, +) -> (Option, Option, Option) { +use str0m::format::Codec; + +for line in answer.media_lines.iter() { +// Iterate rtp_params on every m-line; video check via +// codec.is_video() avoids needing str0m's private MediaType. +for p in line.rtp_params() { +if p.spec().codec.is_video() { +let pt = Some(*p.pt()); +let profile = p.spec().format.profile_level_id; +let codec = match p.spec().codec { +Codec::H264 => Some(crate::StreamCodec::H264), +Codec::H265 => Some(crate::StreamCodec::H265), +Codec::Av1 => Some(crate::StreamCodec::AV1), +_ => None, +}; +return (codec, pt, profile); +} +} +} +(None, None, None) +} + +/// Extract the video codec from an SDP offer's media lines. +/// +/// Walks every m= line's `rtp_params()`, maps str0m's `Codec` enum +/// to our `StreamCodec`. Returns the first video codec found. +pub fn video_codec_from_sdp_offer( + sdp_offer: &str0m::change::SdpOffer, +) -> Option { + use str0m::format::Codec; + + // SdpOffer derefs to Sdp, which has pub media_lines. + // MediaLine is pub(crate) in str0m, but we can call its pub methods. + for line in sdp_offer.media_lines.iter() { + for p in line.rtp_params() { + if p.spec().codec.is_video() { + return match p.spec().codec { + Codec::H264 => Some(crate::StreamCodec::H264), + Codec::H265 => Some(crate::StreamCodec::H265), + Codec::Av1 => Some(crate::StreamCodec::AV1), + _ => None, + }; + } + } + } + None +} diff --git a/crates/server/src/webrtc_proxy.rs b/crates/server/src/webrtc_proxy.rs index e3791f2..6b1250c 100644 --- a/crates/server/src/webrtc_proxy.rs +++ b/crates/server/src/webrtc_proxy.rs @@ -7,7 +7,6 @@ use std::{ use bytes::Bytes; use dashmap::DashMap; -use str0m::net::DatagramRecv; use tokio::{ net::UdpSocket, sync::mpsc::{self, Receiver}, @@ -24,6 +23,10 @@ pub struct WebrtcProxy { clients_addr: Arc>>, socket: Arc, public_addr: SocketAddr, + /// Trickle-ICE candidate channels for WHIP ingest. + /// Keyed by stream_key_id; sender stored here so PATCH handler + /// can forward candidates to the detach task. + pub trickle_tx: Arc>>, } const STUN_MAGIC: u32 = 0x2112A442; @@ -61,6 +64,7 @@ impl WebrtcProxy { clients_ufrag: Arc::new(DashMap::new()), clients_addr: Arc::new(DashMap::new()), public_addr, + trickle_tx: Arc::new(DashMap::new()), }) } pub async fn run(self) { @@ -79,8 +83,9 @@ impl WebrtcProxy { }; let data = Bytes::copy_from_slice(&buf[..b]); - // By addr - if let Some(tx) = by_addr.get(&from) { + // By addr + if let Some(tx) = by_addr.get(&from) { + debug!("proxy: routing {} bytes by addr {}:{} → channel", b, from.ip(), from.port()); match tx.try_send((data, from)) { Ok(_) => continue, Err(e) => { @@ -97,18 +102,24 @@ impl WebrtcProxy { }; }; - let Some(ufrag) = self::WebrtcProxy::ufrag(&data) else { + let Some((part1, part2)) = self::WebrtcProxy::ufrag_pair(&data) else { debug!("huh, packet isnt stun or added as client."); continue; }; - - let Some((_, tx)) = by_ufrag.remove(&ufrag) else { - // warn!("STUN packet ({}), isnt registored", ufrag); + // Try both parts of the STUN username — the first packet + // might be a response to OUR STUN request (remote:local) + // or an incoming request from the remote peer (local:remote). + let part2_lookup = part2.clone(); + let entry = by_ufrag.remove(&part1).or_else(|| { + part2_lookup.and_then(|p2| by_ufrag.remove(&p2)) + }); + let Some((_, tx)) = entry else { + warn!("STUN packet ({}/{:?}), isnt registored", part1, part2); continue; }; by_addr.insert(from, tx.clone()); - debug!("got ufrag {}", ufrag); + info!("proxy: STUN match → promoted {} → ufrag={} (match was {}/{})", from, part1, part1, part2.as_deref().unwrap_or("-")); debug!("sending data"); if let Err(e) = tx.try_send((data, from)) { match e { @@ -136,29 +147,34 @@ impl WebrtcProxy { pub fn public_addr(&self) -> SocketAddr { self.public_addr } - pub fn ufrag(b: &Bytes) -> Option { - if b.len() <= 20 { - return None; - } - let magic = u32::from_be_bytes(b[4..8].try_into().ok()?); - if magic != STUN_MAGIC { - return None; - } - // attribies start at 20 - let mut pos = 20usize; - while (pos + 4) <= b.len() { - let attr_type: u16 = u16::from_be_bytes(b[pos..pos + 2].try_into().ok()?); - let attr_len: u16 = u16::from_be_bytes(b[pos + 2..pos + 4].try_into().ok()?); - pos = pos + 4; - if attr_type == 0x0006 { - let value = std::str::from_utf8(b[pos..pos + (attr_len as usize)].try_into().ok()?); - let local = value.unwrap().split(":").next(); - return Some(local.unwrap().to_string()); + pub fn ufrag_pair(b: &Bytes) -> Option<(String, Option)> { + if b.len() <= 20 { + return None; } - pos += (attr_len as usize + 3) & !3; + let magic = u32::from_be_bytes(b[4..8].try_into().ok()?); + if magic != STUN_MAGIC { + return None; + } + // attribies start at 20 + let mut pos = 20usize; + while (pos + 4) <= b.len() { + let attr_type: u16 = u16::from_be_bytes(b[pos..pos + 2].try_into().ok()?); + let attr_len: u16 = u16::from_be_bytes(b[pos + 2..pos + 4].try_into().ok()?); + pos += 4; + if attr_type == 0x0006 { + let value = std::str::from_utf8( + b[pos..pos + (attr_len as usize)].try_into().ok()?, + ) + .ok()?; + let mut parts = value.split(':'); + let first = parts.next()?.to_string(); + let second = parts.next().map(|s| s.to_string()); + return Some((first, second)); + } + pos += (attr_len as usize + 3) & !3; + } + None } - None - } } async fn resolve_domain(domain: &str) -> Result> { diff --git a/flake.lock b/flake.lock index 162bb82..e269dc5 100644 --- a/flake.lock +++ b/flake.lock @@ -2,11 +2,11 @@ "nodes": { "crane": { "locked": { - "lastModified": 1780532242, - "narHash": "sha256-D+BsdpxmtUwtqGoY0IXPhHgTlmqgcZKCEo1oMyn7ep0=", + "lastModified": 1785284101, + "narHash": "sha256-ghcXEpYEM4a7pbEkoqbn8c0ptJJqgGzuFiG3T6W5g4I=", "owner": "ipetkov", "repo": "crane", - "rev": "59a82a1222dd3b2080b5cc52a1a2e8d5f1b77f37", + "rev": "756d6d07c3818ea95d1e2cdac63fa7d02fe3e61b", "type": "github" }, "original": { @@ -35,11 +35,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1781454065, - "narHash": "sha256-d2xfDjnfRuf/xYGdu9VVRHiav/2w5hDL/5cw2TuVAXw=", + "lastModified": 1785301185, + "narHash": "sha256-eoS3KQTO0aPWXZvIaRbRAzSSHW3l5wdMFXtT1ISfoKA=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "9eac87a12312b8f60dd52e1c6e1a265f6fc7f5fc", + "rev": "9bc02893134c733dd85de46ee4fb2fac696b5529", "type": "github" }, "original": {