Cleanup: over-engineering pass
- merge /api/uptime + /api/version into /api/health and /api/stats - collapse RTMP per-codec dispatch into a single CodecParser path - dedupe STUN attribute walking (ufrag_pair / parse_xor_mapped_address) - dedupe StreamCodec <-> str0m Codec mapping via StreamCodec::from_str0m - dedupe bearer token extraction in WHIP handlers - replace regex charset validation with stdlib checks - flatten one-field wrappers (WebRtcProxyConfig, HttpServerConfig, ServerInfo) - remove SessionCookie extractor (only fed debug logs) - delete dead code: entity ActiveModel helpers, get_stream_session, WebrtcProxy::local_addr, _connected flag, commented-out blocks - drop unused deps: futures, rand, regex, serde_json
This commit is contained in:
Generated
-18
@@ -1151,7 +1151,6 @@ checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"futures-executor",
|
||||
"futures-io",
|
||||
"futures-sink",
|
||||
"futures-task",
|
||||
@@ -1202,17 +1201,6 @@ version = "0.3.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
|
||||
|
||||
[[package]]
|
||||
name = "futures-macro"
|
||||
version = "0.3.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-sink"
|
||||
version = "0.3.32"
|
||||
@@ -1231,10 +1219,8 @@ version = "0.3.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"futures-io",
|
||||
"futures-macro",
|
||||
"futures-sink",
|
||||
"futures-task",
|
||||
"memchr",
|
||||
@@ -2916,16 +2902,12 @@ dependencies = [
|
||||
"chrono",
|
||||
"dashmap",
|
||||
"entity",
|
||||
"futures",
|
||||
"migration",
|
||||
"opus",
|
||||
"rand 0.10.2",
|
||||
"regex",
|
||||
"rml_rtmp",
|
||||
"rubato",
|
||||
"sea-orm",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"str0m",
|
||||
"symphonia",
|
||||
"sysinfo",
|
||||
|
||||
@@ -91,27 +91,3 @@ impl Entity {
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
impl ActiveModel {
|
||||
pub async fn change_label_value(
|
||||
mut self,
|
||||
db: &DatabaseConnection,
|
||||
value: String,
|
||||
) -> Result<Model, DbErr> {
|
||||
self.label = Set(value);
|
||||
self.update(db).await
|
||||
}
|
||||
|
||||
pub async fn password(
|
||||
mut self,
|
||||
db: &DatabaseConnection,
|
||||
value: String,
|
||||
) -> Result<Model, DbErr> {
|
||||
self.password = if value.is_empty() {
|
||||
Set(None)
|
||||
} else {
|
||||
Set(Some(value))
|
||||
};
|
||||
self.update(db).await
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,17 +50,6 @@ impl Model {
|
||||
.insert(db)
|
||||
.await
|
||||
}
|
||||
pub async fn get_stream_session(
|
||||
db: &DatabaseConnection,
|
||||
stream_session_id: i32,
|
||||
) -> Result<Model, DbErr> {
|
||||
Entity::find_by_id(stream_session_id)
|
||||
.one(db)
|
||||
.await?
|
||||
.ok_or(DbErr::RecordNotFound(format!(
|
||||
"stream_session {stream_session_id}"
|
||||
)))
|
||||
}
|
||||
pub async fn get_all_active_sessions(db: &DatabaseConnection) -> Result<Vec<Model>, DbErr> {
|
||||
Ok(Entity::find()
|
||||
.filter(Column::EndedAt.is_null())
|
||||
|
||||
@@ -15,20 +15,17 @@ path = "src/main.rs"
|
||||
async-broadcast = "0.7.2"
|
||||
bytes = "1"
|
||||
dashmap = "6.2.1"
|
||||
rand = "0.10.1"
|
||||
rml_rtmp = "0.8.0"
|
||||
str0m = "0.20.0"
|
||||
tokio = { version = "1", features = ["full"] }
|
||||
axum = { version = "0.8", features = ["macros"] }
|
||||
serde = { version = "1.0.228", features = ["serde_derive"] }
|
||||
serde_json = "1.0.150"
|
||||
sea-orm = { version = "1", features = [ "sqlx-sqlite", "runtime-tokio-rustls", "macros" ] }
|
||||
entity = {path = "../entity"}
|
||||
migration = {path = "../migration"}
|
||||
argon2 = "0.5.3"
|
||||
uuid = { version = "1.23.3", features = ["v4"] }
|
||||
tower-http = { version = "0.6", features = ["cors"] }
|
||||
futures = "0.3.32"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||
symphonia = { version = "0.5", features = ["aac"] }
|
||||
@@ -36,7 +33,6 @@ opus = "0.3.1"
|
||||
rubato = "3.0.0"
|
||||
chrono = "0.4.45"
|
||||
time = "0.3"
|
||||
regex = "1"
|
||||
thiserror = "2.0.18"
|
||||
sysinfo = "0.36"
|
||||
axum-extra = { version = "0.12.6", features = ["cookie"] }
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
use std::{error::Error, fmt::Display};
|
||||
use std::error::Error;
|
||||
|
||||
use bytes::Bytes;
|
||||
use rubato::{audioadapter_buffers::direct::InterleavedSlice, Fft, Resampler};
|
||||
@@ -95,23 +95,6 @@ pub struct AACParser {
|
||||
decoder: Option<Box<dyn Decoder>>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum AudioParseError {
|
||||
InvalidCodec,
|
||||
NoConfigPacket,
|
||||
}
|
||||
|
||||
impl Display for AudioParseError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::InvalidCodec => write!(f, "Not the right codec provided"),
|
||||
Self::NoConfigPacket => write!(f, "No config packet cached"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Error for AudioParseError {}
|
||||
|
||||
// Byte 0: upper nibble = sound format (10 = AAC)
|
||||
// Byte 1: 0 = AudioSpecificConfig, 1 = raw AAC frame
|
||||
impl AACParser {
|
||||
@@ -129,7 +112,7 @@ impl AACParser {
|
||||
}
|
||||
|
||||
if (bytes[0] >> 4) != 10 {
|
||||
return Err(Box::new(AudioParseError::InvalidCodec));
|
||||
return Err("not the right codec provided".into());
|
||||
}
|
||||
|
||||
if bytes[1] == 0 {
|
||||
@@ -142,10 +125,7 @@ impl AACParser {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let decoder = self
|
||||
.decoder
|
||||
.as_mut()
|
||||
.ok_or(AudioParseError::NoConfigPacket)?;
|
||||
let decoder = self.decoder.as_mut().ok_or("no config packet cached")?;
|
||||
|
||||
let packet = Packet::new_from_boxed_slice(
|
||||
0,
|
||||
|
||||
+21
-94
@@ -1,28 +1,20 @@
|
||||
use std::{
|
||||
net::SocketAddr,
|
||||
sync::{
|
||||
Arc, LazyLock,
|
||||
Arc,
|
||||
atomic::{AtomicI32, Ordering},
|
||||
},
|
||||
time::Duration,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use axum_extra::extract::{CookieJar, cookie::Cookie};
|
||||
use regex::Regex;
|
||||
|
||||
// The Rust `regex` crate is guaranteed linear-time and therefore does NOT
|
||||
// support lookaround, so the JS source pattern
|
||||
// /^(?=.*[A-Za-z])[A-Za-z0-9_-]{1,67}$/
|
||||
// cannot be ported verbatim. The `(?=.*[A-Za-z])` lookahead only means
|
||||
// "must contain at least one letter" — we drop it from the pattern and
|
||||
// enforce that condition with a separate `.chars().any(..)` check below.
|
||||
static KEY_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"^[A-Za-z0-9'-]{1,67}$").unwrap());
|
||||
|
||||
// Allowed charset for stream-key passwords and custom IDs: letters, numbers,
|
||||
// dashes and apostrophes only — no spaces. Mirrors the frontend
|
||||
// `/^[A-Za-z0-9'-]{0,67}$/` used by the keys-page popups; empty clears the value.
|
||||
static PASSWORD_RE: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"^[A-Za-z0-9'-]{0,67}$").unwrap());
|
||||
/// Allowed charset for labels, passwords and custom IDs: letters, numbers,
|
||||
/// dashes and apostrophes — no spaces. Mirrors the frontend
|
||||
/// `/^[A-Za-z0-9'-]{0,67}$/` used by the keys-page popups.
|
||||
fn valid_charset(s: &str) -> bool {
|
||||
s.len() <= 67 && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '\'')
|
||||
}
|
||||
|
||||
use axum::{
|
||||
Json, Router,
|
||||
@@ -58,23 +50,15 @@ use crate::{
|
||||
const MAX_USERNAME_LEN: usize = 32;
|
||||
const MAX_LABEL_LEN: usize = 64;
|
||||
|
||||
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, Result<String, String>)>,
|
||||
pub appstate: Arc<Mutex<AppState>>,
|
||||
pub request_count: AtomicI32,
|
||||
pub db: DatabaseConnection,
|
||||
pub config: Arc<HttpServerConfig>,
|
||||
pub info: Arc<ServerInfo>,
|
||||
pub signup_code: String,
|
||||
pub version: &'static str,
|
||||
pub start_time: Instant,
|
||||
}
|
||||
|
||||
impl HttpServer {
|
||||
@@ -112,8 +96,6 @@ impl HttpServer {
|
||||
.get(get_all_stream_keys)
|
||||
.patch(edit_stream_key),
|
||||
)
|
||||
// .route("/api/admin/server_stats", get(todo!()))
|
||||
// .route("/api/whip", post(handle_whip_injest))
|
||||
.route("/api/login", post(login_handler))
|
||||
.route("/api/stream/{slug}", post(stream_handler))
|
||||
.route("/api/whip", post(handle_whip_injest))
|
||||
@@ -121,14 +103,8 @@ impl HttpServer {
|
||||
"/api/whip/{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);
|
||||
@@ -172,25 +148,23 @@ async fn edit_stream_key(
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
// Trim surrounding whitespace so labels aren't stored with leading/trailing spaces.
|
||||
let new_label = payload.new.as_deref().map(str::trim);
|
||||
// Length (1..=67) and allowed charset ([A-Za-z0-9'-], no spaces).
|
||||
// Length (1..=67), allowed charset, and at least one letter.
|
||||
if let Some(label) = new_label {
|
||||
if !KEY_RE.is_match(label) {
|
||||
if label.is_empty() || !valid_charset(label) {
|
||||
return Err(HttpError::BadRequest("invalid label".into()));
|
||||
}
|
||||
// Replaces the JS lookahead: label must contain at least one letter.
|
||||
if !label.chars().any(|c| c.is_ascii_alphabetic()) {
|
||||
return Err(HttpError::BadRequest("label must contain a letter".into()));
|
||||
}
|
||||
}
|
||||
// Custom IDs and passwords share a charset: letters, numbers, dashes,
|
||||
// apostrophes — no spaces. Empty values are allowed (they clear the field).
|
||||
// Empty values are allowed (they clear the field).
|
||||
if let Some(custom_id) = payload.custom_id.as_deref() {
|
||||
if !custom_id.is_empty() && !PASSWORD_RE.is_match(custom_id) {
|
||||
if !custom_id.is_empty() && !valid_charset(custom_id) {
|
||||
return Err(HttpError::BadRequest("invalid custom id".into()));
|
||||
}
|
||||
}
|
||||
if let Some(pwd) = payload.password.as_deref() {
|
||||
if !pwd.is_empty() && !PASSWORD_RE.is_match(pwd) {
|
||||
if !pwd.is_empty() && !valid_charset(pwd) {
|
||||
return Err(HttpError::BadRequest("invalid password".into()));
|
||||
}
|
||||
}
|
||||
@@ -288,22 +262,6 @@ struct CreateStreamKeyBody {
|
||||
label: String,
|
||||
}
|
||||
|
||||
struct SessionCookie(Option<String>);
|
||||
|
||||
impl<S> FromRequestParts<S> for SessionCookie
|
||||
where
|
||||
S: Send + Sync,
|
||||
{
|
||||
type Rejection = HttpError;
|
||||
|
||||
async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
|
||||
let jar = CookieJar::from_request_parts(parts, _state).await.unwrap();
|
||||
Ok(SessionCookie(
|
||||
jar.get("session").map(|c| c.value().to_string()),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
struct AuthUser(entity::users::Model);
|
||||
|
||||
impl FromRequestParts<Arc<HttpServer>> for AuthUser {
|
||||
@@ -409,13 +367,10 @@ fn cookie_for_token(token: &str, dev: bool) -> Cookie<'static> {
|
||||
|
||||
#[axum::debug_handler]
|
||||
async fn login_handler(
|
||||
session: SessionCookie,
|
||||
State(state): State<Arc<HttpServer>>,
|
||||
DevFlag(dev): DevFlag,
|
||||
Json(payload): Json<LoginForm>,
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
tracing::debug!(session = ?session.0, "login: existing session cookie");
|
||||
|
||||
let user = users::Entity::find_by_username(&state.db, payload.username.clone())
|
||||
.await?
|
||||
.ok_or_else(|| {
|
||||
@@ -447,13 +402,11 @@ struct CreateUserForm {
|
||||
}
|
||||
|
||||
async fn create_user_handler(
|
||||
session: SessionCookie,
|
||||
State(state): State<Arc<HttpServer>>,
|
||||
DevFlag(dev): DevFlag,
|
||||
Json(payload): Json<CreateUserForm>,
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
tracing::debug!(session = ?session.0, "create_user: existing session cookie");
|
||||
if state.config.signup_code.is_empty() || payload.ref_token != state.config.signup_code {
|
||||
if state.signup_code.is_empty() || payload.ref_token != state.signup_code {
|
||||
warn!(username = %payload.username, "signup rejected: invalid signup code");
|
||||
return Err(HttpError::Unauthorized);
|
||||
}
|
||||
@@ -596,43 +549,17 @@ struct HealthResponse {
|
||||
async fn health_handler(
|
||||
State(state): State<Arc<HttpServer>>,
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
let uptime = state.info.start_time.elapsed().as_secs();
|
||||
let uptime = state.start_time.elapsed().as_secs();
|
||||
Ok((
|
||||
StatusCode::OK,
|
||||
Json(HealthResponse {
|
||||
status: "ok",
|
||||
uptime_seconds: uptime,
|
||||
version: state.info.version,
|
||||
version: state.version,
|
||||
}),
|
||||
))
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct UptimeResponse {
|
||||
uptime_seconds: u64,
|
||||
}
|
||||
|
||||
async fn uptime_handler(
|
||||
State(state): State<Arc<HttpServer>>,
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
let uptime = state.info.start_time.elapsed().as_secs();
|
||||
Ok((
|
||||
StatusCode::OK,
|
||||
Json(UptimeResponse {
|
||||
uptime_seconds: uptime,
|
||||
}),
|
||||
))
|
||||
}
|
||||
|
||||
async fn version_handler(
|
||||
State(state): State<Arc<HttpServer>>,
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
Ok((
|
||||
StatusCode::OK,
|
||||
Json(serde_json::json!({ "version": state.info.version })),
|
||||
))
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct StatsResponse {
|
||||
version: &'static str,
|
||||
@@ -645,7 +572,7 @@ struct StatsResponse {
|
||||
async fn stats_handler(
|
||||
State(state): State<Arc<HttpServer>>,
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
let uptime = state.info.start_time.elapsed().as_secs();
|
||||
let uptime = state.start_time.elapsed().as_secs();
|
||||
let mut system = System::new();
|
||||
system.refresh_cpu_all();
|
||||
let cpu_usage = system.global_cpu_usage();
|
||||
@@ -657,7 +584,7 @@ async fn stats_handler(
|
||||
Ok((
|
||||
StatusCode::OK,
|
||||
Json(StatsResponse {
|
||||
version: state.info.version,
|
||||
version: state.version,
|
||||
uptime_seconds: uptime,
|
||||
cpu_usage_percent: cpu_usage,
|
||||
active_streams,
|
||||
|
||||
+17
-13
@@ -18,8 +18,8 @@ use uuid::Uuid;
|
||||
use crate::{
|
||||
audio::OpusAudioFrame,
|
||||
codec::VideoFrame,
|
||||
http::{HttpServer, HttpServerConfig, ServerInfo},
|
||||
webrtc_proxy::{WebRtcProxyConfig, WebrtcProxy},
|
||||
http::HttpServer,
|
||||
webrtc_proxy::WebrtcProxy,
|
||||
};
|
||||
|
||||
const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION");
|
||||
@@ -34,7 +34,6 @@ mod webrtc;
|
||||
mod webrtc_ingest;
|
||||
mod webrtc_proxy;
|
||||
|
||||
// #[derive(Debug)]
|
||||
pub struct AppState {
|
||||
pub stream_sessions: Arc<DashMap<i32, StreamSession>>,
|
||||
pub webrtc_proxy: WebrtcProxy,
|
||||
@@ -47,6 +46,19 @@ pub enum StreamCodec {
|
||||
AV1,
|
||||
}
|
||||
|
||||
impl StreamCodec {
|
||||
/// Map str0m's codec enum to ours; None for non-video codecs.
|
||||
pub fn from_str0m(c: str0m::format::Codec) -> Option<Self> {
|
||||
use str0m::format::Codec;
|
||||
match c {
|
||||
Codec::H264 => Some(Self::H264),
|
||||
Codec::H265 => Some(Self::H265),
|
||||
Codec::Av1 => Some(Self::AV1),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct StreamSession {
|
||||
pub stream_key_id: i32,
|
||||
pub stream_key_label: String,
|
||||
@@ -92,12 +104,10 @@ async fn main() -> Result<(), Box<dyn Error>> {
|
||||
stream_session::Model::clean_unended_streams(&db)
|
||||
.await
|
||||
.unwrap();
|
||||
let proxyconfig = WebRtcProxyConfig {
|
||||
proxy_port: env::var("RTC_PORT")
|
||||
let proxyconfig = env::var("RTC_PORT")
|
||||
.unwrap_or("6969".into())
|
||||
.parse()
|
||||
.expect("RTC_PORT needs to be a number (i32)"),
|
||||
};
|
||||
.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 {
|
||||
@@ -120,17 +130,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
|
||||
appstate: appstate.clone(),
|
||||
request_count: std::sync::atomic::AtomicI32::new(0),
|
||||
db: db.clone(),
|
||||
config: HttpServerConfig {
|
||||
signup_code: env::var("SIGNUP_CODE").unwrap_or_else(|_| {
|
||||
warn!("SIGNUP_CODE not set; signup will be disabled");
|
||||
String::new()
|
||||
}),
|
||||
}
|
||||
.into(),
|
||||
info: Arc::new(ServerInfo {
|
||||
version: SERVER_VERSION,
|
||||
start_time: std::time::Instant::now(),
|
||||
}),
|
||||
};
|
||||
|
||||
let app = appstate.lock().await;
|
||||
@@ -164,7 +169,6 @@ async fn main() -> Result<(), Box<dyn Error>> {
|
||||
res = workers.join_next() => {
|
||||
if let Some(Err(e)) = res {
|
||||
tracing::error!("worker panicked: {:?}", e);
|
||||
// fs::File::
|
||||
} else {
|
||||
tracing::error!("a worker exited unexpectedly");
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@ use tokio::{
|
||||
net::{TcpListener, TcpStream},
|
||||
time::timeout,
|
||||
};
|
||||
use tracing::{debug, info, trace, warn};
|
||||
use tracing::{debug, info, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{
|
||||
@@ -134,7 +134,6 @@ impl Rtmp {
|
||||
// ponytail: fixed 512; per-stream dynamic sizing if memory ever matters.
|
||||
let (mut video_tx, video_rx) = broadcast::<Arc<VideoFrame>>(512);
|
||||
let (mut audio_tx, audio_rx) = broadcast::<Arc<OpusAudioFrame>>(512);
|
||||
// video_rx.cycle
|
||||
|
||||
video_tx.set_overflow(true);
|
||||
audio_tx.set_overflow(true);
|
||||
@@ -369,9 +368,6 @@ impl Rtmp {
|
||||
}
|
||||
current_stream_key_id = None;
|
||||
}
|
||||
// TODO: We can totally replace the broadcast with a
|
||||
// circular_buff
|
||||
// Arc<Vec<ArcSwap<Frame>>>
|
||||
ServerSessionEvent::VideoDataReceived {
|
||||
data, timestamp, ..
|
||||
} => match Self::parse_video_codec(&data) {
|
||||
@@ -385,54 +381,15 @@ impl Rtmp {
|
||||
}
|
||||
codec_stamped = true;
|
||||
}
|
||||
match codec {
|
||||
StreamCodec::H264 => {
|
||||
let p = parser.get_or_insert_with(|| {
|
||||
Box::new(H264CodecParser::new())
|
||||
let p = parser.get_or_insert_with(|| match codec {
|
||||
StreamCodec::H264 => Box::new(H264CodecParser::new()),
|
||||
StreamCodec::H265 => Box::new(H265CodecParser::new()),
|
||||
StreamCodec::AV1 => Box::new(Av1CodecParser::new()),
|
||||
});
|
||||
if let Some(frame) = p.parse(&data, timestamp.value)
|
||||
{
|
||||
if let Some(frame) = p.parse(&data, timestamp.value) {
|
||||
video_tx.broadcast(Arc::new(frame)).await.ok();
|
||||
}
|
||||
}
|
||||
StreamCodec::H265 => {
|
||||
let p = parser.get_or_insert_with(|| {
|
||||
Box::new(H265CodecParser::new())
|
||||
});
|
||||
if let Some(frame) = p.parse(&data, timestamp.value)
|
||||
{
|
||||
video_tx.broadcast(Arc::new(frame)).await.ok();
|
||||
}
|
||||
}
|
||||
StreamCodec::AV1 => {
|
||||
let p = parser.get_or_insert_with(|| {
|
||||
Box::new(Av1CodecParser::new())
|
||||
});
|
||||
let pkt_type = data[0] & 0x0F;
|
||||
match p.parse(&data, timestamp.value) {
|
||||
Some(frame) => {
|
||||
trace!(
|
||||
pkt_type,
|
||||
is_keyframe = frame.is_keyframe,
|
||||
ts = frame.timestamp_ms,
|
||||
bytes = frame.data.len(),
|
||||
"AV1 frame → broadcast"
|
||||
);
|
||||
video_tx
|
||||
.broadcast(Arc::new(frame))
|
||||
.await
|
||||
.ok();
|
||||
}
|
||||
None => {
|
||||
trace!(
|
||||
pkt_type,
|
||||
"AV1 packet produced no frame (seq header or unknown type)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(_err) => {
|
||||
warn!("");
|
||||
}
|
||||
|
||||
@@ -58,9 +58,6 @@ impl Webrtc {
|
||||
continue;
|
||||
}
|
||||
|
||||
// let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
||||
// let local_addr = socket.local_addr().unwrap();
|
||||
|
||||
let local_addr = self.proxy.public_addr();
|
||||
|
||||
let session = self.sessions_ref.get(&stream_id);
|
||||
|
||||
@@ -33,18 +33,21 @@ use crate::{
|
||||
/// Kill a WHIP publish if it delivers no media (video or audio) this long.
|
||||
const NO_MEDIA_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
|
||||
fn bearer_token(headers: &HeaderMap) -> Option<String> {
|
||||
headers
|
||||
.get(header::AUTHORIZATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.strip_prefix("Bearer "))
|
||||
.map(|s| s.trim().to_string())
|
||||
}
|
||||
|
||||
pub async fn handle_whip_injest_delete(
|
||||
State(state): State<Arc<HttpServer>>,
|
||||
ConnectInfo(remote): ConnectInfo<SocketAddr>,
|
||||
Path(_slug): Path<String>,
|
||||
headers: HeaderMap,
|
||||
) -> Result<impl IntoResponse, HttpError> {
|
||||
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 {
|
||||
let Some(token) = bearer_token(&headers) else {
|
||||
warn!("Whip: missing bearer token from {}", remote);
|
||||
return Err(HttpError::Unauthorized);
|
||||
};
|
||||
@@ -120,12 +123,7 @@ pub async fn handle_whip_injest(
|
||||
};
|
||||
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 {
|
||||
let Some(token) = bearer_token(&headers) else {
|
||||
warn!("Whip: missing bearer token from {}", remote);
|
||||
return err(StatusCode::UNAUTHORIZED, "missing bearer token");
|
||||
};
|
||||
@@ -187,17 +185,13 @@ pub async fn handle_whip_injest(
|
||||
cc.clear();
|
||||
cc.enable_opus(true);
|
||||
match &stream_codec {
|
||||
Some(crate::StreamCodec::H264) => {
|
||||
info!("Whip: enabling H.264 codec");
|
||||
cc.enable_h264(true);
|
||||
Some(codec) => {
|
||||
info!("Whip: enabling {codec:?} codec");
|
||||
match codec {
|
||||
crate::StreamCodec::H264 => cc.enable_h264(true),
|
||||
crate::StreamCodec::H265 => cc.enable_h265(true),
|
||||
crate::StreamCodec::AV1 => cc.enable_av1(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");
|
||||
@@ -366,12 +360,6 @@ pub async fn handle_whip_injest(
|
||||
.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,
|
||||
// "<stun:stun.l.google.com:19302>; rel=\"ice-server\"",
|
||||
// )
|
||||
.body(axum::body::Body::from(answer_sdp))
|
||||
.unwrap()
|
||||
}
|
||||
@@ -422,7 +410,6 @@ async fn detach_inject_rtc(
|
||||
let mut audio_mid: Option<Mid> = None;
|
||||
let mut video_tx: Option<async_broadcast::Sender<Arc<VideoFrame>>> = None;
|
||||
let mut audio_tx: Option<async_broadcast::Sender<Arc<OpusAudioFrame>>> = None;
|
||||
let mut _connected = false;
|
||||
let mut disconnect_timer: Option<Instant> = None;
|
||||
|
||||
loop {
|
||||
@@ -525,7 +512,6 @@ async fn detach_inject_rtc(
|
||||
video_tx = Some(session.frame_channel.clone());
|
||||
audio_tx = Some(session.audio_channel.clone());
|
||||
}
|
||||
_connected = true;
|
||||
}
|
||||
_ => {}
|
||||
},
|
||||
@@ -610,22 +596,14 @@ async fn detach_inject_rtc(
|
||||
pub fn extract_negotiated_codec_info(
|
||||
answer: &str0m::change::SdpAnswer,
|
||||
) -> (Option<crate::StreamCodec>, Option<u8>, Option<u32>) {
|
||||
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);
|
||||
return (
|
||||
crate::StreamCodec::from_str0m(p.spec().codec),
|
||||
Some(*p.pt()),
|
||||
p.spec().format.profile_level_id,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -634,24 +612,14 @@ pub fn extract_negotiated_codec_info(
|
||||
|
||||
/// 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.
|
||||
/// Walks every m= line's `rtp_params()`; returns the first video codec found.
|
||||
pub fn video_codec_from_sdp_offer(
|
||||
sdp_offer: &str0m::change::SdpOffer,
|
||||
) -> Option<crate::StreamCodec> {
|
||||
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,
|
||||
};
|
||||
return crate::StreamCodec::from_str0m(p.spec().codec);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,10 +14,6 @@ use tokio::{
|
||||
use tracing::{debug, error, info, trace, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
pub struct WebRtcProxyConfig {
|
||||
pub proxy_port: i32,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct WebrtcProxy {
|
||||
clients_ufrag: Arc<DashMap<String, tokio::sync::mpsc::Sender<(Bytes, SocketAddr)>>>,
|
||||
@@ -34,8 +30,8 @@ pub struct WebrtcProxy {
|
||||
const STUN_MAGIC: u32 = 0x2112A442;
|
||||
|
||||
impl WebrtcProxy {
|
||||
pub async fn new(config: WebRtcProxyConfig) -> Result<Self, Box<dyn Error>> {
|
||||
let sock = UdpSocket::bind(format!("0.0.0.0:{}", config.proxy_port)).await?;
|
||||
pub async fn new(proxy_port: i32) -> Result<Self, Box<dyn Error>> {
|
||||
let sock = UdpSocket::bind(format!("0.0.0.0:{}", proxy_port)).await?;
|
||||
let port = sock.local_addr()?.port();
|
||||
|
||||
let public_ip = match env::var("PUBLIC_DOMAIN")
|
||||
@@ -165,10 +161,6 @@ impl WebrtcProxy {
|
||||
self.clients_ufrag.insert(ufrag, tx);
|
||||
(self.socket.clone(), rx)
|
||||
}
|
||||
pub fn local_addr(&self) -> SocketAddr {
|
||||
self.socket.local_addr().unwrap()
|
||||
}
|
||||
|
||||
pub fn public_addr(&self) -> SocketAddr {
|
||||
self.public_addr
|
||||
}
|
||||
@@ -180,26 +172,30 @@ impl WebrtcProxy {
|
||||
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 value = stun_attributes(b).find(|(t, _)| *t == 0x0006)?.1;
|
||||
let value = std::str::from_utf8(value).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
|
||||
Some((
|
||||
parts.next()?.to_string(),
|
||||
parts.next().map(|s| s.to_string()),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
/// Iterate `(attr_type, attr_value)` pairs over a STUN message's attributes
|
||||
/// (RFC 5389 §15: header is 20 bytes, each attribute padded to a 4-byte boundary).
|
||||
fn stun_attributes(data: &[u8]) -> impl Iterator<Item = (u16, &[u8])> {
|
||||
let mut pos = 20usize;
|
||||
std::iter::from_fn(move || {
|
||||
let attr_type = u16::from_be_bytes(data.get(pos..pos + 2)?.try_into().ok()?);
|
||||
let attr_len = u16::from_be_bytes(data.get(pos + 2..pos + 4)?.try_into().ok()?) as usize;
|
||||
pos += 4;
|
||||
let value = data.get(pos..pos + attr_len)?;
|
||||
pos += (attr_len + 3) & !3;
|
||||
Some((attr_type, value))
|
||||
})
|
||||
}
|
||||
|
||||
/// IP of the interface holding the default route, via a UDP connect() trick:
|
||||
/// connect() only does a route lookup (no packets sent), so the kernel binds
|
||||
/// the source address the OS would use for outbound traffic.
|
||||
@@ -251,27 +247,15 @@ fn parse_xor_mapped_address(data: &[u8]) -> Option<std::net::IpAddr> {
|
||||
return None;
|
||||
}
|
||||
|
||||
let mut pos = 20usize;
|
||||
while pos + 4 <= data.len() {
|
||||
let attr_type = u16::from_be_bytes(data[pos..pos + 2].try_into().ok()?);
|
||||
let attr_len = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?) as usize;
|
||||
pos += 4;
|
||||
if pos + attr_len > data.len() {
|
||||
break;
|
||||
let value = stun_attributes(data).find(|(t, _)| *t == 0x0020)?.1;
|
||||
if value.len() < 8 {
|
||||
return None;
|
||||
}
|
||||
if attr_type == 0x0020 && attr_len >= 8 {
|
||||
// byte 0: reserved, byte 1: family (0x01=IPv4, 0x02=IPv6)
|
||||
let family = data[pos + 1];
|
||||
let x_port = u16::from_be_bytes(data[pos + 2..pos + 4].try_into().ok()?);
|
||||
let _ = x_port ^ (STUN_MAGIC >> 16) as u16; // port (unused here)
|
||||
|
||||
if family == 0x01 {
|
||||
let x_addr = u32::from_be_bytes(data[pos + 4..pos + 8].try_into().ok()?);
|
||||
let addr = x_addr ^ STUN_MAGIC;
|
||||
return Some(std::net::IpAddr::V4(std::net::Ipv4Addr::from(addr)));
|
||||
}
|
||||
}
|
||||
pos += (attr_len + 3) & !3;
|
||||
}
|
||||
if value[1] == 0x01 {
|
||||
let x_addr = u32::from_be_bytes(value[4..8].try_into().ok()?);
|
||||
Some(std::net::IpAddr::V4(std::net::Ipv4Addr::from(x_addr ^ magic)))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user