586 lines
25 KiB
Rust
586 lines
25 KiB
Rust
//! WebRTC peer-connection helpers for the native P2P fast path (webrtc 0.17.2,
|
|
//! the classic tokio-callback API). This module builds the `API` + a
|
|
//! `RTCPeerConnection` from the ICE servers the relay hands us in `rtc.config`
|
|
//! / `rtc.offer.ice`. The DataChannel transport ([`crate::transport`]) and the
|
|
//! offer/answer signaling are layered on top of these.
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use anyhow::{anyhow, Result};
|
|
use serde::Deserialize;
|
|
use serde_json::{json, Value};
|
|
use tokio::sync::mpsc;
|
|
use webrtc::api::interceptor_registry::register_default_interceptors;
|
|
use webrtc::api::media_engine::MediaEngine;
|
|
use webrtc::api::{APIBuilder, API};
|
|
use webrtc::data_channel::data_channel_init::RTCDataChannelInit;
|
|
use webrtc::data_channel::RTCDataChannel;
|
|
use webrtc::ice_transport::ice_candidate_type::RTCIceCandidateType;
|
|
use webrtc::ice_transport::ice_server::RTCIceServer;
|
|
use webrtc::interceptor::registry::Registry;
|
|
use webrtc::peer_connection::configuration::RTCConfiguration;
|
|
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
|
|
use webrtc::peer_connection::RTCPeerConnection;
|
|
|
|
/// One ICE server exactly as it arrives in the relay's `rtc.config` reply and
|
|
/// is forwarded in `rtc.offer.ice`: `{"urls":[...],"username":...,"credential":...}`.
|
|
/// coturn REST creds (username = "<expiry>:nd", credential = base64 HMAC-SHA1)
|
|
/// drop straight into webrtc's long-term TURN credential fields.
|
|
#[derive(Debug, Clone, Deserialize, serde::Serialize)]
|
|
pub struct RtcIceServerJson {
|
|
#[serde(default)]
|
|
pub urls: Vec<String>,
|
|
#[serde(default)]
|
|
pub username: Option<String>,
|
|
#[serde(default)]
|
|
pub credential: Option<String>,
|
|
}
|
|
|
|
/// Map the relay's ICE JSON into webrtc `RTCIceServer`s (0.17.2 has only
|
|
/// urls/username/credential).
|
|
pub fn ice_servers_from_json(ice: &[RtcIceServerJson]) -> Vec<RTCIceServer> {
|
|
ice.iter()
|
|
.filter(|s| !s.urls.is_empty())
|
|
.map(|s| RTCIceServer {
|
|
urls: s.urls.clone(),
|
|
username: s.username.clone().unwrap_or_default(),
|
|
credential: s.credential.clone().unwrap_or_default(),
|
|
..Default::default()
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Build a webrtc `API` with the default media engine + interceptors. One per
|
|
/// peer connection is fine (cheap).
|
|
///
|
|
/// Raises the SCTP max send message size to 256 KiB (webrtc-rs defaults to
|
|
/// 64 KiB), matching the web client's advertised `a=max-message-size:262144`.
|
|
/// Without this a whole ~200 KiB video keyframe is rejected with
|
|
/// `ErrOutboundPacketTooLarge`; with it the native peer can send whole E2E
|
|
/// frames exactly like the browser / aiortc, which is what web/Python interop
|
|
/// on the `media` channel requires. (Harmless for the native single channel,
|
|
/// which fragments at 16 KiB.)
|
|
pub fn build_api() -> Result<API> {
|
|
use webrtc::api::setting_engine::{SctpMaxMessageSize, SettingEngine};
|
|
let mut media = MediaEngine::default();
|
|
let mut registry = Registry::new();
|
|
registry = register_default_interceptors(registry, &mut media)?;
|
|
let mut se = SettingEngine::default();
|
|
se.set_sctp_max_message_size_can_send(SctpMaxMessageSize::Bounded(262144));
|
|
Ok(APIBuilder::new()
|
|
.with_media_engine(media)
|
|
.with_interceptor_registry(registry)
|
|
.with_setting_engine(se)
|
|
.build())
|
|
}
|
|
|
|
/// A peer-connection configuration from a list of ICE servers.
|
|
pub fn rtc_config(ice_servers: Vec<RTCIceServer>) -> RTCConfiguration {
|
|
RTCConfiguration {
|
|
ice_servers,
|
|
..Default::default()
|
|
}
|
|
}
|
|
|
|
/// Create a new peer connection on `api` with `cfg`.
|
|
pub async fn new_pc(api: &API, cfg: RTCConfiguration) -> Result<Arc<RTCPeerConnection>> {
|
|
Ok(Arc::new(api.new_peer_connection(cfg).await?))
|
|
}
|
|
|
|
/// Cap on how long we wait for ICE gathering to finish before sending the SDP.
|
|
/// The web/Python peers are NON-TRICKLE (candidates must be embedded in the
|
|
/// offer/answer), so we gather to completion — but never hang on a stuck
|
|
/// network interface: 2.5 s matches the reference signaling timeout.
|
|
pub const GATHER_CAP: Duration = Duration::from_millis(2500);
|
|
|
|
/// The max DataChannel message size we advertise (bytes) — matches the web
|
|
/// client's `a=max-message-size:262144`.
|
|
pub const MAX_MESSAGE_SIZE: u32 = 262144;
|
|
|
|
/// Ensure the SDP's `m=application` section advertises our max-message-size.
|
|
/// webrtc-rs never emits `a=max-message-size`, so a peer defaults our receive
|
|
/// cap to 64 KiB and refuses to send us a whole ~200 KiB keyframe. Injecting it
|
|
/// (right after `a=sctp-port:`) raises the PEER's send-to-us cap to 256 KiB —
|
|
/// and is exactly what aiortc / the browser advertise, so it is also how we
|
|
/// receive whole video frames from the Python host over the `media` channel.
|
|
pub fn advertise_max_message_size(sdp: String) -> String {
|
|
if sdp.contains("a=max-message-size:") {
|
|
return sdp; // already present (e.g. echoed from a web peer)
|
|
}
|
|
let nl = if sdp.contains("\r\n") { "\r\n" } else { "\n" };
|
|
let needle = "a=sctp-port:";
|
|
if let Some(pos) = sdp.find(needle) {
|
|
// insert the attribute on its own line after the sctp-port line
|
|
if let Some(eol) = sdp[pos..].find('\n') {
|
|
let insert_at = pos + eol + 1;
|
|
let mut out = String::with_capacity(sdp.len() + 32);
|
|
out.push_str(&sdp[..insert_at]);
|
|
out.push_str(&format!("a=max-message-size:{MAX_MESSAGE_SIZE}{nl}"));
|
|
out.push_str(&sdp[insert_at..]);
|
|
return out;
|
|
}
|
|
}
|
|
sdp
|
|
}
|
|
|
|
/// Offerer (viewer): create the offer, then gather ICE **to completion** and
|
|
/// return the full local SDP for `rtc.offer.sdp`. Non-trickle: every candidate
|
|
/// is embedded in the returned SDP, because the Python host has no trickle
|
|
/// candidate handler. The caller must create its DataChannel(s) on `pc` BEFORE
|
|
/// calling (they shape the offer). `gathering_complete_promise` is taken before
|
|
/// `set_local_description`, which is what actually starts gathering.
|
|
pub async fn offer_nontrickle(pc: &Arc<RTCPeerConnection>, gather_cap: Duration) -> Result<String> {
|
|
let offer = pc.create_offer(None).await?;
|
|
let mut gather = pc.gathering_complete_promise().await;
|
|
pc.set_local_description(offer).await?;
|
|
let _ = tokio::time::timeout(gather_cap, gather.recv()).await;
|
|
let local = pc
|
|
.local_description()
|
|
.await
|
|
.ok_or_else(|| anyhow!("no local description after ICE gathering"))?;
|
|
Ok(advertise_max_message_size(local.sdp))
|
|
}
|
|
|
|
/// Answerer (host): apply the remote offer SDP, create the answer, gather ICE
|
|
/// to completion (non-trickle) and return the full local SDP for
|
|
/// `rtc.answer.sdp`. Register `on_data_channel` on `pc` BEFORE calling so the
|
|
/// viewer-created channel is captured.
|
|
pub async fn answer_nontrickle(
|
|
pc: &Arc<RTCPeerConnection>,
|
|
offer_sdp: &str,
|
|
gather_cap: Duration,
|
|
) -> Result<String> {
|
|
let offer = RTCSessionDescription::offer(offer_sdp.to_string())?;
|
|
pc.set_remote_description(offer).await?;
|
|
let answer = pc.create_answer(None).await?;
|
|
let mut gather = pc.gathering_complete_promise().await;
|
|
pc.set_local_description(answer).await?;
|
|
let _ = tokio::time::timeout(gather_cap, gather.recv()).await;
|
|
let local = pc
|
|
.local_description()
|
|
.await
|
|
.ok_or_else(|| anyhow!("no local description after ICE gathering"))?;
|
|
Ok(advertise_max_message_size(local.sdp))
|
|
}
|
|
|
|
/// Offerer (viewer): apply the remote answer SDP received in `rtc.answer`.
|
|
pub async fn apply_answer(pc: &Arc<RTCPeerConnection>, answer_sdp: &str) -> Result<()> {
|
|
let answer = RTCSessionDescription::answer(answer_sdp.to_string())?;
|
|
pc.set_remote_description(answer).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Wait for the server to return rtc.config before falling back to a default
|
|
/// STUN server. 2.5 s matches the reference signaling timeout.
|
|
pub const CONFIG_CAP: Duration = Duration::from_millis(2500);
|
|
/// How long to wait for the peer's rtc.answer.
|
|
pub const ANSWER_CAP: Duration = Duration::from_secs(10);
|
|
/// How long to wait for the pc to connect + the DataChannel to open.
|
|
pub const CONNECT_CAP: Duration = Duration::from_secs(15);
|
|
|
|
/// Default ICE if the server does not answer rtc.config in time. A public STUN
|
|
/// is enough to still find a server-reflexive path; host/LAN pairs connect
|
|
/// without it (and loopback needs no ICE server at all).
|
|
pub fn default_ice() -> Vec<RtcIceServerJson> {
|
|
vec![RtcIceServerJson {
|
|
urls: vec!["stun:stun.l.google.com:19302".into()],
|
|
username: None,
|
|
credential: None,
|
|
}]
|
|
}
|
|
|
|
/// Pull rtc.* Values from `rtc_in` until one matches `want` (dropping any other
|
|
/// rtc.* that arrives first), capped by `cap`. rtc.unavailable short-circuits to
|
|
/// an error so the caller abandons the upgrade and stays on the relay.
|
|
async fn wait_rtc(
|
|
rtc_in: &mut mpsc::UnboundedReceiver<Value>,
|
|
want: &str,
|
|
cap: Duration,
|
|
) -> Result<Value> {
|
|
let deadline = tokio::time::Instant::now() + cap;
|
|
loop {
|
|
match tokio::time::timeout_at(deadline, rtc_in.recv()).await {
|
|
Ok(Some(v)) => {
|
|
let t = v.get("t").and_then(|t| t.as_str()).unwrap_or("");
|
|
if t == want {
|
|
return Ok(v);
|
|
}
|
|
if t == "rtc.unavailable" {
|
|
return Err(anyhow!("peer declined P2P (rtc.unavailable)"));
|
|
}
|
|
// ignore any other rtc.* and keep waiting
|
|
}
|
|
Ok(None) => return Err(anyhow!("rtc signaling channel closed")),
|
|
Err(_) => return Err(anyhow!("timed out waiting for {want}")),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Viewer offerer: run the relay-brokered P2P upgrade handshake and return the
|
|
/// connected pc + its DataChannel once the channel opens. `send_text` enqueues
|
|
/// an rtc.* text message onto the relay (the caller supplies one that carries
|
|
/// `sid`); `rtc_in` yields inbound rtc.config / rtc.answer / rtc.unavailable
|
|
/// Values the relay loop forwards. When `ice_override` is `None` the offerer
|
|
/// asks the server for ICE via rtc.config.req (default STUN on timeout); tests
|
|
/// pass `Some(vec![])` for a loopback host-candidate-only pc.
|
|
pub async fn upgrade_offerer(
|
|
api: &API,
|
|
sid: &str,
|
|
ice_override: Option<Vec<RtcIceServerJson>>,
|
|
channel_label: &str,
|
|
send_text: &(dyn Fn(String) + Send + Sync),
|
|
rtc_in: &mut mpsc::UnboundedReceiver<Value>,
|
|
) -> Result<(Arc<RTCPeerConnection>, Arc<RTCDataChannel>)> {
|
|
let ice = match ice_override {
|
|
Some(i) => i,
|
|
None => {
|
|
send_text(json!({ "t": "rtc.config.req", "sid": sid }).to_string());
|
|
match wait_rtc(rtc_in, "rtc.config", CONFIG_CAP).await {
|
|
Ok(v) => v
|
|
.get("iceServers")
|
|
.and_then(|x| serde_json::from_value::<Vec<RtcIceServerJson>>(x.clone()).ok())
|
|
.unwrap_or_else(default_ice),
|
|
Err(_) => default_ice(),
|
|
}
|
|
}
|
|
};
|
|
|
|
let pc = new_pc(api, rtc_config(ice_servers_from_json(&ice))).await?;
|
|
let (open_tx, mut open_rx) = mpsc::channel::<()>(1);
|
|
let dc = pc
|
|
.create_data_channel(
|
|
channel_label,
|
|
Some(RTCDataChannelInit {
|
|
ordered: Some(true),
|
|
..Default::default()
|
|
}),
|
|
)
|
|
.await?;
|
|
{
|
|
let tx = open_tx.clone();
|
|
dc.on_open(Box::new(move || {
|
|
let tx = tx.clone();
|
|
Box::pin(async move {
|
|
let _ = tx.send(()).await;
|
|
})
|
|
}));
|
|
}
|
|
|
|
// Non-trickle: gather fully, then send the offer with the ICE config echoed
|
|
// through so a TURN-only carrier-NAT host can reach us.
|
|
let offer_sdp = offer_nontrickle(&pc, GATHER_CAP).await?;
|
|
// `native:true` marks a native single-channel offerer: a native host only
|
|
// answers these, and declines (rtc.unavailable) a web/3-channel offerer so
|
|
// that pairing safely stays on the relay.
|
|
send_text(
|
|
json!({ "t": "rtc.offer", "sid": sid, "sdp": offer_sdp, "ice": ice, "native": true })
|
|
.to_string(),
|
|
);
|
|
|
|
let answer = wait_rtc(rtc_in, "rtc.answer", ANSWER_CAP).await?;
|
|
let answer_sdp = answer
|
|
.get("sdp")
|
|
.and_then(|s| s.as_str())
|
|
.ok_or_else(|| anyhow!("rtc.answer missing sdp"))?;
|
|
apply_answer(&pc, answer_sdp).await?;
|
|
|
|
tokio::time::timeout(CONNECT_CAP, open_rx.recv())
|
|
.await
|
|
.map_err(|_| anyhow!("P2P DataChannel did not open"))?
|
|
.ok_or_else(|| anyhow!("P2P open signal dropped"))?;
|
|
Ok((pc, dc))
|
|
}
|
|
|
|
/// Host answerer: handle one inbound rtc.offer. Build the answerer pc from the
|
|
/// offer plus the `ice` the offerer echoed, capture the viewer-created channel
|
|
/// via on_data_channel, create+send the answer via `send_text` (carrying
|
|
/// `sid`), and return the connected pc + channel once it opens.
|
|
pub async fn answer_offer(
|
|
api: &API,
|
|
sid: &str,
|
|
offer_sdp: &str,
|
|
ice: Vec<RtcIceServerJson>,
|
|
send_text: &(dyn Fn(String) + Send + Sync),
|
|
) -> Result<(Arc<RTCPeerConnection>, Arc<RTCDataChannel>)> {
|
|
let pc = new_pc(api, rtc_config(ice_servers_from_json(&ice))).await?;
|
|
let (dc_tx, mut dc_rx) = mpsc::channel::<Arc<RTCDataChannel>>(1);
|
|
let (open_tx, mut open_rx) = mpsc::channel::<()>(1);
|
|
{
|
|
let dc_tx = dc_tx.clone();
|
|
let open_tx = open_tx.clone();
|
|
pc.on_data_channel(Box::new(move |d: Arc<RTCDataChannel>| {
|
|
let dc_tx = dc_tx.clone();
|
|
let open_tx = open_tx.clone();
|
|
Box::pin(async move {
|
|
let ot = open_tx.clone();
|
|
d.on_open(Box::new(move || {
|
|
let ot = ot.clone();
|
|
Box::pin(async move {
|
|
let _ = ot.send(()).await;
|
|
})
|
|
}));
|
|
let _ = dc_tx.send(d).await;
|
|
})
|
|
}));
|
|
}
|
|
|
|
let answer_sdp = answer_nontrickle(&pc, offer_sdp, GATHER_CAP).await?;
|
|
send_text(json!({ "t": "rtc.answer", "sid": sid, "sdp": answer_sdp }).to_string());
|
|
|
|
let dc = tokio::time::timeout(CONNECT_CAP, dc_rx.recv())
|
|
.await
|
|
.map_err(|_| anyhow!("no DataChannel from the offerer"))?
|
|
.ok_or_else(|| anyhow!("DataChannel signal dropped"))?;
|
|
tokio::time::timeout(CONNECT_CAP, open_rx.recv())
|
|
.await
|
|
.map_err(|_| anyhow!("P2P DataChannel did not open (host)"))?
|
|
.ok_or_else(|| anyhow!("host open signal dropped"))?;
|
|
Ok((pc, dc))
|
|
}
|
|
|
|
/// True when the pc's nominated ICE pair is a DIRECT path (host / server-
|
|
/// reflexive on both ends) rather than a TURN relay. The relay-first state
|
|
/// machine only reports "Direct P2P" and moves media off the WSS relay for a
|
|
/// direct pair: a TURN-relayed hop is no faster than our own relay, and a
|
|
/// not-yet-selected pair is unknown, so both keep media on the relay.
|
|
pub async fn selected_pair_is_direct(pc: &Arc<RTCPeerConnection>) -> bool {
|
|
let dtls = pc.sctp().transport();
|
|
let ice = dtls.ice_transport();
|
|
match ice.get_selected_candidate_pair().await {
|
|
Some(pair) => {
|
|
pair.local.typ != RTCIceCandidateType::Relay
|
|
&& pair.remote.typ != RTCIceCandidateType::Relay
|
|
}
|
|
None => false,
|
|
}
|
|
}
|
|
|
|
/// The web/Python 3-channel DataChannel topology. Labels + options match
|
|
/// `app.js` exactly so a native peer interops with the browser and the Python
|
|
/// host: `media` carries video (unreliable, 150 ms lifetime — drop beats late),
|
|
/// `ctl` carries input/clipboard/file/keyframe (reliable, ordered), `move`
|
|
/// carries pointer motion (unreliable, no retransmit).
|
|
pub struct WebChannels {
|
|
pub media: Arc<RTCDataChannel>,
|
|
pub ctl: Arc<RTCDataChannel>,
|
|
pub mov: Arc<RTCDataChannel>,
|
|
}
|
|
|
|
/// Offerer: create the three web channels on `pc`. Call BEFORE `offer_nontrickle`
|
|
/// so they are announced in the SDP. The option set is byte-for-byte the web
|
|
/// client's (`media`{ordered:false,maxPacketLifeTime:150},
|
|
/// `ctl`{ordered:true}, `move`{ordered:false,maxRetransmits:0}).
|
|
pub async fn create_web_channels(pc: &Arc<RTCPeerConnection>) -> Result<WebChannels> {
|
|
let media = pc
|
|
.create_data_channel(
|
|
"media",
|
|
Some(RTCDataChannelInit {
|
|
ordered: Some(false),
|
|
max_packet_life_time: Some(150),
|
|
..Default::default()
|
|
}),
|
|
)
|
|
.await?;
|
|
let ctl = pc
|
|
.create_data_channel(
|
|
"ctl",
|
|
Some(RTCDataChannelInit {
|
|
ordered: Some(true),
|
|
..Default::default()
|
|
}),
|
|
)
|
|
.await?;
|
|
let mov = pc
|
|
.create_data_channel(
|
|
"move",
|
|
Some(RTCDataChannelInit {
|
|
ordered: Some(false),
|
|
max_retransmits: Some(0),
|
|
..Default::default()
|
|
}),
|
|
)
|
|
.await?;
|
|
Ok(WebChannels { media, ctl, mov })
|
|
}
|
|
|
|
/// Like [`build_api`] but with DETACHED DataChannels enabled. In detach mode we
|
|
/// read each channel with our OWN buffer instead of webrtc-rs's fixed 64 KiB
|
|
/// `read_loop` buffer — which is the ONLY way to receive a whole >64 KiB
|
|
/// DataChannel message on webrtc-rs 0.17.2. This is required for the web/Python
|
|
/// `media` channel, where the browser / aiortc send whole ~100-200 KiB video
|
|
/// frames. Use this API for the web 3-channel topology; [`build_api`] (on_message,
|
|
/// 64 KiB) stays the default for the native single channel, which fragments at
|
|
/// 16 KiB and never needs a large receive.
|
|
pub fn build_api_web() -> Result<API> {
|
|
use webrtc::api::setting_engine::{SctpMaxMessageSize, SettingEngine};
|
|
let mut media = MediaEngine::default();
|
|
let mut registry = Registry::new();
|
|
registry = register_default_interceptors(registry, &mut media)?;
|
|
let mut se = SettingEngine::default();
|
|
se.set_sctp_max_message_size_can_send(SctpMaxMessageSize::Bounded(MAX_MESSAGE_SIZE));
|
|
se.detach_data_channels();
|
|
Ok(APIBuilder::new()
|
|
.with_media_engine(media)
|
|
.with_interceptor_registry(registry)
|
|
.with_setting_engine(se)
|
|
.build())
|
|
}
|
|
|
|
/// The web DataChannels the viewer actually uses, after detaching (raw
|
|
/// read/write handles with our own buffer — the only way to receive whole
|
|
/// >64KiB frames on webrtc-rs 0.17.2). `move` is created for topology/interop
|
|
/// but the viewer never reads/writes it, so it is not detached (fewer detaches
|
|
/// = more reliable).
|
|
pub struct WebDetached {
|
|
pub media: Arc<webrtc::data::data_channel::DataChannel>,
|
|
pub ctl: Arc<webrtc::data::data_channel::DataChannel>,
|
|
}
|
|
|
|
/// Viewer offerer, WEB topology (interop with the browser / Python host). Builds
|
|
/// the 3 channels (media/ctl/move, labels+options per app.js), runs the
|
|
/// non-trickle rtc.offer/answer handshake over `send_text`+`rtc_in` (offer does
|
|
/// NOT carry `native` — it must look like a standard web offer so the Python
|
|
/// host accepts it), waits for all 3 to open, then detaches them SEQUENTIALLY
|
|
/// (concurrent multi-channel detach deadlocks) and returns the raw handles. The
|
|
/// pc must be built from [`build_api_web`] (detach enabled).
|
|
pub async fn upgrade_offerer_web(
|
|
api: &API,
|
|
sid: &str,
|
|
ice_override: Option<Vec<RtcIceServerJson>>,
|
|
send_text: &(dyn Fn(String) + Send + Sync),
|
|
rtc_in: &mut mpsc::UnboundedReceiver<Value>,
|
|
) -> Result<(Arc<RTCPeerConnection>, WebDetached)> {
|
|
let ice = match ice_override {
|
|
Some(i) => i,
|
|
None => {
|
|
send_text(json!({ "t": "rtc.config.req", "sid": sid }).to_string());
|
|
match wait_rtc(rtc_in, "rtc.config", CONFIG_CAP).await {
|
|
Ok(v) => v
|
|
.get("iceServers")
|
|
.and_then(|x| serde_json::from_value::<Vec<RtcIceServerJson>>(x.clone()).ok())
|
|
.unwrap_or_else(default_ice),
|
|
Err(_) => default_ice(),
|
|
}
|
|
}
|
|
};
|
|
|
|
let pc = new_pc(api, rtc_config(ice_servers_from_json(&ice))).await?;
|
|
let web = create_web_channels(&pc).await?;
|
|
|
|
// open signals for the three channels
|
|
let (open_tx, mut open_rx) = mpsc::unbounded_channel::<&'static str>();
|
|
for (lbl, ch) in [
|
|
("media", &web.media),
|
|
("ctl", &web.ctl),
|
|
("move", &web.mov),
|
|
] {
|
|
let tx = open_tx.clone();
|
|
ch.on_open(Box::new(move || {
|
|
let tx = tx.clone();
|
|
Box::pin(async move {
|
|
let _ = tx.send(lbl);
|
|
})
|
|
}));
|
|
}
|
|
|
|
let offer_sdp = offer_nontrickle(&pc, GATHER_CAP).await?;
|
|
// NOTE: no `native` marker — this is a standard web offer.
|
|
send_text(json!({ "t": "rtc.offer", "sid": sid, "sdp": offer_sdp, "ice": ice }).to_string());
|
|
let answer = wait_rtc(rtc_in, "rtc.answer", ANSWER_CAP).await?;
|
|
let answer_sdp = answer
|
|
.get("sdp")
|
|
.and_then(|s| s.as_str())
|
|
.ok_or_else(|| anyhow!("rtc.answer missing sdp"))?;
|
|
apply_answer(&pc, answer_sdp).await?;
|
|
|
|
// wait for the two channels we use (media+ctl) to open, capped.
|
|
let mut opened = std::collections::HashSet::new();
|
|
let deadline = tokio::time::Instant::now() + CONNECT_CAP;
|
|
while !(opened.contains("media") && opened.contains("ctl")) {
|
|
match tokio::time::timeout_at(deadline, open_rx.recv()).await {
|
|
Ok(Some(l)) => {
|
|
opened.insert(l);
|
|
}
|
|
_ => return Err(anyhow!("web channels did not open")),
|
|
}
|
|
}
|
|
|
|
// detach SEQUENTIALLY (concurrent detach deadlocks) — only media + ctl.
|
|
let media = web.media.detach().await?;
|
|
let ctl = web.ctl.detach().await?;
|
|
Ok((pc, WebDetached { media, ctl }))
|
|
}
|
|
|
|
/// Host answerer, WEB topology (a browser / web viewer connects to a native
|
|
/// host). Builds the answerer pc, binds the viewer-created media/ctl/move
|
|
/// channels by label, runs the non-trickle answer, waits for media+ctl to open,
|
|
/// then detaches them SEQUENTIALLY. The pc must be built from [`build_api_web`].
|
|
/// The host then SENDS video on media (whole frames, cap 256KiB) and READS
|
|
/// input on ctl.
|
|
pub async fn answer_offer_web(
|
|
api: &API,
|
|
sid: &str,
|
|
offer_sdp: &str,
|
|
ice: Vec<RtcIceServerJson>,
|
|
send_text: &(dyn Fn(String) + Send + Sync),
|
|
) -> Result<(Arc<RTCPeerConnection>, WebDetached)> {
|
|
use std::collections::HashMap;
|
|
use std::sync::Mutex as StdMutex;
|
|
|
|
let pc = new_pc(api, rtc_config(ice_servers_from_json(&ice))).await?;
|
|
|
|
// bind channels by label + signal opens
|
|
let chans: Arc<StdMutex<HashMap<String, Arc<RTCDataChannel>>>> =
|
|
Arc::new(StdMutex::new(HashMap::new()));
|
|
let (open_tx, mut open_rx) = mpsc::unbounded_channel::<String>();
|
|
{
|
|
let chans = chans.clone();
|
|
let open_tx = open_tx.clone();
|
|
pc.on_data_channel(Box::new(move |d: Arc<RTCDataChannel>| {
|
|
let chans = chans.clone();
|
|
let open_tx = open_tx.clone();
|
|
let label = d.label().to_string();
|
|
Box::pin(async move {
|
|
chans.lock().unwrap().insert(label.clone(), d.clone());
|
|
let tx = open_tx.clone();
|
|
let l = label.clone();
|
|
d.on_open(Box::new(move || {
|
|
let tx = tx.clone();
|
|
let l = l.clone();
|
|
Box::pin(async move {
|
|
let _ = tx.send(l);
|
|
})
|
|
}));
|
|
})
|
|
}));
|
|
}
|
|
|
|
let answer_sdp = answer_nontrickle(&pc, offer_sdp, GATHER_CAP).await?;
|
|
send_text(json!({ "t": "rtc.answer", "sid": sid, "sdp": answer_sdp }).to_string());
|
|
|
|
// wait for media + ctl to open, capped
|
|
let mut opened = std::collections::HashSet::new();
|
|
let deadline = tokio::time::Instant::now() + CONNECT_CAP;
|
|
while !(opened.contains("media") && opened.contains("ctl")) {
|
|
match tokio::time::timeout_at(deadline, open_rx.recv()).await {
|
|
Ok(Some(l)) => {
|
|
opened.insert(l);
|
|
}
|
|
_ => return Err(anyhow!("web channels (media/ctl) did not open")),
|
|
}
|
|
}
|
|
|
|
// grab wrappers, then detach SEQUENTIALLY (concurrent detach deadlocks).
|
|
let (media_w, ctl_w) = {
|
|
let m = chans.lock().unwrap();
|
|
(
|
|
m.get("media").cloned().ok_or_else(|| anyhow!("no media channel"))?,
|
|
m.get("ctl").cloned().ok_or_else(|| anyhow!("no ctl channel"))?,
|
|
)
|
|
};
|
|
let media = media_w.detach().await?;
|
|
let ctl = ctl_w.detach().await?;
|
|
Ok((pc, WebDetached { media, ctl }))
|
|
}
|