#![cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
use std::collections::HashMap;
use std::net::IpAddr;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::{Arc, Once, RwLock as StdRwLock};
use anyhow::{Context, Result};
use bytes::Bytes;
use tokio::sync::{mpsc, oneshot, Mutex, RwLock};
use webrtc::api::setting_engine::SettingEngine;
use webrtc::api::APIBuilder;
use webrtc::data_channel::data_channel_init::RTCDataChannelInit;
use webrtc::data_channel::data_channel_message::DataChannelMessage;
use webrtc::data_channel::RTCDataChannel;
use webrtc::ice::candidate::CandidateType;
use webrtc::ice::mdns::MulticastDnsMode;
use webrtc::ice::network_type::NetworkType;
use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit;
use webrtc::ice_transport::ice_candidate_type::RTCIceCandidateType;
use webrtc::ice_transport::ice_server::RTCIceServer;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::policy::ice_transport_policy::RTCIceTransportPolicy;
use webrtc::peer_connection::sdp::sdp_type::RTCSdpType;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
use webrtc::peer_connection::RTCPeerConnection;
use super::{NativeWebRTCState, WebRTCSignalMessage};
use crate::heartbeat::codec;
use crate::heartbeat::{HealthTransition, HeartbeatConfig, TransportHeartbeat};
const ICE_DISCONNECT_GRACE_MS: u64 = 12_000;
const WEBRTC_HEARTBEAT_LOG_ENV: &str = "OPENRTC_VERBOSE_WEBRTC_HEARTBEAT";
fn parse_multicast_dns_mode(value: &str) -> Option<MulticastDnsMode> {
match value.trim().to_ascii_lowercase().as_str() {
"disabled" | "disable" | "off" | "false" | "0" => Some(MulticastDnsMode::Disabled),
"query-only" | "query_only" | "queryonly" | "query" => Some(MulticastDnsMode::QueryOnly),
"query-and-gather" | "query_and_gather" | "queryandgather" | "gather" => {
Some(MulticastDnsMode::QueryAndGather)
}
_ => None,
}
}
fn configured_multicast_dns_mode(config: &crate::client::WebRTCConfig) -> MulticastDnsMode {
if config.privacy_mode {
return MulticastDnsMode::Disabled;
}
for key in ["OPENRTC_WEBRTC_MDNS_MODE", "PLUTO_WEBRTC_MDNS_MODE"] {
if let Ok(value) = std::env::var(key) {
if let Some(mode) = parse_multicast_dns_mode(&value) {
return mode;
}
}
}
#[cfg(target_os = "ios")]
{
MulticastDnsMode::Disabled
}
#[cfg(not(target_os = "ios"))]
{
if config.lan_mode {
MulticastDnsMode::QueryAndGather
} else {
MulticastDnsMode::QueryOnly
}
}
}
fn verbose_webrtc_heartbeat_logs() -> bool {
std::env::var(WEBRTC_HEARTBEAT_LOG_ENV)
.map(|value| {
matches!(
value.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
})
.unwrap_or(false)
}
fn host_has_routable_ipv6() -> bool {
use std::net::IpAddr;
use std::sync::OnceLock;
static CACHED: OnceLock<bool> = OnceLock::new();
*CACHED.get_or_init(|| {
let interfaces = netdev::get_interfaces();
interfaces.iter().any(|iface| {
iface.ipv6.iter().any(|net| {
let ip = IpAddr::V6(net.addr());
if ip.is_loopback() || ip.is_unspecified() {
return false;
}
let segments = match ip {
IpAddr::V6(v6) => v6.segments(),
_ => return false,
};
(segments[0] & 0xffc0) != 0xfe80
})
})
})
}
fn network_types_for_routable_ipv6(has_routable_ipv6: bool) -> Vec<NetworkType> {
let mut net_types = vec![NetworkType::Udp4, NetworkType::Tcp4];
if has_routable_ipv6 {
net_types.push(NetworkType::Udp6);
net_types.push(NetworkType::Tcp6);
}
net_types
}
#[derive(Debug, serde::Deserialize)]
struct SignalSessionDescription {
#[serde(rename = "type")]
sdp_type: String,
sdp: String,
}
pub type NativeWebRTCMessageHandler =
Arc<dyn Fn(Bytes) -> futures::future::BoxFuture<'static, ()> + Send + Sync>;
pub type NativeWebRTCMessageInterceptor =
Arc<dyn Fn(Bytes) -> futures::future::BoxFuture<'static, bool> + Send + Sync>;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NativeWebRTCIcePairSummary {
pub local_candidate_type: CandidateType,
pub remote_candidate_type: CandidateType,
pub local_ip: String,
pub remote_ip: String,
}
fn native_candidate_type(candidate_type: RTCIceCandidateType) -> Option<CandidateType> {
match candidate_type {
RTCIceCandidateType::Host => Some(CandidateType::Host),
RTCIceCandidateType::Srflx => Some(CandidateType::ServerReflexive),
RTCIceCandidateType::Prflx => Some(CandidateType::PeerReflexive),
RTCIceCandidateType::Relay => Some(CandidateType::Relay),
RTCIceCandidateType::Unspecified => None,
}
}
impl NativeWebRTCIcePairSummary {
pub fn is_lan_direct_pair(&self) -> bool {
self.local_candidate_type != CandidateType::Relay
&& self.remote_candidate_type != CandidateType::Relay
&& selected_address_is_local(&self.local_ip)
&& selected_address_is_local(&self.remote_ip)
}
pub fn transport_label(&self) -> &'static str {
if self.is_lan_direct_pair() {
crate::transport_label::WEBRTC_LAN
} else if self.local_candidate_type == CandidateType::Relay
|| self.remote_candidate_type == CandidateType::Relay
{
crate::transport_label::WEBRTC_TURN
} else {
crate::transport_label::WEBRTC
}
}
}
fn selected_address_is_local(address: &str) -> bool {
match address.trim().parse::<IpAddr>() {
Ok(IpAddr::V4(address)) => {
address.is_private() || address.is_link_local() || address.is_loopback()
}
Ok(IpAddr::V6(address)) => {
address.is_unique_local() || address.is_unicast_link_local() || address.is_loopback()
}
Err(_) => false,
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NativeWebRTCRole {
Auto,
Initiator,
Responder,
}
#[derive(Clone)]
pub struct NativeWebRTCDataChannel {
#[allow(dead_code)]
local_node_id: String,
#[allow(dead_code)]
remote_node_id: String,
is_initiator: bool,
state: Arc<AtomicU8>,
started: Arc<AtomicBool>,
peer_connection: Arc<RTCPeerConnection>,
data_channel: Arc<RwLock<Option<Arc<RTCDataChannel>>>>,
message_handler: Arc<RwLock<Option<NativeWebRTCMessageHandler>>>,
message_interceptors: Arc<RwLock<Vec<NativeWebRTCMessageInterceptor>>>,
pending_messages: Arc<Mutex<Vec<Bytes>>>,
pending_candidates: Arc<Mutex<Vec<RTCIceCandidateInit>>>,
signal_sender: super::NativeWebRTCSignalSender,
negotiation_id: Arc<StdRwLock<String>>,
heartbeat_config: Arc<Mutex<Option<HeartbeatConfig>>>,
heartbeat_health_tx: Arc<Mutex<Option<mpsc::Sender<HealthTransition>>>>,
heartbeat_transport_name: Arc<Mutex<String>>,
heartbeat_pong_tx: Arc<Mutex<Option<mpsc::Sender<codec::Pong>>>>,
heartbeat_ping_tx: Arc<Mutex<Option<mpsc::Sender<codec::Ping>>>>,
heartbeat_handle: Arc<Mutex<Option<TransportHeartbeat>>>,
latency_waiters: Arc<Mutex<HashMap<u64, oneshot::Sender<codec::Pong>>>>,
ice_restart_attempted: Arc<AtomicBool>,
ice_grace_timer: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
}
impl NativeWebRTCDataChannel {
pub async fn new(
local_node_id: &str,
remote_node_id: &str,
config: &crate::client::WebRTCConfig,
signal_sender: super::NativeWebRTCSignalSender,
negotiation_id: &str,
) -> Result<Self> {
Self::new_with_role(
local_node_id,
remote_node_id,
config,
signal_sender,
negotiation_id,
NativeWebRTCRole::Auto,
)
.await
}
pub async fn new_with_role(
local_node_id: &str,
remote_node_id: &str,
config: &crate::client::WebRTCConfig,
signal_sender: super::NativeWebRTCSignalSender,
negotiation_id: &str,
role: NativeWebRTCRole,
) -> Result<Self> {
anyhow::ensure!(
!local_node_id.trim().is_empty(),
"local_node_id is required"
);
anyhow::ensure!(
!remote_node_id.trim().is_empty(),
"remote_node_id is required"
);
anyhow::ensure!(
!negotiation_id.trim().is_empty(),
"negotiation_id is required"
);
ensure_rustls_crypto_provider();
let rtc_config = RTCConfiguration {
ice_servers: config
.ice_servers
.iter()
.map(|server| RTCIceServer {
urls: server.urls.clone(),
username: server.username.clone().unwrap_or_default(),
credential: server.credential.clone().unwrap_or_default(),
..Default::default()
})
.collect(),
ice_transport_policy: if config.privacy_mode {
RTCIceTransportPolicy::Relay
} else {
RTCIceTransportPolicy::All
},
..Default::default()
};
let mut setting_engine = SettingEngine::default();
setting_engine.set_network_types(network_types_for_routable_ipv6(host_has_routable_ipv6()));
setting_engine.set_ice_multicast_dns_mode(configured_multicast_dns_mode(config));
let api = APIBuilder::new()
.with_setting_engine(setting_engine)
.build();
let peer_connection = Arc::new(api.new_peer_connection(rtc_config).await?);
let is_initiator = match role {
NativeWebRTCRole::Auto => local_node_id > remote_node_id,
NativeWebRTCRole::Initiator => true,
NativeWebRTCRole::Responder => false,
};
Ok(Self {
local_node_id: local_node_id.to_string(),
remote_node_id: remote_node_id.to_string(),
is_initiator,
state: Arc::new(AtomicU8::new(0)),
started: Arc::new(AtomicBool::new(false)),
peer_connection,
data_channel: Arc::new(RwLock::new(None)),
message_handler: Arc::new(RwLock::new(None)),
message_interceptors: Arc::new(RwLock::new(Vec::new())),
pending_messages: Arc::new(Mutex::new(Vec::new())),
pending_candidates: Arc::new(Mutex::new(Vec::new())),
signal_sender,
negotiation_id: Arc::new(StdRwLock::new(negotiation_id.to_string())),
heartbeat_config: Arc::new(Mutex::new(None)),
heartbeat_health_tx: Arc::new(Mutex::new(None)),
heartbeat_transport_name: Arc::new(Mutex::new(String::new())),
heartbeat_pong_tx: Arc::new(Mutex::new(None)),
heartbeat_ping_tx: Arc::new(Mutex::new(None)),
heartbeat_handle: Arc::new(Mutex::new(None)),
latency_waiters: Arc::new(Mutex::new(HashMap::new())),
ice_restart_attempted: Arc::new(AtomicBool::new(false)),
ice_grace_timer: Arc::new(Mutex::new(None)),
})
}
pub async fn with_heartbeat(
&self,
transport_name: String,
config: HeartbeatConfig,
health_tx: mpsc::Sender<HealthTransition>,
) {
*self.heartbeat_transport_name.lock().await = transport_name;
*self.heartbeat_config.lock().await = Some(config);
*self.heartbeat_health_tx.lock().await = Some(health_tx);
}
pub fn negotiation_id(&self) -> String {
self.negotiation_id
.read()
.expect("negotiation_id lock poisoned")
.clone()
}
pub fn is_initiator(&self) -> bool {
self.is_initiator
}
async fn dispatch_or_buffer_message(
handler: &Arc<RwLock<Option<NativeWebRTCMessageHandler>>>,
interceptors: &Arc<RwLock<Vec<NativeWebRTCMessageInterceptor>>>,
pending_messages: &Arc<Mutex<Vec<Bytes>>>,
pong_tx: &Arc<Mutex<Option<mpsc::Sender<codec::Pong>>>>,
ping_tx: &Arc<Mutex<Option<mpsc::Sender<codec::Ping>>>>,
latency_waiters: &Arc<Mutex<HashMap<u64, oneshot::Sender<codec::Pong>>>>,
message: Bytes,
) {
if let Some(&first_byte) = message.first() {
if first_byte == codec::TYPE_PING {
let decoded = codec::decode_ping(&message[1..]);
let guard = ping_tx.lock().await;
let has_tx = guard.is_some();
if verbose_webrtc_heartbeat_logs() {
println!(
"[NativeWebRTC] ping received byte_len={} decoded={} has_ping_tx={}",
message.len(),
decoded.is_some(),
has_tx
);
}
if let Some(ping) = decoded {
if let Some(tx) = guard.as_ref() {
let _ = tx.try_send(ping);
}
}
return;
}
if first_byte == codec::TYPE_PONG {
if let Some(pong) = codec::decode_pong(&message[1..]) {
if let Some(waiter) = latency_waiters.lock().await.remove(&pong.seq) {
let _ = waiter.send(pong);
return;
}
let guard = pong_tx.lock().await;
if let Some(tx) = guard.as_ref() {
let _ = tx.try_send(pong);
}
}
return;
}
}
let interceptor_callbacks = {
let guard = interceptors.read().await;
guard.clone()
};
for interceptor in interceptor_callbacks {
if interceptor(message.clone()).await {
return;
}
}
let callback = {
let guard = handler.read().await;
guard.clone()
};
if let Some(callback) = callback {
callback(message).await;
return;
}
let mut pending = pending_messages.lock().await;
pending.push(message);
}
pub fn state(&self) -> NativeWebRTCState {
match self.state.load(Ordering::SeqCst) {
1 => NativeWebRTCState::Connecting,
2 => NativeWebRTCState::Connected,
3 => NativeWebRTCState::Failed,
4 => NativeWebRTCState::Closed,
_ => NativeWebRTCState::Idle,
}
}
#[cfg(test)]
pub(crate) fn force_state_for_test(&self, state: NativeWebRTCState) {
let value = match state {
NativeWebRTCState::Idle => 0,
NativeWebRTCState::Connecting => 1,
NativeWebRTCState::Connected => 2,
NativeWebRTCState::Failed => 3,
NativeWebRTCState::Closed => 4,
};
self.state.store(value, Ordering::SeqCst);
}
pub async fn has_remote_description(&self) -> bool {
self.peer_connection.remote_description().await.is_some()
}
pub async fn start(&self) -> Result<()> {
if self.started.swap(true, Ordering::SeqCst) {
return Ok(());
}
self.state.store(1, Ordering::SeqCst);
println!(
"[NativeWebRTC] start local_node_id={} remote_node_id={} role={}",
self.local_node_id,
self.remote_node_id,
if self.is_initiator {
"initiator"
} else {
"responder"
}
);
let pc_state = self.state.clone();
let hb_config = self.heartbeat_config.clone();
let hb_health_tx = self.heartbeat_health_tx.clone();
let hb_transport_name = self.heartbeat_transport_name.clone();
let hb_pong_tx_slot = self.heartbeat_pong_tx.clone();
let hb_ping_tx_slot = self.heartbeat_ping_tx.clone();
let hb_handle_slot = self.heartbeat_handle.clone();
let hb_data_channel = self.data_channel.clone();
let ice_restart_flag = self.ice_restart_attempted.clone();
let ice_grace_timer_slot = self.ice_grace_timer.clone();
let pc_for_restart = self.peer_connection.clone();
let is_initiator_for_restart = self.is_initiator;
let signal_sender_for_restart = self.signal_sender.clone();
let negotiation_id_for_restart = self.negotiation_id.clone();
self.peer_connection
.on_peer_connection_state_change(Box::new(
move |next_state: RTCPeerConnectionState| {
let state = pc_state.clone();
let config_arc = hb_config.clone();
let health_tx_arc = hb_health_tx.clone();
let name_arc = hb_transport_name.clone();
let pong_tx_slot = hb_pong_tx_slot.clone();
let ping_tx_slot = hb_ping_tx_slot.clone();
let handle_slot = hb_handle_slot.clone();
let dc_slot = hb_data_channel.clone();
let restart_flag = ice_restart_flag.clone();
let grace_slot = ice_grace_timer_slot.clone();
let pc_restart = pc_for_restart.clone();
let is_initiator_clone = is_initiator_for_restart;
let signal_sender_clone = signal_sender_for_restart.clone();
let negotiation_id_clone = negotiation_id_for_restart.clone();
Box::pin(async move {
println!("[NativeWebRTC] peer_connection_state={:?}", next_state);
match next_state {
RTCPeerConnectionState::Connected => {
state.store(2, Ordering::SeqCst);
restart_flag.store(false, Ordering::SeqCst);
if let Some(handle) = grace_slot.lock().await.take() {
handle.abort();
println!("[NativeWebRTC] ice-recovered — grace timer cancelled");
}
let config = config_arc.lock().await.clone();
let health_tx = health_tx_arc.lock().await.clone();
if let (Some(config), Some(health_tx)) = (config, health_tx) {
let transport_name = name_arc.lock().await.clone();
let dc_slot_for_send = dc_slot.clone();
let dc_slot_for_pong = dc_slot.clone();
let send_fn: std::sync::Arc<dyn Fn(Bytes) -> futures::future::BoxFuture<'static, anyhow::Result<()>> + Send + Sync> =
std::sync::Arc::new(move |frame: Bytes| {
let dc2 = dc_slot_for_send.clone();
Box::pin(async move {
if let Some(channel) = dc2.read().await.as_ref() {
channel.send(&frame).await.map_err(|e| anyhow::anyhow!("{e}"))?;
}
Ok(())
})
});
let (pong_tx, mut pong_rx) = mpsc::channel::<codec::Pong>(32);
let (ping_tx, mut ping_rx) = mpsc::channel::<codec::Ping>(32);
let hb = TransportHeartbeat::spawn(
transport_name,
send_fn,
health_tx,
config,
true, );
let hb_pong = hb.pong_sender();
tokio::spawn(async move {
while let Some(pong) = pong_rx.recv().await {
let _ = hb_pong.try_send(pong);
}
});
tokio::spawn(async move {
while let Some(ping) = ping_rx.recv().await {
let recv_ms = codec::now_unix_ms();
let pong_frame = codec::build_webrtc_pong(&ping, recv_ms);
let channel_opt = dc_slot_for_pong.read().await.as_ref().cloned();
match channel_opt {
Some(channel) => {
match channel.send(&Bytes::from(pong_frame)).await {
Ok(_) => {
if verbose_webrtc_heartbeat_logs() {
println!("[NativeWebRTC] pong sent seq={}", ping.seq);
}
}
Err(e) => println!("[NativeWebRTC] pong send failed seq={} err={}", ping.seq, e),
}
}
None => {
println!("[NativeWebRTC] pong send skipped seq={} reason=no-data-channel", ping.seq);
}
}
}
});
*pong_tx_slot.lock().await = Some(pong_tx);
*ping_tx_slot.lock().await = Some(ping_tx);
*handle_slot.lock().await = Some(hb);
}
}
RTCPeerConnectionState::Failed => {
state.store(3, Ordering::SeqCst);
if let Some(handle) = grace_slot.lock().await.take() {
handle.abort();
}
stop_heartbeat(&handle_slot, &pong_tx_slot, &ping_tx_slot).await;
}
RTCPeerConnectionState::Closed => {
state.store(4, Ordering::SeqCst);
if let Some(handle) = grace_slot.lock().await.take() {
handle.abort();
}
stop_heartbeat(&handle_slot, &pong_tx_slot, &ping_tx_slot).await;
}
RTCPeerConnectionState::Disconnected => {
let prior_state = state.load(Ordering::SeqCst);
if prior_state == 3 || prior_state == 4 {
println!(
"[NativeWebRTC] ice-restart skipped: session already terminal state={}",
prior_state
);
return;
}
let already = restart_flag.swap(true, Ordering::SeqCst);
if !already {
println!("[NativeWebRTC] ice-restart-attempted reason=pc-disconnected");
pc_restart.restart_ice().await.ok();
if is_initiator_clone {
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
match pc_restart.create_offer(None).await {
Ok(offer) => {
if let Err(e) = pc_restart
.set_local_description(offer.clone())
.await
{
println!(
"[NativeWebRTC] ice-restart set_local_description failed err={}",
e
);
} else {
let negotiation_id_value = negotiation_id_clone
.read()
.expect("negotiation_id lock poisoned")
.clone();
send_sdp_envelope(
&signal_sender_clone,
negotiation_id_value.as_str(),
&offer,
)
.await;
println!(
"[NativeWebRTC] ice-restart re-offer sent negotiation_id={}",
negotiation_id_value
);
}
let _ = std::marker::PhantomData::<RTCSessionDescription>;
}
Err(e) => {
println!(
"[NativeWebRTC] ice-restart create_offer failed err={}",
e
);
}
}
}
}
let mut guard = grace_slot.lock().await;
if guard.is_none() {
let pc_check = pc_restart.clone();
let state_for_timer = state.clone();
let grace_slot_inner = grace_slot.clone();
let handle_slot_timer = handle_slot.clone();
let pong_tx_slot_timer = pong_tx_slot.clone();
let ping_tx_slot_timer = ping_tx_slot.clone();
let handle = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(
ICE_DISCONNECT_GRACE_MS,
))
.await;
let current = pc_check.connection_state();
if matches!(
current,
RTCPeerConnectionState::Connected
) {
println!("[NativeWebRTC] ice-grace expired — recovered");
} else {
println!(
"[NativeWebRTC] ice-grace expired — marking failed (pc_state={:?})",
current
);
state_for_timer.store(3, Ordering::SeqCst);
stop_heartbeat(&handle_slot_timer, &pong_tx_slot_timer, &ping_tx_slot_timer).await;
let _ = pc_check.close().await;
}
*grace_slot_inner.lock().await = None;
});
*guard = Some(handle);
}
}
_ => {}
}
})
},
));
let dc_slot_incoming = self.data_channel.clone();
let incoming_open_state = self.state.clone();
let incoming_close_state = self.state.clone();
let incoming_error_state = self.state.clone();
let incoming_message_handler = self.message_handler.clone();
let incoming_message_interceptors = self.message_interceptors.clone();
let incoming_pending_messages = self.pending_messages.clone();
let incoming_pong_tx = self.heartbeat_pong_tx.clone();
let incoming_ping_tx = self.heartbeat_ping_tx.clone();
let incoming_latency_waiters = self.latency_waiters.clone();
self.peer_connection
.on_data_channel(Box::new(move |channel: Arc<RTCDataChannel>| {
let slot = dc_slot_incoming.clone();
let open_state = incoming_open_state.clone();
let close_state = incoming_close_state.clone();
let error_state = incoming_error_state.clone();
let message_handler = incoming_message_handler.clone();
let message_interceptors = incoming_message_interceptors.clone();
let pending_messages = incoming_pending_messages.clone();
let pong_tx = incoming_pong_tx.clone();
let ping_tx = incoming_ping_tx.clone();
let latency_waiters = incoming_latency_waiters.clone();
Box::pin(async move {
println!(
"[NativeWebRTC] on_data_channel label={} id={:?}",
channel.label(),
channel.id()
);
let open_state_inner = open_state.clone();
channel.on_open(Box::new(move || {
let state = open_state_inner.clone();
Box::pin(async move {
println!("[NativeWebRTC] incoming datachannel_open state=Connected");
state.store(2, Ordering::SeqCst);
})
}));
let close_state_inner = close_state.clone();
channel.on_close(Box::new(move || {
let state = close_state_inner.clone();
Box::pin(async move {
state.store(4, Ordering::SeqCst);
})
}));
let error_state_inner = error_state.clone();
channel.on_error(Box::new(move |_| {
let state = error_state_inner.clone();
Box::pin(async move {
state.store(3, Ordering::SeqCst);
})
}));
let message_handler_inner = message_handler.clone();
let message_interceptors_inner = message_interceptors.clone();
let pending_messages_inner = pending_messages.clone();
let pong_tx_inner = pong_tx.clone();
let ping_tx_inner = ping_tx.clone();
let latency_waiters_inner = latency_waiters.clone();
channel.on_message(Box::new(move |message: DataChannelMessage| {
let handler = message_handler_inner.clone();
let interceptors = message_interceptors_inner.clone();
let pending_messages = pending_messages_inner.clone();
let pong_tx = pong_tx_inner.clone();
let ping_tx = ping_tx_inner.clone();
let latency_waiters = latency_waiters_inner.clone();
Box::pin(async move {
Self::dispatch_or_buffer_message(
&handler,
&interceptors,
&pending_messages,
&pong_tx,
&ping_tx,
&latency_waiters,
message.data,
)
.await;
})
}));
let mut guard = slot.write().await;
if guard.is_none() {
*guard = Some(channel);
}
})
}));
let signal_sender = self.signal_sender.clone();
let negotiation_id = self.negotiation_id.clone();
self.peer_connection
.on_ice_candidate(Box::new(move |candidate| {
let sender = signal_sender.clone();
let negotiation_id = negotiation_id.clone();
Box::pin(async move {
let Some(candidate) = candidate else {
return;
};
let Ok(candidate_init) = candidate.to_json() else {
return;
};
let negotiation_id_value = negotiation_id
.read()
.expect("negotiation_id lock poisoned")
.clone();
let envelope = serde_json::json!({
"type": "#pluto-signal",
"content": {
"transport": "webrtc",
"type": "candidate",
"negotiationId": negotiation_id_value,
"candidate": candidate_init,
}
});
sender(envelope).await;
})
}));
let dc_init = RTCDataChannelInit {
negotiated: Some(0),
..Default::default()
};
let data_channel = self
.peer_connection
.create_data_channel("pluto-dc", Some(dc_init))
.await?;
let open_state = self.state.clone();
let open_channel = data_channel.clone();
data_channel.on_open(Box::new(move || {
let state = open_state.clone();
let _channel = open_channel.clone();
Box::pin(async move {
println!("[NativeWebRTC] datachannel_open state=Connected");
state.store(2, Ordering::SeqCst);
})
}));
let close_state = self.state.clone();
data_channel.on_close(Box::new(move || {
let state = close_state.clone();
Box::pin(async move {
println!("[NativeWebRTC] datachannel_close state=Closed");
state.store(4, Ordering::SeqCst);
})
}));
let error_state = self.state.clone();
let outgoing_pending_messages = self.pending_messages.clone();
data_channel.on_error(Box::new(move |_| {
let state = error_state.clone();
Box::pin(async move {
println!("[NativeWebRTC] datachannel_error state=Failed");
state.store(3, Ordering::SeqCst);
})
}));
let outgoing_message_handler = self.message_handler.clone();
let outgoing_message_interceptors = self.message_interceptors.clone();
let outgoing_pending_messages_inner = outgoing_pending_messages.clone();
let outgoing_pong_tx = self.heartbeat_pong_tx.clone();
let outgoing_ping_tx = self.heartbeat_ping_tx.clone();
let outgoing_latency_waiters = self.latency_waiters.clone();
data_channel.on_message(Box::new(move |message: DataChannelMessage| {
let handler = outgoing_message_handler.clone();
let interceptors = outgoing_message_interceptors.clone();
let pending_messages = outgoing_pending_messages_inner.clone();
let pong_tx = outgoing_pong_tx.clone();
let ping_tx = outgoing_ping_tx.clone();
let latency_waiters = outgoing_latency_waiters.clone();
Box::pin(async move {
Self::dispatch_or_buffer_message(
&handler,
&interceptors,
&pending_messages,
&pong_tx,
&ping_tx,
&latency_waiters,
message.data,
)
.await;
})
}));
{
let mut guard = self.data_channel.write().await;
*guard = Some(data_channel);
}
if self.is_initiator {
println!(
"[NativeWebRTC] role=initiator creating initial offer local_node_id={} remote_node_id={}",
self.local_node_id,
self.remote_node_id,
);
let offer = self.peer_connection.create_offer(None).await?;
self.peer_connection
.set_local_description(offer.clone())
.await?;
self.send_sdp_signal(&offer).await;
} else {
println!(
"[NativeWebRTC] role=responder awaiting remote offer local_node_id={} remote_node_id={}",
self.local_node_id,
self.remote_node_id,
);
self.send_renegotiate_signal().await;
}
Ok(())
}
pub async fn set_message_handler(&self, handler: NativeWebRTCMessageHandler) {
{
let mut guard = self.message_handler.write().await;
*guard = Some(handler.clone());
}
let pending = {
let mut guard = self.pending_messages.lock().await;
std::mem::take(&mut *guard)
};
if !pending.is_empty() {
println!(
"[NativeWebRTC] flushing {} buffered datachannel message(s) after handler registration",
pending.len()
);
}
for message in pending {
handler(message).await;
}
}
pub async fn add_message_interceptor(&self, interceptor: NativeWebRTCMessageInterceptor) {
let mut guard = self.message_interceptors.write().await;
guard.push(interceptor);
}
pub async fn handle_signal(&self, signal: WebRTCSignalMessage) -> Result<()> {
if signal.transport != "webrtc" {
return Ok(());
}
if let Some(signal_negotiation_id) = signal
.negotiation_id
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
let active_negotiation_id = self.negotiation_id();
if signal_negotiation_id != active_negotiation_id {
if !self.is_initiator
&& signal.signal_type == "sdp"
&& self.peer_connection.remote_description().await.is_none()
&& signal.sdp.as_ref().is_some_and(|value| {
value
.get("type")
.and_then(serde_json::Value::as_str)
.map(|t| t == "offer")
.unwrap_or(false)
})
{
if let Ok(mut guard) = self.negotiation_id.write() {
*guard = signal_negotiation_id.to_string();
}
} else {
println!(
"[NativeWebRTC] ignoring stale signal type={} signal_negotiation_id={} active_negotiation_id={}",
signal.signal_type,
signal_negotiation_id,
active_negotiation_id
);
return Ok(());
}
}
}
if self.state() == NativeWebRTCState::Idle {
self.state.store(1, Ordering::SeqCst);
}
println!(
"[NativeWebRTC] handle_signal type={} role={} local_node_id={} remote_node_id={}",
signal.signal_type,
if self.is_initiator {
"initiator"
} else {
"responder"
},
self.local_node_id,
self.remote_node_id,
);
match signal.signal_type.as_str() {
"sdp" => {
let sdp_value = signal
.sdp
.context("missing webrtc sdp payload for signal type=sdp")?;
let session = parse_session_description(sdp_value)?;
let remote_sdp_type = session.sdp_type;
self.peer_connection.set_remote_description(session).await?;
self.flush_pending_candidates().await?;
if !self.is_initiator && remote_sdp_type == RTCSdpType::Offer {
let answer = self.peer_connection.create_answer(None).await?;
self.peer_connection
.set_local_description(answer.clone())
.await?;
self.send_sdp_signal(&answer).await;
}
}
"candidate" => {
let candidate = signal
.candidate
.context("missing webrtc candidate payload for signal type=candidate")?;
let candidate_init = parse_ice_candidate_init(candidate)?;
if self.peer_connection.remote_description().await.is_some() {
self.peer_connection
.add_ice_candidate(candidate_init)
.await?;
} else {
self.pending_candidates.lock().await.push(candidate_init);
}
}
_ => {}
}
Ok(())
}
pub async fn send(&self, data: &[u8]) -> Result<()> {
anyhow::ensure!(
self.state() == NativeWebRTCState::Connected,
"native webrtc datachannel not connected"
);
let channel = {
let guard = self.data_channel.read().await;
guard
.as_ref()
.cloned()
.context("native webrtc datachannel handle is missing")?
};
channel.send(&Bytes::copy_from_slice(data)).await?;
Ok(())
}
pub async fn selected_ice_pair_summary(&self) -> Option<NativeWebRTCIcePairSummary> {
let sctp = self.peer_connection.sctp();
let dtls = sctp.transport();
let pair = dtls.ice_transport().get_selected_candidate_pair().await?;
Some(NativeWebRTCIcePairSummary {
local_candidate_type: native_candidate_type(pair.local.typ)?,
remote_candidate_type: native_candidate_type(pair.remote.typ)?,
local_ip: pair.local.address,
remote_ip: pair.remote.address,
})
}
pub async fn measure_latency(
&self,
timeout: std::time::Duration,
samples: usize,
) -> Result<u64> {
anyhow::ensure!(
self.state() == NativeWebRTCState::Connected,
"native webrtc datachannel not connected"
);
let sample_count = samples.clamp(1, 10);
let mut measured = Vec::with_capacity(sample_count);
for index in 0..sample_count {
let seq = next_latency_probe_seq(index);
let (tx, rx) = oneshot::channel();
self.latency_waiters.lock().await.insert(seq, tx);
let started = std::time::Instant::now();
let frame = codec::build_webrtc_ping(seq, codec::now_unix_ms());
if let Err(error) = self.send(&frame).await {
self.latency_waiters.lock().await.remove(&seq);
return Err(error);
}
match tokio::time::timeout(timeout, rx).await {
Ok(Ok(_pong)) => {
measured.push(started.elapsed().as_millis() as u64);
}
Ok(Err(_closed)) => {
self.latency_waiters.lock().await.remove(&seq);
anyhow::bail!("native webrtc latency waiter closed");
}
Err(_elapsed) => {
self.latency_waiters.lock().await.remove(&seq);
anyhow::bail!(
"native webrtc latency probe timed out after {}ms",
timeout.as_millis()
);
}
}
}
anyhow::ensure!(
!measured.is_empty(),
"native webrtc latency probe produced no samples"
);
Ok(measured.iter().sum::<u64>() / measured.len() as u64)
}
pub async fn latency_ms(&self) -> Option<u64> {
let guard = self.heartbeat_handle.lock().await;
let heartbeat = guard.as_ref()?;
heartbeat.latency_ms().await
}
pub fn close(&self) {
self.state.store(4, Ordering::SeqCst);
let handle_slot = self.heartbeat_handle.clone();
let pong_tx_slot = self.heartbeat_pong_tx.clone();
let ping_tx_slot = self.heartbeat_ping_tx.clone();
let grace_slot = self.ice_grace_timer.clone();
let data_channel_slot = self.data_channel.clone();
let message_handler_slot = self.message_handler.clone();
let pending_messages = self.pending_messages.clone();
let pending_candidates = self.pending_candidates.clone();
let latency_waiters = self.latency_waiters.clone();
let pc = self.peer_connection.clone();
if let Ok(handle) = tokio::runtime::Handle::try_current() {
handle.spawn(async move {
if let Some(timer) = grace_slot.lock().await.take() {
timer.abort();
}
stop_heartbeat(&handle_slot, &pong_tx_slot, &ping_tx_slot).await;
*data_channel_slot.write().await = None;
*message_handler_slot.write().await = None;
pending_messages.lock().await.clear();
pending_candidates.lock().await.clear();
latency_waiters.lock().await.clear();
let _ = pc.close().await;
});
}
}
async fn flush_pending_candidates(&self) -> Result<()> {
if self.peer_connection.remote_description().await.is_none() {
return Ok(());
}
let mut pending = self.pending_candidates.lock().await;
for candidate in pending.drain(..) {
self.peer_connection.add_ice_candidate(candidate).await?;
}
Ok(())
}
async fn send_sdp_signal(&self, description: &RTCSessionDescription) {
let negotiation_id = self.negotiation_id();
send_sdp_envelope(&self.signal_sender, &negotiation_id, description).await;
}
async fn send_renegotiate_signal(&self) {
let negotiation_id = self.negotiation_id();
let envelope = serde_json::json!({
"type": "#pluto-signal",
"content": {
"transport": "webrtc",
"type": "renegotiate",
"negotiationId": negotiation_id,
}
});
self.signal_sender.as_ref()(envelope).await;
}
}
fn next_latency_probe_seq(index: usize) -> u64 {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|value| value.as_nanos() as u64)
.unwrap_or_default();
now ^ ((std::process::id() as u64) << 32) ^ index as u64
}
async fn send_sdp_envelope(
signal_sender: &super::NativeWebRTCSignalSender,
negotiation_id: &str,
description: &RTCSessionDescription,
) {
let envelope = serde_json::json!({
"type": "#pluto-signal",
"content": {
"transport": "webrtc",
"type": "sdp",
"negotiationId": negotiation_id,
"sdp": {
"type": sdp_type_to_wire(&description.sdp_type),
"sdp": description.sdp.clone(),
}
}
});
signal_sender(envelope).await;
}
async fn stop_heartbeat(
handle_slot: &Arc<Mutex<Option<TransportHeartbeat>>>,
pong_tx_slot: &Arc<Mutex<Option<mpsc::Sender<codec::Pong>>>>,
ping_tx_slot: &Arc<Mutex<Option<mpsc::Sender<codec::Ping>>>>,
) {
if let Some(hb) = handle_slot.lock().await.take() {
hb.cancel();
}
pong_tx_slot.lock().await.take();
ping_tx_slot.lock().await.take();
}
fn ensure_rustls_crypto_provider() {
static INIT: Once = Once::new();
INIT.call_once(|| {
let _ = rustls::crypto::ring::default_provider().install_default();
});
}
fn parse_session_description(value: serde_json::Value) -> Result<RTCSessionDescription> {
let parsed: SignalSessionDescription =
serde_json::from_value(value).context("invalid webrtc session description payload")?;
match parsed.sdp_type.as_str() {
"offer" => RTCSessionDescription::offer(parsed.sdp)
.context("failed to build webrtc offer from signal"),
"answer" => RTCSessionDescription::answer(parsed.sdp)
.context("failed to build webrtc answer from signal"),
other => anyhow::bail!("unsupported webrtc sdp type from signal: {other}"),
}
}
fn parse_ice_candidate_init(value: serde_json::Value) -> Result<RTCIceCandidateInit> {
if let Ok(candidate) = serde_json::from_value::<RTCIceCandidateInit>(value.clone()) {
return Ok(candidate);
}
let normalized = normalize_ice_candidate_payload(value)?;
serde_json::from_value(normalized).context("invalid webrtc ICE candidate payload")
}
fn normalize_ice_candidate_payload(value: serde_json::Value) -> Result<serde_json::Value> {
use serde_json::{Map, Value};
let source = value
.as_object()
.context("invalid webrtc ICE candidate payload: expected object")?;
let nested_candidate = source
.get("candidate")
.and_then(|candidate| candidate.as_object());
let candidate_obj = nested_candidate.unwrap_or(source);
let mut normalized = Map::new();
if let Some(candidate) = candidate_obj.get("candidate").cloned() {
normalized.insert("candidate".to_string(), candidate);
}
let sdp_mid = candidate_obj
.get("sdpMid")
.cloned()
.or_else(|| candidate_obj.get("sdp_mid").cloned());
if let Some(value) = sdp_mid {
normalized.insert("sdpMid".to_string(), value.clone());
normalized.insert("sdp_mid".to_string(), value);
}
let sdp_mline_index = candidate_obj
.get("sdpMLineIndex")
.cloned()
.or_else(|| candidate_obj.get("sdp_mline_index").cloned());
if let Some(value) = sdp_mline_index {
normalized.insert("sdpMLineIndex".to_string(), value.clone());
normalized.insert("sdp_mline_index".to_string(), value);
}
let username_fragment = candidate_obj
.get("usernameFragment")
.cloned()
.or_else(|| candidate_obj.get("username_fragment").cloned());
if let Some(value) = username_fragment {
normalized.insert("usernameFragment".to_string(), value.clone());
normalized.insert("username_fragment".to_string(), value);
}
if !normalized.contains_key("candidate") {
anyhow::bail!("invalid webrtc ICE candidate payload: missing candidate");
}
Ok(Value::Object(normalized))
}
fn sdp_type_to_wire(sdp_type: &RTCSdpType) -> &'static str {
match sdp_type {
RTCSdpType::Offer => "offer",
RTCSdpType::Pranswer => "pranswer",
RTCSdpType::Answer => "answer",
RTCSdpType::Rollback => "rollback",
RTCSdpType::Unspecified => "offer",
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
use crate::client::WebRTCConfig;
use tokio::sync::mpsc;
fn parse_signal_frame(frame: serde_json::Value) -> Option<WebRTCSignalMessage> {
if frame.get("type").and_then(|value| value.as_str()) != Some("#pluto-signal") {
return None;
}
let content = frame.get("content")?;
Some(WebRTCSignalMessage {
transport: content
.get("transport")
.and_then(|value| value.as_str())
.unwrap_or("webrtc")
.to_string(),
signal_type: content
.get("type")
.and_then(|value| value.as_str())
.unwrap_or("candidate")
.to_string(),
negotiation_id: content
.get("negotiationId")
.and_then(|value| value.as_str())
.map(ToOwned::to_owned),
sdp: content.get("sdp").cloned(),
candidate: content.get("candidate").cloned(),
})
}
fn noop_sender() -> super::super::NativeWebRTCSignalSender {
Arc::new(|_| Box::pin(async move {}))
}
#[test]
fn network_type_selection_only_adds_ipv6_when_routable() {
assert_eq!(
network_types_for_routable_ipv6(false),
vec![NetworkType::Udp4, NetworkType::Tcp4]
);
assert_eq!(
network_types_for_routable_ipv6(true),
vec![
NetworkType::Udp4,
NetworkType::Tcp4,
NetworkType::Udp6,
NetworkType::Tcp6,
]
);
}
#[test]
fn parses_multicast_dns_mode_overrides() {
assert_eq!(
parse_multicast_dns_mode("disabled"),
Some(MulticastDnsMode::Disabled)
);
assert_eq!(
parse_multicast_dns_mode("query-only"),
Some(MulticastDnsMode::QueryOnly)
);
assert_eq!(
parse_multicast_dns_mode("query-and-gather"),
Some(MulticastDnsMode::QueryAndGather)
);
assert_eq!(parse_multicast_dns_mode("surprise"), None);
}
#[test]
fn privacy_mode_disables_multicast_dns() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: true,
lan_mode: false,
};
assert_eq!(
configured_multicast_dns_mode(&config),
MulticastDnsMode::Disabled
);
}
#[test]
fn transport_label_classifies_selected_ice_pairs_by_route_kind() {
let pair = |local_candidate_type, remote_candidate_type| NativeWebRTCIcePairSummary {
local_candidate_type,
remote_candidate_type,
local_ip: "192.168.1.10".to_string(),
remote_ip: "192.168.1.11".to_string(),
};
assert_eq!(
pair(CandidateType::Host, CandidateType::Host).transport_label(),
crate::transport_label::WEBRTC_LAN
);
assert_eq!(
pair(CandidateType::Relay, CandidateType::Host).transport_label(),
crate::transport_label::WEBRTC_TURN
);
assert_eq!(
pair(CandidateType::ServerReflexive, CandidateType::PeerReflexive).transport_label(),
crate::transport_label::WEBRTC_LAN
);
assert_eq!(
pair(CandidateType::PeerReflexive, CandidateType::Host).transport_label(),
crate::transport_label::WEBRTC_LAN
);
let public_pair = NativeWebRTCIcePairSummary {
local_candidate_type: CandidateType::PeerReflexive,
remote_candidate_type: CandidateType::ServerReflexive,
local_ip: "203.0.113.10".to_string(),
remote_ip: "198.51.100.11".to_string(),
};
assert_eq!(
public_pair.transport_label(),
crate::transport_label::WEBRTC
);
let private_relay_pair = NativeWebRTCIcePairSummary {
local_candidate_type: CandidateType::Relay,
remote_candidate_type: CandidateType::Host,
local_ip: "192.168.1.10".to_string(),
remote_ip: "192.168.1.11".to_string(),
};
assert_eq!(
private_relay_pair.transport_label(),
crate::transport_label::WEBRTC_TURN
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn starts_connecting_after_start() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
};
let ch = NativeWebRTCDataChannel::new(
"local-b",
"remote-a",
&config,
noop_sender(),
"negotiation-start",
)
.await
.unwrap();
ch.start().await.unwrap();
assert_eq!(ch.state(), NativeWebRTCState::Connecting);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responder_requests_initiator_offer_on_start() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
};
let (tx, mut rx) = mpsc::unbounded_channel::<serde_json::Value>();
let sender: super::super::NativeWebRTCSignalSender = Arc::new(move |signal| {
let tx = tx.clone();
Box::pin(async move {
let _ = tx.send(signal);
})
});
let ch = NativeWebRTCDataChannel::new(
"node-a",
"node-z",
&config,
sender,
"negotiation-responder",
)
.await
.unwrap();
ch.start().await.unwrap();
let frame = rx
.recv()
.await
.expect("responder should request renegotiate");
let signal = parse_signal_frame(frame).expect("renegotiate frame should parse");
assert_eq!(signal.signal_type, "renegotiate");
assert_eq!(
signal.negotiation_id.as_deref(),
Some("negotiation-responder")
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn explicit_responder_role_overrides_node_id_ordering() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
};
let (tx, mut rx) = mpsc::unbounded_channel::<serde_json::Value>();
let sender: super::super::NativeWebRTCSignalSender = Arc::new(move |signal| {
let tx = tx.clone();
Box::pin(async move {
let _ = tx.send(signal);
})
});
let ch = NativeWebRTCDataChannel::new_with_role(
"node-z",
"node-a",
&config,
sender,
"browser-offer-negotiation",
NativeWebRTCRole::Responder,
)
.await
.unwrap();
ch.start().await.unwrap();
let frame = rx
.recv()
.await
.expect("forced responder should request a browser offer");
let signal = parse_signal_frame(frame).expect("renegotiate frame should parse");
assert_eq!(signal.signal_type, "renegotiate");
assert!(!ch.is_initiator());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responder_adopts_first_offer_negotiation_id_once() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
};
let (offer_tx, mut offer_rx) = mpsc::unbounded_channel::<serde_json::Value>();
let initiator_sender: super::super::NativeWebRTCSignalSender = Arc::new(move |signal| {
let offer_tx = offer_tx.clone();
Box::pin(async move {
let _ = offer_tx.send(signal);
})
});
let initiator = NativeWebRTCDataChannel::new(
"node-z",
"node-a",
&config,
initiator_sender,
"offer-negotiation",
)
.await
.unwrap();
initiator.start().await.unwrap();
let offer = loop {
let frame = offer_rx.recv().await.expect("initiator should emit offer");
let Some(signal) = parse_signal_frame(frame) else {
continue;
};
if signal.signal_type == "sdp" {
break signal;
}
};
let responder = NativeWebRTCDataChannel::new(
"node-a",
"node-z",
&config,
noop_sender(),
"responder-provisional",
)
.await
.unwrap();
responder.start().await.unwrap();
responder.handle_signal(offer.clone()).await.unwrap();
assert_eq!(responder.negotiation_id(), "offer-negotiation");
let mut stale_offer = offer;
stale_offer.negotiation_id = Some("stale-offer".to_string());
responder.handle_signal(stale_offer).await.unwrap();
assert_eq!(
responder.negotiation_id(),
"offer-negotiation",
"after applying a remote description, later mismatched offers are stale"
);
initiator.close();
responder.close();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responder_ignores_stale_candidate_after_adopting_offer_negotiation_id() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
};
let (offer_tx, mut offer_rx) = mpsc::unbounded_channel::<serde_json::Value>();
let initiator_sender: super::super::NativeWebRTCSignalSender = Arc::new(move |signal| {
let offer_tx = offer_tx.clone();
Box::pin(async move {
let _ = offer_tx.send(signal);
})
});
let initiator = NativeWebRTCDataChannel::new(
"node-z",
"node-a",
&config,
initiator_sender,
"offer-negotiation",
)
.await
.unwrap();
initiator.start().await.unwrap();
let offer = loop {
let frame = offer_rx.recv().await.expect("initiator should emit offer");
let Some(signal) = parse_signal_frame(frame) else {
continue;
};
if signal.signal_type == "sdp" {
break signal;
}
};
let responder = NativeWebRTCDataChannel::new(
"node-a",
"node-z",
&config,
noop_sender(),
"responder-provisional",
)
.await
.unwrap();
responder.start().await.unwrap();
responder.handle_signal(offer).await.unwrap();
assert_eq!(responder.negotiation_id(), "offer-negotiation");
responder
.handle_signal(WebRTCSignalMessage {
transport: "webrtc".to_string(),
signal_type: "candidate".to_string(),
negotiation_id: Some("stale-offer".to_string()),
sdp: None,
candidate: Some(serde_json::json!({
"candidate": "this malformed candidate must never be parsed",
})),
})
.await
.expect("stale mismatched candidate should be ignored before parsing");
assert!(
responder.pending_candidates.lock().await.is_empty(),
"stale candidate must not be buffered into the live negotiation"
);
initiator.close();
responder.close();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn loopback_offer_answer_candidate_reaches_connected() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: true,
};
let (a_to_b_tx, mut a_to_b_rx) = mpsc::unbounded_channel::<serde_json::Value>();
let (b_to_a_tx, mut b_to_a_rx) = mpsc::unbounded_channel::<serde_json::Value>();
let left_candidates = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let right_candidates = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let left_candidates_for_sender = left_candidates.clone();
let sender_a: super::super::NativeWebRTCSignalSender = Arc::new(move |signal| {
if let Some(candidate) = signal
.get("content")
.and_then(|content| content.get("candidate"))
.and_then(|candidate| candidate.get("candidate"))
.and_then(|candidate| candidate.as_str())
{
left_candidates_for_sender
.lock()
.expect("left candidates lock")
.push(candidate.to_string());
}
let tx = a_to_b_tx.clone();
Box::pin(async move {
let _ = tx.send(signal);
})
});
let right_candidates_for_sender = right_candidates.clone();
let sender_b: super::super::NativeWebRTCSignalSender = Arc::new(move |signal| {
if let Some(candidate) = signal
.get("content")
.and_then(|content| content.get("candidate"))
.and_then(|candidate| candidate.get("candidate"))
.and_then(|candidate| candidate.as_str())
{
right_candidates_for_sender
.lock()
.expect("right candidates lock")
.push(candidate.to_string());
}
let tx = b_to_a_tx.clone();
Box::pin(async move {
let _ = tx.send(signal);
})
});
let left = Arc::new(
NativeWebRTCDataChannel::new(
"node-z",
"node-a",
&config,
sender_a,
"negotiation-loopback",
)
.await
.unwrap(),
);
let right = Arc::new(
NativeWebRTCDataChannel::new(
"node-a",
"node-z",
&config,
sender_b,
"negotiation-loopback",
)
.await
.unwrap(),
);
let right_for_relay = right.clone();
let relay_a_to_b = tokio::spawn(async move {
while let Some(frame) = a_to_b_rx.recv().await {
let Some(signal) = parse_signal_frame(frame) else {
continue;
};
let _ = right_for_relay.handle_signal(signal).await;
}
});
let left_for_relay = left.clone();
let relay_b_to_a = tokio::spawn(async move {
while let Some(frame) = b_to_a_rx.recv().await {
let Some(signal) = parse_signal_frame(frame) else {
continue;
};
let _ = left_for_relay.handle_signal(signal).await;
}
});
left.start().await.unwrap();
right.start().await.unwrap();
tokio::time::timeout(Duration::from_secs(10), async {
loop {
if left.state() == NativeWebRTCState::Connected
&& right.state() == NativeWebRTCState::Connected
{
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
})
.await
.expect("native webrtc loopback should connect");
left.send(b"hello-from-left").await.unwrap();
let exchanged_candidates = {
let mut candidates = left_candidates
.lock()
.expect("left candidates lock")
.clone();
candidates.extend(
right_candidates
.lock()
.expect("right candidates lock")
.iter()
.cloned(),
);
candidates
};
assert!(
!exchanged_candidates.is_empty(),
"native webrtc LAN test should exchange ICE candidates"
);
assert!(
exchanged_candidates
.iter()
.all(|candidate| candidate.contains(" typ host")),
"native webrtc LAN test should exchange only host candidates: {:?}",
exchanged_candidates
);
assert!(
exchanged_candidates
.iter()
.all(|candidate| !candidate.contains(" typ relay")),
"native webrtc LAN test should not exchange relay candidates: {:?}",
exchanged_candidates
);
left.close();
right.close();
relay_a_to_b.abort();
relay_b_to_a.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn ice_restart_state_defaults() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
};
let ch = NativeWebRTCDataChannel::new(
"local-b",
"remote-a",
&config,
noop_sender(),
"negotiation-ice-defaults",
)
.await
.unwrap();
assert!(
!ch.ice_restart_attempted.load(Ordering::SeqCst),
"ice_restart_attempted must default false"
);
assert!(
ch.ice_grace_timer.lock().await.is_none(),
"ice_grace_timer must default None"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn wrong_transport_signal_is_ignored() {
let config = WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
};
let ch = NativeWebRTCDataChannel::new(
"local-b",
"remote-a",
&config,
noop_sender(),
"negotiation-ignore",
)
.await
.unwrap();
ch.start().await.unwrap();
ch.handle_signal(WebRTCSignalMessage {
transport: "moq".to_string(),
signal_type: "sdp".to_string(),
negotiation_id: Some("negotiation-ignore".to_string()),
sdp: Some(serde_json::json!({ "type": "offer", "sdp": "v=0\r\n" })),
candidate: None,
})
.await
.unwrap();
assert_eq!(ch.state(), NativeWebRTCState::Connecting);
}
}