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

feat: WHIP support

Browse Source
This commit is contained in:
doloro
2026-08-07 11:48:31 +01:00
parent c1435a70fb
commit 5f152e9db7
3 changed files with 349 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/server/src/webrtc_ingest.rs
+160 -148
View File
@@ -64,29 +64,19 @@ pub async fn handle_whip_injest_patch(
body: String,
) -> Result<impl IntoResponse, HttpError> {
// 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
})?;
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
})?;
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();
@@ -145,113 +135,117 @@ pub async fn handle_whip_injest(
};
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);
// 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");
// 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);
}
// 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 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);
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");
}
};
// 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");
}
// 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);
}
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:"))
{
// 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)
let cand_replacement = format!(
"m=video 9 UDP/TLS/RTP/SAVPF 96\r\nc=IN IP4 {}\r\n{}\r\n",
public_addr.ip(), 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);
info!("Serving Whip SDP answer to {}:\n{}", remote, answer_sdp);
let ufrag = answer_sdp
.lines()
@@ -267,12 +261,21 @@ pub async fn handle_whip_injest(
.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");
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::<Arc<VideoFrame>>(4);
let (audio_tx, audio_rx) = broadcast::<Arc<OpusAudioFrame>>(4);
let (mut video_tx, video_rx) = broadcast::<Arc<VideoFrame>>(32);
let (mut audio_tx, audio_rx) = broadcast::<Arc<OpusAudioFrame>>(32);
// Never block the ingest loop on slow/missing viewers: overwrite the
// oldest frame instead (same as the RTMP path; the viewer re-waits for a
// keyframe on Overflowed). The undrained rx below must not stall sends.
video_tx.set_overflow(true);
audio_tx.set_overflow(true);
stream_sessions.insert(
key.id,
@@ -288,7 +291,10 @@ pub async fn handle_whip_injest(
video_profile_level_id: video_profile,
},
);
info!(stream_key_id = key.id, "Whip: StreamSession inserted, spawning detach task");
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.
@@ -326,7 +332,10 @@ pub async fn handle_whip_injest(
.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\"")
.header(
header::LINK,
"<stun:stun.l.google.com:19302>; rel=\"ice-server\"",
)
.body(axum::body::Body::from(answer_sdp))
.unwrap()
}
@@ -372,14 +381,14 @@ async fn detach_inject_rtc(
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 {
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;
@@ -430,7 +439,10 @@ async fn detach_inject_rtc(
}
}
Event::Connected => {
info!(stream_key_id, "Whip DTLS+ICE connected, wiring broadcast channels");
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());
@@ -448,7 +460,7 @@ async fn detach_inject_rtc(
};
let sleep = tokio::time::sleep_until(deadline.max(Instant::now()).into());
tokio::select! {
tokio::select! {
_ = sleep => {
if let Err(e) = rtc.handle_input(Input::Timeout(Instant::now())) {
error!(stream_key_id, "Whip handle_input(Timeout) error: {:?}", e);
@@ -510,28 +522,28 @@ async fn detach_inject_rtc(
/// 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,
answer: &str0m::change::SdpAnswer,
) -> (Option<crate::StreamCodec>, Option<u8>, Option<u32>) {
use str0m::format::Codec;
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)
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.