use std::sync::Arc;
use std::time::Duration;
use anyhow::{anyhow, bail, ensure, Context, Result};
use bytes::Bytes;
use serde_json::{json, Value};
use url::Url;
use webrtc::api::APIBuilder;
use webrtc::data_channel::data_channel_init::RTCDataChannelInit;
use webrtc::data_channel::data_channel_state::RTCDataChannelState;
use webrtc::data_channel::RTCDataChannel;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
use webrtc::peer_connection::RTCPeerConnection;
use super::{BroadcastAdapterAction, BroadcastAdapterObservation, BroadcastRole, BroadcastSession};
const MAGIC: u32 = 0x4f525443;
const FRAGMENT_BYTES: usize = 8_000;
const MAX_OBJECT_BYTES: usize = 1_000_000;
pub struct SfuControlAccess {
pub control_url: Url,
pub origin: String,
pub grant: String,
pub binding_public_jwk: Value,
pub publication_verifying_key: String,
}
pub struct NativeSfuPublisher {
session: BroadcastSession,
access: SfuControlAccess,
http: reqwest::Client,
session_id: String,
peer: Arc<RTCPeerConnection>,
channel: Arc<RTCDataChannel>,
broadcast_generation: u64,
grant_generation: u64,
next_object_id: std::sync::atomic::AtomicU32,
}
impl NativeSfuPublisher {
pub async fn connect(session: BroadcastSession, access: SfuControlAccess) -> Result<Self> {
ensure!(
matches!(
session.role(),
BroadcastRole::Owner | BroadcastRole::Publisher | BroadcastRole::Processor
),
"broadcast grant cannot publish"
);
ensure!(
session.source_slots() == ["main"],
"native SFU demo requires one source slot"
);
ensure!(
matches!(access.control_url.scheme(), "http" | "https"),
"SFU control URL must be HTTP(S)"
);
let Some(BroadcastAdapterAction::Open {
broadcast_generation,
grant_generation,
..
}) = session.take_actions(1).into_iter().next()
else {
bail!("broadcast session has no opening action");
};
let result = Self::connect_open(
session.clone(),
access,
broadcast_generation,
grant_generation,
)
.await;
if result.is_err() {
let _ = session.observe(BroadcastAdapterObservation::Failed {
broadcast_generation,
grant_generation,
retryable: false,
code: "sfu-open-failed".into(),
});
}
result
}
async fn connect_open(
session: BroadcastSession,
access: SfuControlAccess,
broadcast_generation: u64,
grant_generation: u64,
) -> Result<Self> {
let http = reqwest::Client::new();
let opened = Self::call(&http, &access, "session", json!({})).await?;
let session_id = opened["sessionId"]
.as_str()
.context("SFU session ID missing")?
.to_string();
let result: Result<(Arc<RTCPeerConnection>, Arc<RTCDataChannel>)> = async {
let peer = Arc::new(APIBuilder::new().build().new_peer_connection(RTCConfiguration::default()).await?);
let offer = Self::call(&http, &access, "establish", json!({ "sessionId": session_id })).await?;
let offer_sdp = offer["sessionDescription"]["sdp"].as_str().context("SFU offer SDP missing")?;
peer.set_remote_description(RTCSessionDescription::offer(offer_sdp.to_string())?).await?;
let answer = peer.create_answer(None).await?;
let mut gathering = peer.gathering_complete_promise().await;
peer.set_local_description(answer).await?;
tokio::time::timeout(Duration::from_secs(15), gathering.recv()).await.context("SFU ICE gathering timed out")?;
let local = peer.local_description().await.context("SFU answer SDP missing")?;
Self::call(&http, &access, "answer", json!({ "sessionId": session_id, "sessionDescription": { "type": "answer", "sdp": local.sdp } })).await?;
tokio::time::timeout(Duration::from_secs(15), async {
loop {
match peer.connection_state() {
RTCPeerConnectionState::Connected => break Ok(()),
RTCPeerConnectionState::Failed | RTCPeerConnectionState::Closed => break Err(anyhow!("SFU connection failed")),
_ => tokio::time::sleep(Duration::from_millis(20)).await,
}
}
}).await.context("SFU connection timed out")??;
let allocated = Self::call(&http, &access, "channel", json!({ "sessionId": session_id, "sourceSlot": "main" })).await?;
let channel_id = u16::try_from(allocated["channelId"].as_u64().context("SFU channel ID missing")?).context("SFU channel ID out of range")?;
let channel = peer.create_data_channel("openrtc-main", Some(RTCDataChannelInit {
ordered: Some(true), negotiated: Some(channel_id), ..Default::default()
})).await?;
tokio::time::timeout(Duration::from_secs(15), async {
loop {
match channel.ready_state() {
RTCDataChannelState::Open => break Ok(()),
RTCDataChannelState::Closing | RTCDataChannelState::Closed => break Err(anyhow!("SFU channel closed")),
_ => tokio::time::sleep(Duration::from_millis(20)).await,
}
}
}).await.context("SFU channel timed out")??;
Ok((peer, channel))
}.await;
let (peer, channel) = match result {
Ok(parts) => parts,
Err(error) => {
let _ =
Self::call(&http, &access, "close", json!({ "sessionId": session_id })).await;
return Err(error);
}
};
let mechanism = Self {
session,
access,
http,
session_id,
peer,
channel,
broadcast_generation,
grant_generation,
next_object_id: std::sync::atomic::AtomicU32::new(0),
};
mechanism
.session
.observe(BroadcastAdapterObservation::Opened {
broadcast_generation,
grant_generation,
})?;
Ok(mechanism)
}
async fn call(
http: &reqwest::Client,
access: &SfuControlAccess,
path: &str,
input: Value,
) -> Result<Value> {
let url = Url::parse(&format!(
"{}/{}",
access.control_url.as_str().trim_end_matches('/'),
path
))?;
let mut body = json!({
"grant": access.grant,
"bindingPublicJwk": access.binding_public_jwk,
"publicationVerifyingKey": access.publication_verifying_key,
});
if let Some(fields) = input.as_object() {
for (key, value) in fields {
body[key] = value.clone();
}
}
let response = http
.post(url)
.header("Origin", &access.origin)
.json(&body)
.send()
.await?;
let status = response.status();
let value: Value = response.json().await?;
ensure!(
status.is_success(),
"SFU {path} rejected: {}",
value["error"].as_str().unwrap_or("unknown error")
);
Ok(value)
}
pub async fn flush(&self) -> Result<()> {
for action in self.session.take_actions(super::MAX_BROADCAST_ACTIONS) {
match action {
BroadcastAdapterAction::Publish {
broadcast_generation,
grant_generation,
source_slot,
object,
..
} if broadcast_generation == self.broadcast_generation
&& grant_generation == self.grant_generation
&& source_slot == "main" =>
{
if let Err(error) = self.publish(&object).await {
let _ = self.session.observe(BroadcastAdapterObservation::Failed {
broadcast_generation,
grant_generation,
retryable: false,
code: "sfu-publish-failed".into(),
});
return Err(error);
}
}
BroadcastAdapterAction::Close {
broadcast_generation,
grant_generation,
..
} if broadcast_generation == self.broadcast_generation
&& grant_generation == self.grant_generation =>
{
return Ok(())
}
_ => bail!("broadcast adapter received an unexpected action"),
}
}
Ok(())
}
async fn publish(&self, object: &[u8]) -> Result<()> {
ensure!(
!object.is_empty() && object.len() <= MAX_OBJECT_BYTES,
"SFU object exceeds bound"
);
let count = object.len().div_ceil(FRAGMENT_BYTES);
let id = self
.next_object_id
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
for (index, part) in object.chunks(FRAGMENT_BYTES).enumerate() {
let mut framed = Vec::with_capacity(12 + part.len());
framed.extend_from_slice(&MAGIC.to_be_bytes());
framed.extend_from_slice(&id.to_be_bytes());
framed.extend_from_slice(&(index as u16).to_be_bytes());
framed.extend_from_slice(&(count as u16).to_be_bytes());
framed.extend_from_slice(part);
tokio::time::timeout(Duration::from_secs(10), async {
while self.channel.buffered_amount().await > 128_000 {
ensure!(
self.channel.ready_state() == RTCDataChannelState::Open,
"SFU channel closed"
);
tokio::time::sleep(Duration::from_millis(10)).await;
}
self.channel.send(&Bytes::from(framed)).await?;
Ok::<(), anyhow::Error>(())
})
.await
.context("SFU send stalled")??;
}
Ok(())
}
pub async fn close(self) {
let _ = self.channel.close().await;
let _ = self.peer.close().await;
let _ = Self::call(
&self.http,
&self.access,
"close",
json!({ "sessionId": self.session_id }),
)
.await;
self.session.close();
}
}