#![cfg(target_arch = "wasm32")]
use std::{
cell::{Cell, RefCell},
rc::Rc,
};
use js_sys::{ArrayBuffer, Uint8Array};
use wasm_bindgen::{closure::Closure, JsCast, JsValue};
use wasm_bindgen_futures::spawn_local;
use web_sys::{Event, MessageEvent, RtcDataChannel, RtcDataChannelType};
use crate::{
iroh_carrier::{
segment_packet, CarrierControl, CarrierFrame, CarrierFrameExpectation, CarrierReassembler,
CARRIER_CONTROL_PACKET_ID,
},
packet_carrier_transport::PacketCarrierSession,
};
const DATA_CHANNEL_MESSAGE_CEILING: usize = 16 * 1024;
const BUFFERED_AMOUNT_LOW_THRESHOLD: u32 = 256 * 1024;
const BUFFERED_AMOUNT_HARD_LIMIT: u32 = 1024 * 1024;
pub struct WasmWebRtcCarrierSession {
channel: RtcDataChannel,
_session: Rc<PacketCarrierSession>,
_on_message: Closure<dyn FnMut(MessageEvent)>,
_on_buffered_low: Closure<dyn FnMut(Event)>,
_on_close: Closure<dyn FnMut(Event)>,
expected: CarrierFrameExpectation,
application_key: [u8; 32],
terminal_acknowledged: Rc<Cell<bool>>,
}
impl std::fmt::Debug for WasmWebRtcCarrierSession {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("WasmWebRtcCarrierSession")
.field("label", &self.channel.label())
.field("ordered", &data_channel_ordered(&self.channel))
.field("max_retransmits", &self.channel.max_retransmits())
.finish_non_exhaustive()
}
}
impl WasmWebRtcCarrierSession {
pub fn attach(
channel: RtcDataChannel,
session: PacketCarrierSession,
expected: CarrierFrameExpectation,
application_key: [u8; 32],
on_terminal: Rc<dyn Fn(&'static str)>,
) -> Result<Self, JsValue> {
if data_channel_ordered(&channel) != Some(false) {
return Err(JsValue::from_str(
"Iroh WebRTC carrier requires ordered=false",
));
}
if channel.max_retransmits() != Some(0) {
return Err(JsValue::from_str(
"Iroh WebRTC carrier requires maxRetransmits=0",
));
}
channel.set_binary_type(RtcDataChannelType::Arraybuffer);
channel.set_buffered_amount_low_threshold(BUFFERED_AMOUNT_LOW_THRESHOLD);
let session = Rc::new(session);
let reassembler = Rc::new(RefCell::new(CarrierReassembler::default()));
let inbound_session = session.clone();
let inbound_reassembler = reassembler.clone();
let inbound_channel = channel.clone();
let terminal_acknowledged = Rc::new(Cell::new(false));
let inbound_terminal_acknowledged = terminal_acknowledged.clone();
let terminal_notified = Rc::new(Cell::new(false));
let inbound_terminal_notified = terminal_notified.clone();
let inbound_terminal = on_terminal.clone();
let on_message = Closure::<dyn FnMut(MessageEvent)>::new(move |event: MessageEvent| {
let Ok(buffer) = event.data().dyn_into::<ArrayBuffer>() else {
return;
};
let bytes = Uint8Array::new(&buffer).to_vec();
let Ok(frame) = CarrierFrame::decode(&bytes, expected) else {
return;
};
match frame.terminal_control_kind(expected, &application_key) {
Ok(Some(CarrierControl::SessionTokenRevokedAck)) => {
inbound_terminal_acknowledged.set(true);
return;
}
Ok(Some(control @ CarrierControl::SessionTokenRevoked)) => {
if let Ok(ack) = CarrierFrame::terminal_control(
expected,
&application_key,
CarrierControl::SessionTokenRevokedAck,
)
.encode()
{
let _ = inbound_channel.send_with_u8_array(&ack);
}
if !inbound_terminal_notified.replace(true) {
inbound_session.close();
if let Some(reason) = control.lifecycle_reason() {
inbound_terminal(reason);
}
}
return;
}
Err(_) if frame.header.packet_id == CARRIER_CONTROL_PACKET_ID => return,
_ => {}
}
let now_ms = js_sys::Date::now().max(0.0) as u64;
let Ok(Some(packet)) = inbound_reassembler.borrow_mut().push(frame, now_ms) else {
return;
};
let deliver_session = inbound_session.clone();
spawn_local(async move {
let _ = deliver_session.deliver_inbound(packet).await;
});
});
channel.set_onmessage(Some(on_message.as_ref().unchecked_ref()));
let (buffered_wake_tx, buffered_wake_rx) = async_channel::bounded::<()>(1);
let wake_sender = buffered_wake_tx.clone();
let on_buffered_low = Closure::<dyn FnMut(Event)>::new(move |_: Event| {
let _ = wake_sender.try_send(());
});
channel.set_onbufferedamountlow(Some(on_buffered_low.as_ref().unchecked_ref()));
let close_sender = buffered_wake_tx;
let close_session = session.clone();
let close_terminal_notified = terminal_notified;
let on_close = Closure::<dyn FnMut(Event)>::new(move |_: Event| {
close_session.close();
close_sender.close();
if !close_terminal_notified.replace(true) {
on_terminal("iroh-carrier-ended");
}
});
channel.set_onclose(Some(on_close.as_ref().unchecked_ref()));
let outbound_channel = channel.clone();
let outbound_session = session.clone();
let next_packet_id = Rc::new(Cell::new(0_u64));
spawn_local(async move {
while let Ok(packet) = outbound_session.recv_outbound().await {
while outbound_channel.buffered_amount() > BUFFERED_AMOUNT_HARD_LIMIT {
if buffered_wake_rx.recv().await.is_err() {
return;
}
}
let packet_id = next_packet_id.get().wrapping_add(1);
let packet_id = if packet_id == CARRIER_CONTROL_PACKET_ID {
0
} else {
packet_id
};
next_packet_id.set(packet_id);
let Ok(frames) =
segment_packet(&packet, expected, packet_id, DATA_CHANNEL_MESSAGE_CEILING)
else {
return;
};
for frame in frames {
let Ok(encoded) = frame.encode() else {
return;
};
if outbound_channel.send_with_u8_array(&encoded).is_err() {
return;
}
}
}
});
Ok(Self {
channel,
_session: session,
_on_message: on_message,
_on_buffered_low: on_buffered_low,
_on_close: on_close,
expected,
application_key,
terminal_acknowledged,
})
}
pub async fn send_terminal(&self, reason: &str) -> Result<(), JsValue> {
let control = match reason {
crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED => {
CarrierControl::SessionTokenRevoked
}
_ => return Ok(()),
};
let encoded = CarrierFrame::terminal_control(self.expected, &self.application_key, control)
.encode()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
self.terminal_acknowledged.set(false);
for _ in 0..12 {
self.channel.send_with_u8_array(&encoded)?;
gloo_timers::future::sleep(std::time::Duration::from_millis(25)).await;
if self.terminal_acknowledged.get() {
break;
}
}
Ok(())
}
}
fn data_channel_ordered(channel: &RtcDataChannel) -> Option<bool> {
js_sys::Reflect::get(channel.as_ref(), &JsValue::from_str("ordered"))
.ok()
.and_then(|value| value.as_bool())
}
impl Drop for WasmWebRtcCarrierSession {
fn drop(&mut self) {
self._session.close();
self.channel.set_onmessage(None);
self.channel.set_onbufferedamountlow(None);
self.channel.set_onclose(None);
self.channel.close();
}
}