444 lines
16 KiB
Rust
444 lines
16 KiB
Rust
use crate::whip;
|
|
use anyhow::{anyhow, Result};
|
|
use bytes::Bytes;
|
|
use log::{debug, error, info, trace};
|
|
use reqwest::Url;
|
|
use std::{
|
|
boxed::Box,
|
|
sync::{Arc, Mutex, RwLock},
|
|
thread::JoinHandle,
|
|
time::{Duration, Instant},
|
|
};
|
|
use tokio::{
|
|
select,
|
|
sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
|
|
time::interval,
|
|
};
|
|
use webrtc::{
|
|
api::{
|
|
interceptor_registry::register_default_interceptors,
|
|
media_engine::{MediaEngine, MIME_TYPE_H264, MIME_TYPE_OPUS},
|
|
setting_engine::SettingEngine,
|
|
APIBuilder,
|
|
},
|
|
ice_transport::ice_connection_state::RTCIceConnectionState,
|
|
interceptor::registry::Registry,
|
|
media::Sample,
|
|
peer_connection::{
|
|
configuration::RTCConfiguration, peer_connection_state::RTCPeerConnectionState,
|
|
RTCPeerConnection,
|
|
},
|
|
rtp_transceiver::{
|
|
rtp_codec::{RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType},
|
|
rtp_transceiver_direction::RTCRtpTransceiverDirection,
|
|
RTCPFeedback, RTCRtpTransceiverInit,
|
|
},
|
|
stats::StatsReportType,
|
|
track::track_local::track_local_static_sample::TrackLocalStaticSample,
|
|
};
|
|
|
|
pub type ErrorCallback = Box<dyn FnMut(OutputStreamError) + Send + Sync>;
|
|
#[derive(Debug)]
|
|
pub struct EncodedPacket {
|
|
pub data: Vec<u8>,
|
|
pub duration: Duration,
|
|
pub typ: EncodedPacketType,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum EncodedPacketType {
|
|
Audio,
|
|
Video,
|
|
}
|
|
|
|
/// Errors for [`OutputStream`] worker thread
|
|
#[derive(Debug)]
|
|
pub enum OutputStreamError {
|
|
ConnectFailed,
|
|
NetworkError,
|
|
WriteError(webrtc::Error),
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum Message {
|
|
Packet(EncodedPacket),
|
|
Error(OutputStreamError),
|
|
Close,
|
|
}
|
|
|
|
pub enum WorkerResult {
|
|
Error(OutputStreamError),
|
|
Close,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct OutputStreamStats {
|
|
bytes_sent: u64,
|
|
congestion: f64,
|
|
dropped_frames: i32,
|
|
connect_time: Option<Instant>,
|
|
connect_duration: Duration,
|
|
}
|
|
|
|
pub struct OutputStream {
|
|
video_track: Arc<TrackLocalStaticSample>,
|
|
audio_track: Arc<TrackLocalStaticSample>,
|
|
peer_connection: Arc<RTCPeerConnection>,
|
|
stats: Arc<RwLock<OutputStreamStats>>,
|
|
worker_tx: UnboundedSender<Message>,
|
|
worker_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
|
|
whip_resource: Arc<Mutex<Option<Url>>>,
|
|
error_callback: Arc<Mutex<Option<ErrorCallback>>>,
|
|
}
|
|
|
|
impl OutputStream {
|
|
fn start_worker(&mut self, mut worker_rx: UnboundedReceiver<Message>) {
|
|
let worker_handle = std::thread::spawn({
|
|
let audio_track = self.audio_track.clone();
|
|
let video_track = self.video_track.clone();
|
|
let peer_connection = self.peer_connection.clone();
|
|
let stats = self.stats.clone();
|
|
let error_callback = self.error_callback.clone();
|
|
move || {
|
|
let runtime = tokio::runtime::Builder::new_multi_thread()
|
|
.enable_all()
|
|
.thread_name("webrtc_worker")
|
|
.worker_threads(4)
|
|
.build()
|
|
.unwrap();
|
|
let runtime = runtime;
|
|
let result = runtime.block_on(
|
|
async move {
|
|
let mut interval = interval(Duration::from_millis(500));
|
|
'worker_loop: loop {
|
|
let result = select! {
|
|
message = worker_rx.recv() => {
|
|
match message {
|
|
Some(Message::Packet(packet)) => {
|
|
let sample = Sample {
|
|
data: Bytes::from(packet.data),
|
|
duration: packet.duration,
|
|
..Default::default()
|
|
};
|
|
match (packet.typ, peer_connection.connection_state()) {
|
|
(_, RTCPeerConnectionState::Failed | RTCPeerConnectionState::Closed) => Err(WorkerResult::Error(OutputStreamError::NetworkError)),
|
|
(EncodedPacketType::Audio, _) => audio_track.write_sample(&sample).await.map_err(|e| WorkerResult::Error(OutputStreamError::WriteError(e))),
|
|
(EncodedPacketType::Video, _) => video_track.write_sample(&sample).await.map_err(|e| WorkerResult::Error(OutputStreamError::WriteError(e)))
|
|
}
|
|
},
|
|
Some(Message::Error(e)) => Err(WorkerResult::Error(e)),
|
|
Some(Message::Close) | None => Err(WorkerResult::Close),
|
|
}
|
|
}
|
|
_ = interval.tick() => {
|
|
let pc_stats = peer_connection.get_stats().await;
|
|
|
|
let stats = &mut stats.write().unwrap();
|
|
if let Some(StatsReportType::Transport(transport_stats)) = pc_stats.reports.get("ice_transport") {
|
|
stats.bytes_sent = transport_stats.bytes_sent as u64;
|
|
}
|
|
Ok(())
|
|
}
|
|
};
|
|
|
|
if let Err(e) = result {
|
|
break 'worker_loop e;
|
|
}
|
|
}
|
|
}
|
|
);
|
|
|
|
match result {
|
|
WorkerResult::Close => {}
|
|
WorkerResult::Error(e) => {
|
|
if let Some(callback) = &mut *error_callback.lock().unwrap() {
|
|
error!("Worker encountered error: {e:?}");
|
|
callback(e);
|
|
}
|
|
}
|
|
}
|
|
|
|
info!("Exiting worker thread");
|
|
}
|
|
});
|
|
|
|
*self.worker_handle.lock().unwrap() = Some(worker_handle);
|
|
}
|
|
|
|
pub async fn new() -> Result<Self> {
|
|
let video_track = Arc::new(TrackLocalStaticSample::new(
|
|
RTCRtpCodecCapability {
|
|
mime_type: MIME_TYPE_H264.to_owned(),
|
|
clock_rate: 90000,
|
|
sdp_fmtp_line: "profile-level-id=42e01f; max-fs=3600; max-mbps=108000; max-br=1400"
|
|
.to_string(),
|
|
..Default::default()
|
|
},
|
|
"video".to_owned(),
|
|
"webrtc-rs".to_owned(),
|
|
));
|
|
|
|
let audio_track = Arc::new(TrackLocalStaticSample::new(
|
|
RTCRtpCodecCapability {
|
|
mime_type: MIME_TYPE_OPUS.to_owned(),
|
|
..Default::default()
|
|
},
|
|
"audio".to_owned(),
|
|
"webrtc-rs".to_owned(),
|
|
));
|
|
|
|
let mut m = MediaEngine::default();
|
|
m.register_codec(
|
|
RTCRtpCodecParameters {
|
|
capability: RTCRtpCodecCapability {
|
|
mime_type: MIME_TYPE_OPUS.to_owned(),
|
|
clock_rate: 48000,
|
|
channels: 2,
|
|
sdp_fmtp_line: "minptime=10;useinbandfec=1".to_owned(),
|
|
rtcp_feedback: vec![],
|
|
},
|
|
payload_type: 111,
|
|
..Default::default()
|
|
},
|
|
RTPCodecType::Audio,
|
|
)?;
|
|
|
|
m.register_codec(
|
|
RTCRtpCodecParameters {
|
|
capability: RTCRtpCodecCapability {
|
|
mime_type: MIME_TYPE_H264.to_owned(),
|
|
clock_rate: 90000,
|
|
channels: 0,
|
|
sdp_fmtp_line:
|
|
"level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f"
|
|
.to_owned(),
|
|
rtcp_feedback: vec![
|
|
RTCPFeedback {
|
|
typ: "goog-remb".to_owned(),
|
|
parameter: "".to_owned(),
|
|
},
|
|
RTCPFeedback {
|
|
typ: "ccm".to_owned(),
|
|
parameter: "fir".to_owned(),
|
|
},
|
|
RTCPFeedback {
|
|
typ: "nack".to_owned(),
|
|
parameter: "".to_owned(),
|
|
},
|
|
RTCPFeedback {
|
|
typ: "nack".to_owned(),
|
|
parameter: "pli".to_owned(),
|
|
},
|
|
],
|
|
},
|
|
payload_type: 125,
|
|
..Default::default()
|
|
},
|
|
RTPCodecType::Video,
|
|
)?;
|
|
|
|
let mut registry = Registry::new();
|
|
registry = register_default_interceptors(registry, &mut m)?;
|
|
|
|
let gather_ips: Vec<_> = if_addrs::get_if_addrs()
|
|
.unwrap_or_else(|_| Vec::new())
|
|
.into_iter()
|
|
.filter(|addr| {
|
|
if !addr.is_loopback() {
|
|
trace!(
|
|
"Valid gather address for interface {}: {:#?}",
|
|
addr.name,
|
|
addr.ip()
|
|
);
|
|
true
|
|
} else {
|
|
trace!(
|
|
"Loopback address ignored for interface {}: {:#?}",
|
|
addr.name,
|
|
addr.ip()
|
|
);
|
|
false
|
|
}
|
|
})
|
|
.map(|addr| addr.ip())
|
|
.collect();
|
|
|
|
let mut setting_engine = SettingEngine::default();
|
|
|
|
// Only attempt to limit if we found valid interfaces, otherwise do not attempt
|
|
// to limit as we may be on a platform that doesn't expose enough information
|
|
if !gather_ips.is_empty() {
|
|
debug!("Gather addresses: {gather_ips:#?}");
|
|
setting_engine.set_ip_filter(Box::new(move |ip: std::net::IpAddr| -> bool {
|
|
gather_ips.contains(&ip)
|
|
}));
|
|
}
|
|
|
|
let api = APIBuilder::new()
|
|
.with_media_engine(m)
|
|
.with_interceptor_registry(registry)
|
|
.with_setting_engine(setting_engine)
|
|
.build();
|
|
|
|
// Prepare the configuration
|
|
let config = RTCConfiguration {
|
|
..Default::default()
|
|
};
|
|
|
|
let peer_connection = Arc::new(api.new_peer_connection(config).await?);
|
|
let stats: Arc<RwLock<OutputStreamStats>> = Arc::new(RwLock::new(OutputStreamStats {
|
|
connect_time: None,
|
|
..Default::default()
|
|
}));
|
|
|
|
let (worker_tx, worker_rx): (UnboundedSender<Message>, UnboundedReceiver<Message>) =
|
|
mpsc::unbounded_channel();
|
|
let error_callback: Arc<Mutex<Option<ErrorCallback>>> = Arc::new(Mutex::new(None));
|
|
|
|
let mut output_stream = Self {
|
|
audio_track,
|
|
video_track,
|
|
peer_connection,
|
|
stats,
|
|
worker_tx,
|
|
worker_handle: Arc::new(Mutex::new(None)),
|
|
whip_resource: Arc::new(Mutex::new(None)),
|
|
error_callback,
|
|
};
|
|
|
|
output_stream.start_worker(worker_rx);
|
|
|
|
Ok(output_stream)
|
|
}
|
|
|
|
pub async fn connect(&self, url: &str, bearer_token: Option<&str>) {
|
|
self.connect_internal(url, bearer_token)
|
|
.await
|
|
.unwrap_or_else(|e| {
|
|
error!("Failed connecting to: {e}");
|
|
let _ = self
|
|
.worker_tx
|
|
.send(Message::Error(OutputStreamError::ConnectFailed));
|
|
})
|
|
}
|
|
|
|
async fn connect_internal(&self, url: &str, bearer_token: Option<&str>) -> Result<()> {
|
|
println!("Setting up webrtc!");
|
|
|
|
self.peer_connection
|
|
.add_transceiver_from_track(
|
|
self.video_track.clone(),
|
|
&[RTCRtpTransceiverInit {
|
|
direction: RTCRtpTransceiverDirection::Sendonly,
|
|
send_encodings: Vec::new(),
|
|
}],
|
|
)
|
|
.await?;
|
|
|
|
self.peer_connection
|
|
.add_transceiver_from_track(
|
|
self.audio_track.clone(),
|
|
&[RTCRtpTransceiverInit {
|
|
direction: RTCRtpTransceiverDirection::Sendonly,
|
|
send_encodings: Vec::new(),
|
|
}],
|
|
)
|
|
.await?;
|
|
|
|
self.peer_connection
|
|
.on_ice_connection_state_change(Box::new(
|
|
move |connection_state: RTCIceConnectionState| {
|
|
info!("Connection State has changed {}", connection_state);
|
|
if connection_state == RTCIceConnectionState::Connected {
|
|
// on_success();
|
|
}
|
|
Box::pin(async {})
|
|
},
|
|
));
|
|
|
|
self.peer_connection
|
|
.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
|
|
debug!("Peer Connection State has changed: {}", s);
|
|
|
|
if s == RTCPeerConnectionState::Failed {
|
|
error!("Peer connection state went to failed")
|
|
}
|
|
|
|
Box::pin(async {})
|
|
}));
|
|
|
|
let offer = self.peer_connection.create_offer(None).await?;
|
|
let mut gather_complete = self.peer_connection.gathering_complete_promise().await;
|
|
self.peer_connection.set_local_description(offer).await?;
|
|
|
|
// Block until gathering complete
|
|
let _ = gather_complete.recv().await;
|
|
|
|
let offer = self
|
|
.peer_connection
|
|
.local_description()
|
|
.await
|
|
.ok_or_else(|| anyhow!("No local description available"))?;
|
|
let (answer, whip_resource) = whip::offer(url, bearer_token, offer).await?;
|
|
self.peer_connection.set_remote_description(answer).await?;
|
|
|
|
*self.whip_resource.lock().unwrap() = Some(whip_resource);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn close(&self) -> Result<()> {
|
|
let close_result = self.worker_tx.send(Message::Close);
|
|
|
|
// Take worker handle so it's dropped
|
|
let worker_future = self.worker_handle.lock().unwrap().take();
|
|
// If close was a success (worker could receive messages), join on thread
|
|
// Otherwise, worker thread is already dead or currently closing due to error callback
|
|
if close_result.is_ok() {
|
|
if let Some(worker_future) = worker_future {
|
|
worker_future
|
|
.join()
|
|
.unwrap_or_else(|e| error!("Failed joining worker thread: {e:?}"));
|
|
}
|
|
}
|
|
|
|
let whip_resource = self.whip_resource.lock().unwrap().take();
|
|
if let Some(whip_resource) = whip_resource {
|
|
whip::delete(&whip_resource).await?;
|
|
}
|
|
Ok(self.peer_connection.close().await?)
|
|
}
|
|
|
|
pub fn bytes_sent(&self) -> u64 {
|
|
self.stats.read().unwrap().bytes_sent
|
|
}
|
|
|
|
pub fn congestion(&self) -> f64 {
|
|
self.stats.read().unwrap().congestion
|
|
}
|
|
|
|
pub fn connect_time(&self) -> Duration {
|
|
let mut stats = self.stats.write().unwrap();
|
|
|
|
if let Some(connect_time) = stats.connect_time {
|
|
stats.connect_duration = Instant::now() - connect_time;
|
|
}
|
|
|
|
stats.connect_duration
|
|
}
|
|
|
|
pub fn dropped_frames(&self) -> i32 {
|
|
self.stats.read().unwrap().dropped_frames
|
|
}
|
|
|
|
pub fn write(&self, packet: EncodedPacket) -> Result<()> {
|
|
self.worker_tx
|
|
.send(Message::Packet(packet))
|
|
.map_err(|e| e.into())
|
|
}
|
|
|
|
pub fn set_error_callback(&self, callback: ErrorCallback) {
|
|
*self.error_callback.lock().unwrap() = Some(callback);
|
|
}
|
|
}
|