doloro/simple-instant-stream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: firefox not resolving codec (h264)

Browse Source
This commit is contained in:
Doloro1978
2026-07-10 00:31:01 +01:00
parent 888eace0bf
commit 2f0a02131e
7 changed files with 206 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -2872,6 +2872,7 @@ dependencies = [
"async-broadcast",
"axum",
"bytes",
"chrono",
"dashmap",
"entity",
"futures",
Dockerfile
+1 -1
View File
@@ -43,6 +43,6 @@ EXPOSE 1935
EXPOSE 3000
EXPOSE 6969/udp
ENV RUST_LOG="warn"
ENV RUST_LOG="info,warn"
CMD ["/rtmp-to-whip"]
crates/server/Cargo.toml
+1
View File
@@ -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"
crates/server/src/http.rs
+14 -1
View File
@@ -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<Utc>, //UNIX TIMESTAMP
}
async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoResponse {
@@ -118,6 +120,7 @@ async fn catalog_handler(State(state): State<Arc<HttpServer>>) -> impl IntoRespo
let catalog: Vec<StreamListing> = 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<Arc<HttpServer>>) -> 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,
}
}
}))
crates/server/src/main.rs
+3
View File
@@ -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<Arc<VideoFrame>>,
pub audio_channel: async_broadcast::Sender<Arc<OpusAudioFrame>>,
pub codec: Option<StreamCodec>,
//
pub started_at: DateTime<Utc>,
}
#[tokio::main]
crates/server/src/rtmp.rs
+3
View File
@@ -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(),
},
);
crates/server/src/webrtc.rs
+183 -111
View File
@@ -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<async_broadcast::Receiver<Arc<VideoFrame>>>,
) -> Result<Arc<VideoFrame>, async_broadcast::RecvError> {
stream.as_mut().unwrap().recv().await
}
async fn recv_audio(
stream: &mut Option<async_broadcast::Receiver<Arc<OpusAudioFrame>>>,
) -> Result<Arc<OpusAudioFrame>, async_broadcast::RecvError> {
stream.as_mut().unwrap().recv().await
}
fn write_video_frame(
frame: Arc<VideoFrame>,
saw_keyframe: &mut bool,
video_pt: Option<Pt>,
video_mid: Option<Mid>,
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<async_broadcast::Receiver<Arc<VideoFrame>>>,
saw_keyframe: &mut bool,
video_pt: Option<Pt>,
video_mid: Option<Mid>,
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<OpusAudioFrame>,
audio_pt: Option<Pt>,
audio_mid: Option<Mid>,
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<async_broadcast::Receiver<Arc<OpusAudioFrame>>>,
audio_pt: Option<Pt>,
audio_mid: Option<Mid>,
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(_)) => {}
}
}
}