use std::{
collections::{HashMap, VecDeque},
io::Write,
net::{Ipv4Addr, SocketAddrV4},
path::Path,
process::Stdio,
sync::atomic::{AtomicBool, Ordering},
};
use rand::random;
use tempfile::NamedTempFile;
use tokio::{
io::{AsyncBufReadExt, AsyncRead, BufReader},
process::Command,
};
use uuid::Uuid;
use super::media::{
GatewayChildTasks, annex_b_nals, build_rtcp_sender_report, current_unix_time,
packetize_h264_payloads,
};
use super::runtime::MAX_VOICE_RECONNECT_ATTEMPTS;
use super::*;
const STREAM_RTP_PACKET_BYTES: usize = 4096;
const LOCAL_H264_MAX_PAYLOAD_BYTES: usize = 1200;
const STREAM_STARTUP_BUFFER_MAX_FRAMES: usize = 180;
const STREAM_STARTUP_BUFFER_MAX_BYTES: usize = 32 * 1024 * 1024;
const STREAM_H264_ACCESS_UNIT_MAX_BYTES: usize = 8 * 1024 * 1024;
const STREAM_H264_ACCESS_UNIT_MAX_PACKETS: usize = 4096;
const STREAM_STARTUP_REPLAY_FRAME_TICKS: u32 = 90;
const STREAM_PLAYER_READY_TIMEOUT: Duration = Duration::from_secs(10);
const OPUS_RTP_CLOCK_RATE: u32 = 48_000;
const VIDEO_RTP_CLOCK_RATE: u32 = 90_000;
const STREAM_AUDIO_CLOCK_DRIFT_TOLERANCE_TICKS: u32 = DISCORD_OPUS_TIMESTAMP_INCREMENT * 6;
const STREAM_KEYFRAME_REQUEST_INTERVAL: Duration = Duration::from_secs(1);
const STREAM_VIDEO_NACK_INTERVAL: Duration = Duration::from_millis(100);
const STREAM_VIDEO_GAP_TIMEOUT: Duration = Duration::from_millis(500);
const STREAM_VIDEO_REORDER_WINDOW: u16 = 128;
const STREAM_VIDEO_MAX_NACK_SEQUENCES: usize = 64;
const LOCAL_RTCP_REPORT_INTERVAL: Duration = Duration::from_secs(1);
const STREAM_CONNECTION_STABLE_INTERVAL: Duration = Duration::from_secs(10);
const STREAM_WATCH_RECONNECT_BASE_DELAY: Duration = Duration::from_millis(250);
const STREAM_WATCH_RECONNECT_MAX_DELAY: Duration = Duration::from_secs(2);
const RTCP_TRANSPORT_LAYER_FEEDBACK: u8 = 205;
const RTCP_GENERIC_NACK_FORMAT: u8 = 1;
const RTCP_PAYLOAD_SPECIFIC_FEEDBACK: u8 = 206;
const RTCP_PLI_FORMAT: u8 = 1;
const RTCP_PLI_LENGTH_WORDS_MINUS_ONE: u16 = 2;
#[derive(Clone, Eq, PartialEq)]
pub(super) struct StreamGatewaySession {
pub(super) connection_id: u64,
pub(super) request: StreamWatchRequest,
pub(super) current_user_id: Id<UserMarker>,
pub(super) session_id: String,
pub(super) rtc_server_id: String,
pub(super) rtc_channel_id: Id<ChannelMarker>,
pub(super) endpoint: String,
pub(super) token: String,
reconnect_delay: Duration,
}
impl std::fmt::Debug for StreamGatewaySession {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("StreamGatewaySession")
.field("connection_id", &self.connection_id)
.field("request", &self.request)
.field("current_user_id", &self.current_user_id)
.field("session_id", &self.session_id)
.field("rtc_server_id", &self.rtc_server_id)
.field("rtc_channel_id", &self.rtc_channel_id)
.field("endpoint", &self.endpoint)
.field("token", &"<redacted>")
.field("reconnect_delay", &self.reconnect_delay)
.finish()
}
}
struct StreamPlayerLogTasks {
tasks: Vec<JoinHandle<()>>,
}
impl StreamPlayerLogTasks {
fn new(tasks: [JoinHandle<()>; 2]) -> Self {
Self {
tasks: Vec::from(tasks),
}
}
async fn finish(mut self) {
for task in self.tasks.drain(..) {
let _ = task.await;
}
}
}
impl Drop for StreamPlayerLogTasks {
fn drop(&mut self) {
for task in &self.tasks {
task.abort();
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct ObservedStreamVoiceState {
scope: VoiceScope,
channel_id: Id<ChannelMarker>,
session_id: String,
}
#[derive(Clone)]
struct StreamPlayerReadySignal {
player_ready: Arc<AtomicBool>,
ready_tx: mpsc::UnboundedSender<u64>,
media_generation: u64,
display_name: String,
}
#[derive(Debug, Eq, PartialEq)]
struct StreamConnectionFailure {
message: String,
outcome: VoiceConnectionEnd,
}
impl StreamConnectionFailure {
fn reconnect(message: impl Into<String>) -> Self {
Self {
message: message.into(),
outcome: VoiceConnectionEnd::Reconnect,
}
}
fn stop(message: impl Into<String>) -> Self {
Self {
message: message.into(),
outcome: VoiceConnectionEnd::Stop,
}
}
}
impl From<String> for StreamConnectionFailure {
fn from(message: String) -> Self {
Self::reconnect(message)
}
}
#[derive(Default)]
pub(super) struct StreamRuntimeState {
current_user_id: Option<Id<UserMarker>>,
current_voice: Option<ObservedStreamVoiceState>,
requested: Option<StreamWatchRequest>,
create: Option<StreamCreateInfo>,
server: Option<StreamServerInfo>,
active: Option<StreamGatewaySession>,
reconnect_attempts: u8,
next_connection_id: u64,
}
#[derive(Default)]
pub(super) struct StreamRuntimeUpdate {
pub(super) close_stream_key: Option<String>,
pub(super) send_delete: bool,
pub(super) connect: Option<StreamGatewaySession>,
pub(super) playback_ended: Option<StreamPlaybackEnded>,
pub(super) error: Option<String>,
}
pub(super) struct StreamPlaybackEnded {
pub(super) request: StreamWatchRequest,
pub(super) reconnecting: bool,
}
impl StreamRuntimeState {
pub(super) fn apply(&mut self, event: &VoiceRuntimeEvent) -> StreamRuntimeUpdate {
let mut update = StreamRuntimeUpdate::default();
match event {
VoiceRuntimeEvent::CurrentUserReady(user_id) => self.current_user_id = *user_id,
VoiceRuntimeEvent::VoiceState(state) => self.record_voice_state(state, &mut update),
VoiceRuntimeEvent::WatchStreamRequested(request) => {
if self
.requested
.as_ref()
.is_none_or(|current| current.stream_key != request.stream_key)
{
update.playback_ended =
self.requested.take().map(|request| StreamPlaybackEnded {
request,
reconnecting: false,
});
update.close_stream_key =
self.active.take().map(|active| active.request.stream_key);
update.send_delete = update.close_stream_key.is_some();
self.create = None;
self.server = None;
self.reconnect_attempts = 0;
}
self.requested = Some(request.clone());
}
VoiceRuntimeEvent::WatchStreamCancelled { stream_key } => {
self.clear_matching(stream_key, &mut update, false);
}
VoiceRuntimeEvent::StreamCreate(stream) => {
if self
.requested
.as_ref()
.is_some_and(|request| request.stream_key == stream.stream_key)
{
self.create = Some(stream.clone());
}
}
VoiceRuntimeEvent::StreamServer(server) => {
if self
.requested
.as_ref()
.is_some_and(|request| request.stream_key == server.stream_key)
{
if self.active.as_ref().is_some_and(|active| {
!server.matches_connection(&active.endpoint, &active.token)
}) {
update.playback_ended =
self.requested
.as_ref()
.cloned()
.map(|request| StreamPlaybackEnded {
request,
reconnecting: true,
});
update.close_stream_key =
self.active.take().map(|active| active.request.stream_key);
}
self.server = Some(server.clone());
}
}
VoiceRuntimeEvent::StreamDelete(stream) => {
if let Some(request) = self
.requested
.as_ref()
.filter(|request| request.stream_key == stream.stream_key)
&& (!stream.reason.is_empty() || stream.unavailable)
{
let reason = if stream.reason.is_empty() {
"stream unavailable"
} else {
stream.reason.as_str()
};
update.error = Some(format!(
"Could not watch {}'s stream: {reason}",
request.display_name
));
}
self.clear_matching(&stream.stream_key, &mut update, false);
}
VoiceRuntimeEvent::StreamConnectionEstablished {
connection_id,
stream_key,
} => {
if self.active.as_ref().is_some_and(|active| {
active.connection_id == *connection_id
&& active.request.stream_key == *stream_key
}) {
self.reconnect_attempts = 0;
}
}
VoiceRuntimeEvent::StreamConnectionEnded {
connection_id,
stream_key,
outcome,
} => {
if self.active.as_ref().is_some_and(|active| {
active.connection_id == *connection_id
&& active.request.stream_key == *stream_key
}) {
self.active = None;
if *outcome == VoiceConnectionEnd::Stop
|| self.reconnect_attempts >= MAX_VOICE_RECONNECT_ATTEMPTS
{
update.playback_ended =
self.requested.take().map(|request| StreamPlaybackEnded {
request,
reconnecting: false,
});
self.create = None;
self.server = None;
self.reconnect_attempts = 0;
update.close_stream_key = Some(stream_key.clone());
update.send_delete = true;
} else {
self.reconnect_attempts = self.reconnect_attempts.saturating_add(1);
update.playback_ended =
self.requested
.as_ref()
.cloned()
.map(|request| StreamPlaybackEnded {
request,
reconnecting: true,
});
}
}
}
VoiceRuntimeEvent::Shutdown => {
update.playback_ended = self.requested.take().map(|request| StreamPlaybackEnded {
request,
reconnecting: false,
});
update.close_stream_key =
self.active.take().map(|active| active.request.stream_key);
update.send_delete = update.close_stream_key.is_some();
self.create = None;
self.server = None;
}
_ => {}
}
if self.active.is_none() {
update.connect = self.connect_if_ready();
}
update
}
fn record_voice_state(&mut self, state: &VoiceStateInfo, update: &mut StreamRuntimeUpdate) {
if self.current_user_id != Some(state.user_id) {
return;
}
let Some(channel_id) = state.channel_id else {
self.current_voice = None;
update.playback_ended = self.requested.take().map(|request| StreamPlaybackEnded {
request,
reconnecting: false,
});
update.close_stream_key = self.active.take().map(|active| active.request.stream_key);
update.send_delete = update.close_stream_key.is_some();
self.create = None;
self.server = None;
return;
};
let Some(scope) = state.scope() else {
return;
};
let Some(session_id) = state
.session_id
.as_ref()
.filter(|session_id| !session_id.is_empty())
else {
return;
};
self.current_voice = Some(ObservedStreamVoiceState {
scope,
channel_id,
session_id: session_id.clone(),
});
if self
.requested
.as_ref()
.is_some_and(|request| request.scope != scope || request.channel_id != channel_id)
{
update.playback_ended = self.requested.take().map(|request| StreamPlaybackEnded {
request,
reconnecting: false,
});
update.close_stream_key = self.active.take().map(|active| active.request.stream_key);
update.send_delete = update.close_stream_key.is_some();
self.create = None;
self.server = None;
}
}
fn clear_matching(
&mut self,
stream_key: &str,
update: &mut StreamRuntimeUpdate,
send_delete: bool,
) {
if self
.requested
.as_ref()
.is_some_and(|request| request.stream_key == stream_key)
{
update.playback_ended = self.requested.take().map(|request| StreamPlaybackEnded {
request,
reconnecting: false,
});
self.create = None;
self.server = None;
self.reconnect_attempts = 0;
}
if self
.active
.as_ref()
.is_some_and(|active| active.request.stream_key == stream_key)
{
self.active = None;
update.close_stream_key = Some(stream_key.to_owned());
update.send_delete = send_delete;
}
}
fn connect_if_ready(&mut self) -> Option<StreamGatewaySession> {
let request = self.requested.as_ref()?;
let current_voice = self.current_voice.as_ref()?;
if request.scope != current_voice.scope || request.channel_id != current_voice.channel_id {
return None;
}
let create = self.create.as_ref()?;
let server = self.server.as_ref()?;
if create.stream_key != request.stream_key || server.stream_key != request.stream_key {
return None;
}
let endpoint = server.endpoint.as_ref()?.trim_end_matches('/').to_owned();
if endpoint.is_empty() || server.token.is_empty() {
return None;
}
self.next_connection_id = self.next_connection_id.wrapping_add(1).max(1);
let session = StreamGatewaySession {
connection_id: self.next_connection_id,
request: request.clone(),
current_user_id: self.current_user_id?,
session_id: current_voice.session_id.clone(),
rtc_server_id: create.rtc_server_id.clone(),
rtc_channel_id: create.rtc_channel_id,
endpoint,
token: server.token.clone(),
reconnect_delay: stream_watch_reconnect_delay(self.reconnect_attempts),
};
self.active = Some(session.clone());
Some(session)
}
}
fn stream_watch_reconnect_delay(reconnect_attempts: u8) -> Duration {
if reconnect_attempts == 0 {
return Duration::ZERO;
}
let multiplier = 1u32 << u32::from(reconnect_attempts.saturating_sub(1).min(3));
let base_delay = STREAM_WATCH_RECONNECT_BASE_DELAY
.saturating_mul(multiplier)
.min(STREAM_WATCH_RECONNECT_MAX_DELAY);
let jitter_limit_millis =
u64::try_from((base_delay / 4).as_millis()).expect("bounded retry jitter fits u64");
let jitter = Duration::from_millis(random::<u64>() % (jitter_limit_millis + 1));
base_delay
.saturating_add(jitter)
.min(STREAM_WATCH_RECONNECT_MAX_DELAY)
}
pub(super) async fn run_stream_gateway_session(
session: StreamGatewaySession,
events_tx: mpsc::UnboundedSender<VoiceRuntimeEvent>,
status_publisher: VoiceStatusPublisher,
) {
if !session.reconnect_delay.is_zero() {
logging::debug(
"stream",
format!(
"waiting {:?} before reconnecting stream watch",
session.reconnect_delay
),
);
sleep(session.reconnect_delay).await;
}
let outcome = match connect_stream_gateway(&session, &events_tx, &status_publisher).await {
Ok(outcome) => outcome,
Err(error) => {
logging::error("stream", &error.message);
status_publisher
.publish_error(format!(
"Could not watch {}'s stream: {}",
session.request.display_name, error.message
))
.await;
error.outcome
}
};
let _ = events_tx.send(VoiceRuntimeEvent::StreamConnectionEnded {
connection_id: session.connection_id,
stream_key: session.request.stream_key.clone(),
outcome,
});
}
async fn connect_stream_gateway(
session: &StreamGatewaySession,
events_tx: &mpsc::UnboundedSender<VoiceRuntimeEvent>,
status_publisher: &VoiceStatusPublisher,
) -> Result<VoiceConnectionEnd, StreamConnectionFailure> {
let url = gateway::voice_gateway_url(&session.endpoint)?;
logging::debug("stream", format!("connecting stream websocket: {url}"));
let (ws, response) = timeout(VOICE_WEBSOCKET_CONNECT_TIMEOUT, connect_async(&url))
.await
.map_err(|_| "stream websocket connect timed out after 10s".to_owned())?
.map_err(|error| format!("stream websocket connect failed: {error}"))?;
logging::debug(
"stream",
format!("stream websocket connected: status={}", response.status()),
);
let (writer, mut reader) = ws.split();
let writer = Arc::new(Mutex::new(writer));
let mut gateway_control = gateway::StreamVoiceGatewayControl::new(
Arc::clone(&writer),
session.current_user_id,
&session.rtc_server_id,
)?;
let dave_state = gateway_control.dave_state();
let mut child_tasks = GatewayChildTasks::default();
let (media_finished_tx, mut media_finished_rx) =
mpsc::unbounded_channel::<Result<(), StreamConnectionFailure>>();
let (player_ready_tx, mut player_ready_rx) = mpsc::unbounded_channel::<u64>();
let (video_source_tx, video_source_rx) = watch::channel(StreamVideoSource::default());
let mut udp_socket: Option<Arc<UdpSocket>> = None;
let mut local_ssrc: Option<u32> = None;
let mut current_description: Option<VoiceSessionDescription> = None;
let mut media_generation = 0u64;
let mut connection_stable_deadline: Option<Instant> = None;
gateway::send_voice_text(&writer, stream_identify_payload(session)).await?;
logging::debug("stream", "stream identify sent");
let result = loop {
let frame = tokio::select! {
_ = gateway_control.heartbeat_timed_out() => {
break Ok(VoiceConnectionEnd::Reconnect);
}
media_result = media_finished_rx.recv(), if child_tasks.has_media() => {
match media_result {
Some(Ok(())) => break Ok(VoiceConnectionEnd::Stop),
Some(Err(error)) => break Err(error),
None => break Ok(VoiceConnectionEnd::Reconnect),
}
}
ready_generation = player_ready_rx.recv(), if child_tasks.has_media() => {
if stream_player_ready_is_current(ready_generation, media_generation) {
status_publisher
.publish_stream_playback_ready(
session.request.scope,
session.request.channel_id,
session.request.owner_id,
)
.await;
connection_stable_deadline =
Some(Instant::now() + STREAM_CONNECTION_STABLE_INTERVAL);
}
continue;
}
_ = tokio::time::sleep_until(tokio::time::Instant::from_std(
connection_stable_deadline.unwrap_or_else(Instant::now)
)), if connection_stable_deadline.is_some() => {
connection_stable_deadline = None;
let _ = events_tx.send(VoiceRuntimeEvent::StreamConnectionEstablished {
connection_id: session.connection_id,
stream_key: session.request.stream_key.clone(),
});
continue;
}
frame = reader.next() => frame,
};
let Some(frame) = frame else {
break Ok(VoiceConnectionEnd::Reconnect);
};
let frame = match frame {
Ok(frame) => frame,
Err(error) => {
break Err(StreamConnectionFailure::reconnect(format!(
"stream websocket read failed: {error}"
)));
}
};
match gateway_control.frame_action(&frame).await? {
gateway::StreamVoiceGatewayFrameAction::Payload => {}
gateway::StreamVoiceGatewayFrameAction::Continue => continue,
gateway::StreamVoiceGatewayFrameAction::End(outcome) => break Ok(outcome),
}
match frame {
WsMessage::Text(text) => {
let value: Value = serde_json::from_str(&text)
.map_err(|error| format!("stream websocket JSON parse failed: {error}"))?;
gateway_control.record_sequence(&value).await;
let opcode = value.get("op").and_then(Value::as_u64).unwrap_or_default() as u8;
match opcode {
VOICE_OP_READY => {
let ready = gateway::parse_voice_ready_payload(&value)?;
let mode = gateway::choose_encryption_mode(&ready.modes)?;
let (socket, discovered) =
gateway::discover_voice_udp_address(&ready).await?;
gateway::send_voice_text(
&writer,
stream_select_protocol_payload(&discovered, &mode),
)
.await?;
gateway::send_voice_text(
&writer,
stream_receive_only_video_payload(ready.ssrc),
)
.await?;
local_ssrc = Some(ready.ssrc);
udp_socket = Some(socket);
}
VOICE_OP_SESSION_DESCRIPTION => {
let description = gateway::parse_voice_session_description(&value)?;
if description
.video_codec
.as_deref()
.is_some_and(|codec| !codec.eq_ignore_ascii_case("H264"))
{
break Err(StreamConnectionFailure::reconnect(format!(
"stream selected unsupported video codec: {}",
description.video_codec.as_deref().unwrap_or("none")
)));
}
dave_state
.lock()
.await
.apply_protocol_version(description.dave_protocol_version)?;
let Some(socket) = udp_socket.as_ref() else {
break Err(StreamConnectionFailure::reconnect(
"stream session description arrived before UDP ready",
));
};
let Some(local_ssrc) = local_ssrc else {
break Err(StreamConnectionFailure::reconnect(
"stream session description arrived before local SSRC",
));
};
if current_description.as_ref() == Some(&description) {
continue;
}
let socket_for_media = Arc::clone(socket);
let description_for_media = description.clone();
let dave_for_media = Arc::clone(&dave_state);
let source_for_media = video_source_rx.clone();
let finished = media_finished_tx.clone();
let owner_id = session.request.owner_id;
media_generation = media_generation.wrapping_add(1);
connection_stable_deadline = None;
let stream_player_ready = StreamPlayerReadySignal {
player_ready: Arc::new(AtomicBool::new(false)),
ready_tx: player_ready_tx.clone(),
media_generation,
display_name: session.request.display_name.clone(),
};
child_tasks
.replace_media(tokio::spawn(async move {
let result = run_stream_media(
socket_for_media,
description_for_media,
dave_for_media,
source_for_media,
owner_id,
local_ssrc,
stream_player_ready,
)
.await;
let _ = finished.send(result);
}))
.await;
child_tasks
.replace_keepalive(tokio::spawn(gateway::run_voice_udp_keepalive(
Arc::clone(socket),
)))
.await;
current_description = Some(description);
}
VOICE_OP_VIDEO => {
if let Some(source) =
parse_stream_video_source(&value, session.request.owner_id)
{
logging::debug(
"stream",
format!(
"stream video source selected: audio_ssrc={} video_ssrc={} rtx_ssrc={:?}",
source.audio_ssrc, source.video_ssrc, source.rtx_ssrc
),
);
{
let mut dave = dave_state.lock().await;
dave.record_ssrc_user(source.audio_ssrc, session.request.owner_id);
dave.record_ssrc_user(source.video_ssrc, session.request.owner_id);
}
gateway::send_voice_text(
&writer,
stream_media_sink_wants_payload(
source.audio_ssrc,
source.video_ssrc,
),
)
.await?;
video_source_tx.send_replace(source);
}
}
other => {
if !gateway_control
.handle_json_op(other, &value, &mut child_tasks)
.await?
{
logging::debug(
"stream",
format!("unhandled stream gateway op={other}"),
);
}
}
}
}
WsMessage::Binary(payload) => {
gateway_control.handle_binary(&payload).await?;
}
WsMessage::Ping(_) | WsMessage::Pong(_) | WsMessage::Close(_) | WsMessage::Frame(_) => {
unreachable!("gateway control frames are handled first")
}
}
};
child_tasks.shutdown().await;
result
}
fn stream_identify_payload(session: &StreamGatewaySession) -> String {
json!({
"op": 0,
"d": {
"server_id": session.rtc_server_id,
"user_id": session.current_user_id.to_string(),
"channel_id": session.rtc_channel_id.to_string(),
"session_id": session.session_id,
"token": session.token,
"video": true,
"max_dave_protocol_version": davey::DAVE_PROTOCOL_VERSION,
},
})
.to_string()
}
fn stream_select_protocol_payload(discovered: &DiscoveredVoiceAddress, mode: &str) -> String {
json!({
"op": 1,
"d": {
"protocol": "udp",
"data": {
"address": discovered.address,
"port": discovered.port,
"mode": mode,
},
"codecs": [
{
"name": "opus",
"type": "audio",
"priority": 1000,
"payload_type": DISCORD_VOICE_PAYLOAD_TYPE,
"encode": false,
"decode": true,
},
{
"name": "H264",
"type": "video",
"priority": 1000,
"payload_type": DISCORD_STREAM_VIDEO_PAYLOAD_TYPE,
"rtx_payload_type": DISCORD_STREAM_VIDEO_RTX_PAYLOAD_TYPE,
"encode": false,
"decode": true,
},
],
"rtc_connection_id": Uuid::new_v4().to_string(),
},
})
.to_string()
}
fn stream_receive_only_video_payload(audio_ssrc: u32) -> String {
json!({
"op": VOICE_OP_VIDEO,
"d": {
"audio_ssrc": audio_ssrc,
"video_ssrc": 0,
"rtx_ssrc": 0,
"streams": [],
},
})
.to_string()
}
fn stream_media_sink_wants_payload(audio_ssrc: u32, video_ssrc: u32) -> String {
let mut wants = serde_json::Map::new();
if audio_ssrc != 0 {
wants.insert(audio_ssrc.to_string(), Value::from(100));
}
if video_ssrc != 0 {
wants.insert(video_ssrc.to_string(), Value::from(100));
}
wants.insert("any".to_owned(), Value::from(0));
json!({
"op": VOICE_OP_MEDIA_SINK_WANTS,
"d": Value::Object(wants),
})
.to_string()
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
struct StreamVideoSource {
audio_ssrc: u32,
video_ssrc: u32,
rtx_ssrc: Option<u32>,
}
#[derive(Debug, Eq, PartialEq)]
struct RecoveredStreamVideoPacket {
header: RtpHeader,
payload: Vec<u8>,
}
#[derive(Default)]
struct StreamVideoRecovery {
next_sequence: Option<u16>,
pending: HashMap<u16, RecoveredStreamVideoPacket>,
gap_started_at: Option<Instant>,
last_nack_at: Option<Instant>,
}
#[derive(Default)]
struct StreamVideoRecoveryUpdate {
ready: Vec<RecoveredStreamVideoPacket>,
reset_decoder: bool,
}
impl StreamVideoRecovery {
fn push(
&mut self,
packet: RecoveredStreamVideoPacket,
now: Instant,
) -> StreamVideoRecoveryUpdate {
let sequence = packet.header.sequence;
let expected = *self.next_sequence.get_or_insert(sequence);
let distance = sequence.wrapping_sub(expected);
if distance >= 0x8000 {
return StreamVideoRecoveryUpdate::default();
}
let reset_decoder = distance > STREAM_VIDEO_REORDER_WINDOW;
if reset_decoder {
self.reset();
self.next_sequence = Some(sequence);
}
self.pending.entry(sequence).or_insert(packet);
let mut ready = Vec::new();
while let Some(expected) = self.next_sequence {
let Some(packet) = self.pending.remove(&expected) else {
break;
};
self.next_sequence = Some(expected.wrapping_add(1));
ready.push(packet);
}
if self.pending.is_empty() {
self.gap_started_at = None;
self.last_nack_at = None;
} else {
self.gap_started_at.get_or_insert(now);
}
StreamVideoRecoveryUpdate {
ready,
reset_decoder,
}
}
fn take_nack_if_due(&mut self, now: Instant) -> Option<Vec<u16>> {
if self
.last_nack_at
.is_some_and(|last| now.saturating_duration_since(last) < STREAM_VIDEO_NACK_INTERVAL)
{
return None;
}
let missing = self.missing_sequences();
if missing.is_empty() {
return None;
}
self.last_nack_at = Some(now);
Some(missing)
}
fn gap_expired(&self, now: Instant) -> bool {
self.gap_started_at.is_some_and(|started| {
now.saturating_duration_since(started) >= STREAM_VIDEO_GAP_TIMEOUT
})
}
fn reset(&mut self) {
self.next_sequence = None;
self.pending.clear();
self.gap_started_at = None;
self.last_nack_at = None;
}
fn missing_sequences(&self) -> Vec<u16> {
let Some(expected) = self.next_sequence else {
return Vec::new();
};
let Some(farthest) = self
.pending
.keys()
.map(|sequence| sequence.wrapping_sub(expected))
.filter(|distance| *distance < 0x8000)
.max()
else {
return Vec::new();
};
(0..=farthest)
.map(|distance| expected.wrapping_add(distance))
.filter(|sequence| !self.pending.contains_key(sequence))
.take(STREAM_VIDEO_MAX_NACK_SEQUENCES)
.collect()
}
}
fn recover_stream_video_packet(
header: RtpHeader,
mut payload: Vec<u8>,
source: StreamVideoSource,
) -> Option<RecoveredStreamVideoPacket> {
if header.payload_type == DISCORD_STREAM_VIDEO_PAYLOAD_TYPE && header.ssrc == source.video_ssrc
{
return Some(RecoveredStreamVideoPacket { header, payload });
}
if header.payload_type != DISCORD_STREAM_VIDEO_RTX_PAYLOAD_TYPE
|| source.rtx_ssrc != Some(header.ssrc)
{
return None;
}
let original_sequence = u16::from_be_bytes([*payload.first()?, *payload.get(1)?]);
if payload.len() <= 2 {
return None;
}
payload.drain(..2);
Some(RecoveredStreamVideoPacket {
header: RtpHeader {
payload_type: DISCORD_STREAM_VIDEO_PAYLOAD_TYPE,
sequence: original_sequence,
ssrc: source.video_ssrc,
..header
},
payload,
})
}
fn parse_stream_video_source(value: &Value, owner_id: Id<UserMarker>) -> Option<StreamVideoSource> {
let data = value.get("d")?;
if data.get("user_id").and_then(Value::as_str) != Some(owner_id.to_string().as_str()) {
return None;
}
let audio_ssrc = data
.get("audio_ssrc")
.and_then(Value::as_u64)
.and_then(|ssrc| u32::try_from(ssrc).ok())?;
let fallback_video_ssrc = data
.get("video_ssrc")
.and_then(Value::as_u64)
.and_then(|ssrc| u32::try_from(ssrc).ok())
.filter(|ssrc| *ssrc != 0);
let selected = data
.get("streams")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter(|stream| {
stream
.get("active")
.and_then(Value::as_bool)
.unwrap_or(true)
})
.filter_map(|stream| {
let ssrc = stream
.get("ssrc")
.and_then(Value::as_u64)
.and_then(|ssrc| u32::try_from(ssrc).ok())?;
let quality = stream
.get("quality")
.and_then(Value::as_u64)
.unwrap_or_default();
let rtx_ssrc = stream
.get("rtx_ssrc")
.and_then(Value::as_u64)
.and_then(|ssrc| u32::try_from(ssrc).ok());
Some((quality, ssrc, rtx_ssrc))
})
.max_by_key(|(quality, _, _)| *quality);
let (video_ssrc, rtx_ssrc) = selected
.map(|(_, ssrc, rtx)| (ssrc, rtx))
.or_else(|| fallback_video_ssrc.map(|ssrc| (ssrc, Some(ssrc.wrapping_add(1)))))?;
Some(StreamVideoSource {
audio_ssrc,
video_ssrc,
rtx_ssrc,
})
}
async fn run_stream_media(
discord_socket: Arc<UdpSocket>,
description: VoiceSessionDescription,
dave_state: Arc<Mutex<VoiceDaveState>>,
video_source_rx: watch::Receiver<StreamVideoSource>,
owner_id: Id<UserMarker>,
local_ssrc: u32,
stream_player_ready: StreamPlayerReadySignal,
) -> Result<(), StreamConnectionFailure> {
let audio_ports = reserve_local_udp_port_pair()?;
let video_ports = reserve_local_udp_port_pair()?;
let audio_port = audio_ports.rtp_port;
let audio_rtcp_port = audio_ports.rtcp_port;
let video_port = video_ports.rtp_port;
let video_rtcp_port = video_ports.rtcp_port;
let mut sdp =
NamedTempFile::new().map_err(|error| format!("create stream SDP failed: {error}"))?;
sdp.write_all(stream_sdp(audio_port, audio_rtcp_port, video_port, video_rtcp_port).as_bytes())
.map_err(|error| format!("write stream SDP failed: {error}"))?;
sdp.flush()
.map_err(|error| format!("flush stream SDP failed: {error}"))?;
drop(audio_ports);
drop(video_ports);
let mut player = stream_player_command(sdp.path(), &stream_player_ready.display_name);
let mut player = player.spawn().map_err(stream_player_spawn_failure)?;
let player_id = player.id();
let player_stdout = player
.stdout
.take()
.ok_or_else(|| StreamConnectionFailure::stop("capture stream mpv stdout failed"))?;
let player_stderr = player
.stderr
.take()
.ok_or_else(|| StreamConnectionFailure::stop("capture stream mpv stderr failed"))?;
let last_player_error = Arc::new(Mutex::new(None));
let video_player_ready = Arc::clone(&stream_player_ready.player_ready);
let player_log_tasks = StreamPlayerLogTasks::new([
tokio::spawn(log_stream_player_output(
"stream",
"stdout",
player_stdout,
Arc::clone(&last_player_error),
Some(stream_player_ready),
)),
tokio::spawn(log_stream_player_output(
"stream",
"stderr",
player_stderr,
Arc::clone(&last_player_error),
None,
)),
]);
logging::debug(
"stream",
format!(
"stream mpv started: pid={player_id:?} audio_port={audio_port} audio_rtcp_port={audio_rtcp_port} video_port={video_port} video_rtcp_port={video_rtcp_port}"
),
);
let local_socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))
.await
.map_err(|error| format!("bind local stream RTP socket failed: {error}"))?;
let audio_target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, audio_port);
let audio_rtcp_target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, audio_rtcp_port);
let video_target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, video_port);
let video_rtcp_target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, video_rtcp_port);
let decryptor = VoiceRtpDecryptor::new(&description.mode, &description.secret_key)?;
let encryptor = VoiceRtpEncryptor::new(&description.mode, &description.secret_key)?;
let mut packet = [0u8; STREAM_RTP_PACKET_BYTES];
let mut h264 = H264Depacketizer::default();
let mut h264_startup = H264StartupGate::default();
let mut h264_startup_buffer = H264StartupBuffer::default();
let mut video_recovery = StreamVideoRecovery::default();
let mut active_video_source = (0u32, None);
let mut local_audio_sequence = 0u16;
let mut local_audio_packets = 0u32;
let mut local_audio_octets = 0u32;
let mut local_video = LocalStreamVideoForwarder::default();
let mut rtcp_feedback_nonce = 0u32;
let media_started_at = Instant::now();
let mut local_audio_clock = LocalStreamAudioClock::default();
let mut logged_first_audio = false;
let mut logged_first_video_frame = false;
let mut logged_video_before_player_ready = false;
let mut logged_keyframe_request = false;
let mut logged_local_sender_reports = false;
let mut local_rtcp_report_ticks = 0u64;
let mut keyframe_request_interval = tokio::time::interval(STREAM_KEYFRAME_REQUEST_INTERVAL);
keyframe_request_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut video_recovery_interval = tokio::time::interval(STREAM_VIDEO_NACK_INTERVAL);
video_recovery_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut local_rtcp_report_interval = tokio::time::interval(LOCAL_RTCP_REPORT_INTERVAL);
local_rtcp_report_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let player_ready_timeout = tokio::time::sleep(STREAM_PLAYER_READY_TIMEOUT);
tokio::pin!(player_ready_timeout);
loop {
tokio::select! {
_ = &mut player_ready_timeout,
if !video_player_ready.load(Ordering::Acquire) =>
{
let last_error = last_player_error.lock().await.clone();
return Err(StreamConnectionFailure::stop(match last_error {
Some(error) => format!(
"stream mpv did not open its SDP input within {} seconds: {error}",
STREAM_PLAYER_READY_TIMEOUT.as_secs(),
),
None => format!(
"stream mpv did not open its SDP input within {} seconds; local RTP ports may no longer be available",
STREAM_PLAYER_READY_TIMEOUT.as_secs(),
),
}));
}
_ = local_rtcp_report_interval.tick() => {
let elapsed = media_started_at.elapsed();
let source = *video_source_rx.borrow();
let unix_time = current_unix_time();
let mut sent_report = false;
if source.audio_ssrc != 0
&& local_audio_packets != 0
&& let Some(audio_timestamp) = local_audio_clock.timestamp_at(elapsed)
{
let report = build_rtcp_sender_report(
source.audio_ssrc,
unix_time,
audio_timestamp,
local_audio_packets,
local_audio_octets,
);
let _ = local_socket.send_to(&report, audio_rtcp_target).await;
sent_report = true;
}
if source.video_ssrc != 0 && local_video.packets != 0 {
let report = build_rtcp_sender_report(
source.video_ssrc,
unix_time,
elapsed_rtp_timestamp(elapsed, VIDEO_RTP_CLOCK_RATE),
local_video.packets,
local_video.octets,
);
let _ = local_socket.send_to(&report, video_rtcp_target).await;
sent_report = true;
}
if sent_report {
local_rtcp_report_ticks = local_rtcp_report_ticks.wrapping_add(1);
if !logged_local_sender_reports {
logged_local_sender_reports = true;
logging::debug(
"stream",
format!(
"local RTCP sender reports started: elapsed_ms={}",
elapsed.as_millis()
),
);
} else if local_rtcp_report_ticks.is_multiple_of(10) {
logging::debug(
"stream",
format!(
"local stream RTP stats: elapsed_ms={} audio_packets={local_audio_packets} video_packets={} video_frames={} last_video_timestamp={:?}",
elapsed.as_millis(),
local_video.packets,
local_video.frames,
local_video.last_timestamp,
),
);
}
}
}
_ = video_recovery_interval.tick() => {
let source = *video_source_rx.borrow();
let source_identity = (source.video_ssrc, source.rtx_ssrc);
if source_identity != active_video_source {
active_video_source = source_identity;
video_recovery.reset();
reset_stream_h264_pipeline(
&mut h264,
&mut h264_startup,
&mut h264_startup_buffer,
);
}
let now = Instant::now();
if video_recovery.gap_expired(now) {
video_recovery.reset();
reset_stream_h264_pipeline(
&mut h264,
&mut h264_startup,
&mut h264_startup_buffer,
);
if source.video_ssrc != 0 {
let feedback = build_rtcp_pli(local_ssrc, source.video_ssrc);
send_stream_rtcp_feedback(
&discord_socket,
&encryptor,
&mut rtcp_feedback_nonce,
&feedback,
"PLI",
)
.await?;
}
logging::debug(
"stream",
"stream video packet gap expired; requested a new keyframe",
);
} else if let Some(missing) = video_recovery.take_nack_if_due(now)
&& source.video_ssrc != 0
{
let feedback = build_rtcp_nack(local_ssrc, source.video_ssrc, &missing);
send_stream_rtcp_feedback(
&discord_socket,
&encryptor,
&mut rtcp_feedback_nonce,
&feedback,
"NACK",
)
.await?;
}
}
_ = keyframe_request_interval.tick(), if !h264_startup.is_started() => {
let source = *video_source_rx.borrow();
if source.video_ssrc != 0 {
let feedback = build_rtcp_pli(local_ssrc, source.video_ssrc);
send_stream_rtcp_feedback(
&discord_socket,
&encryptor,
&mut rtcp_feedback_nonce,
&feedback,
"PLI",
)
.await?;
if !logged_keyframe_request {
logged_keyframe_request = true;
logging::debug(
"stream",
format!(
"stream keyframe request sent: sender_ssrc={local_ssrc} media_ssrc={}",
source.video_ssrc
),
);
}
}
}
status = player.wait() => {
let status = status.map_err(|error| {
StreamConnectionFailure::stop(format!(
"wait for stream mpv failed: {error}"
))
})?;
player_log_tasks.finish().await;
let last_error = last_player_error.lock().await.clone();
logging::debug("stream", format!("stream mpv exited: status={status}"));
if status.success() {
return Ok(());
}
return Err(StreamConnectionFailure::stop(match last_error {
Some(error) => format!("stream mpv exited with {status}: {error}"),
None => format!("stream mpv exited with {status}"),
}));
}
received = discord_socket.recv(&mut packet) => {
let received =
received.map_err(|error| format!("stream UDP receive failed: {error}"))?;
let packet = &packet[..received];
if looks_like_rtcp_packet(packet) {
continue;
}
let header = match parse_rtp_header(packet) {
Ok(header) => header,
Err(_) => continue,
};
let source = *video_source_rx.borrow();
let source_identity = (source.video_ssrc, source.rtx_ssrc);
if source_identity != active_video_source {
active_video_source = source_identity;
video_recovery.reset();
reset_stream_h264_pipeline(
&mut h264,
&mut h264_startup,
&mut h264_startup_buffer,
);
}
if header.payload_type != DISCORD_VOICE_PAYLOAD_TYPE
&& header.payload_type != DISCORD_STREAM_VIDEO_PAYLOAD_TYPE
&& header.payload_type != DISCORD_STREAM_VIDEO_RTX_PAYLOAD_TYPE
{
continue;
}
let decrypted = match decryptor.decrypt_packet_any(packet, &header) {
Ok(decrypted) => decrypted,
Err(error) => {
logging::debug("stream", format!("stream RTP decrypt failed: {error}"));
continue;
}
};
if header.payload_type == DISCORD_VOICE_PAYLOAD_TYPE
&& source.audio_ssrc != 0
&& header.ssrc == source.audio_ssrc
{
let media = dave_state
.lock()
.await
.unwrap_media_payload_for_ssrc(header.ssrc, &decrypted.media_payload);
let opus = match media {
VoiceMediaPayload::Plain(opus)
| VoiceMediaPayload::DaveDecrypted { opus, .. } => opus,
_ => continue,
};
let real_audio_at = media_started_at.elapsed();
let audio_timestamp =
local_audio_clock.rebase(header.timestamp, real_audio_at);
let local_timestamp = audio_timestamp.local_timestamp;
if let Some(discontinuity) = audio_timestamp.discontinuity {
logging::debug(
"stream",
format!(
"stream audio RTP clock re-anchored: source_timestamp={} source_delta_ticks={} elapsed_delta_ticks={} local_timestamp={local_timestamp}",
header.timestamp,
discontinuity.source_delta_ticks,
discontinuity.elapsed_delta_ticks,
),
);
}
let packet = build_local_rtp_packet(
LOCAL_STREAM_AUDIO_PAYLOAD_TYPE,
header.marker,
local_audio_sequence,
local_timestamp,
source.audio_ssrc,
&opus,
);
local_audio_sequence = local_audio_sequence.wrapping_add(1);
let _ = local_socket.send_to(&packet, audio_target).await;
local_audio_packets = local_audio_packets.wrapping_add(1);
local_audio_octets = local_audio_octets.wrapping_add(opus.len() as u32);
if !logged_first_audio {
logged_first_audio = true;
logging::debug(
"stream",
format!(
"first stream audio forwarded: elapsed_ms={} source_timestamp={} local_timestamp={}",
media_started_at.elapsed().as_millis(),
header.timestamp,
local_timestamp
),
);
}
} else if source.video_ssrc != 0
&& let Some(video_packet) =
recover_stream_video_packet(header, decrypted.media_payload, source)
{
let now = Instant::now();
let recovery = video_recovery.push(video_packet, now);
if recovery.reset_decoder {
reset_stream_h264_pipeline(
&mut h264,
&mut h264_startup,
&mut h264_startup_buffer,
);
let feedback = build_rtcp_pli(local_ssrc, source.video_ssrc);
send_stream_rtcp_feedback(
&discord_socket,
&encryptor,
&mut rtcp_feedback_nonce,
&feedback,
"PLI",
)
.await?;
}
if let Some(missing) = video_recovery.take_nack_if_due(now) {
let feedback = build_rtcp_nack(local_ssrc, source.video_ssrc, &missing);
send_stream_rtcp_feedback(
&discord_socket,
&encryptor,
&mut rtcp_feedback_nonce,
&feedback,
"NACK",
)
.await?;
}
for video_packet in recovery.ready {
let header = video_packet.header;
let frame = match h264.push(&header, &video_packet.payload) {
H264DepacketizerOutput::Pending => continue,
H264DepacketizerOutput::Frame(frame) => frame,
H264DepacketizerOutput::BudgetExceeded => {
reset_stream_h264_pipeline(
&mut h264,
&mut h264_startup,
&mut h264_startup_buffer,
);
let feedback = build_rtcp_pli(local_ssrc, source.video_ssrc);
send_stream_rtcp_feedback(
&discord_socket,
&encryptor,
&mut rtcp_feedback_nonce,
&feedback,
"PLI",
)
.await?;
logging::debug(
"stream",
"stream H264 access unit exceeded its safety budget; requested a new keyframe",
);
continue;
}
};
let frame = match dave_state
.lock()
.await
.decrypt_video_frame(owner_id, &frame)
{
Ok(Some(frame)) => frame,
Ok(None) => continue,
Err(error) => {
logging::debug("stream", error);
continue;
}
};
if !logged_first_video_frame {
logged_first_video_frame = true;
logging::debug(
"stream",
format!(
"first stream video frame decrypted: elapsed_ms={} nal_types={:?}",
media_started_at.elapsed().as_millis(),
h264_nal_types(&frame),
),
);
}
let player_ready = video_player_ready.load(Ordering::Acquire);
let local_video_destination = LocalStreamVideoDestination {
socket: &local_socket,
target: video_target,
ssrc: source.video_ssrc,
media_started_at,
};
if player_ready {
local_video
.replay_startup(
&mut h264_startup_buffer,
&local_video_destination,
)
.await;
}
if !player_ready && !logged_video_before_player_ready {
logged_video_before_player_ready = true;
logging::debug(
"stream",
"buffering stream video until mpv opens its SDP input",
);
}
let waiting_for_keyframe = !h264_startup.is_started();
let frame = accept_or_buffer_h264(
player_ready,
&mut h264_startup,
&mut h264_startup_buffer,
frame,
header.timestamp,
);
if waiting_for_keyframe && h264_startup.is_started() {
logging::debug(
"stream",
format!(
"stream H264 keyframe {}: elapsed_ms={}",
if player_ready { "accepted" } else { "buffered" },
media_started_at.elapsed().as_millis()
),
);
}
let Some(frame) = frame else {
continue;
};
local_video
.forward_live(&local_video_destination, &frame)
.await;
}
}
}
}
}
}
fn stream_player_spawn_failure(error: std::io::Error) -> StreamConnectionFailure {
let message = if error.kind() == std::io::ErrorKind::NotFound {
"mpv is required to watch Discord streams; install mpv and make sure it is on PATH"
.to_owned()
} else {
format!("start mpv for stream failed: {error}")
};
StreamConnectionFailure::stop(message)
}
fn stream_player_command(sdp_path: &Path, display_name: &str) -> Command {
let mut player = Command::new("mpv");
player
.arg("--no-config")
.arg("--terminal=yes")
.arg("--load-scripts=no")
.arg("--msg-level=all=warn,cplayer=v,lavf=v,vd=v,ad=v")
.arg("--profile=low-latency")
.arg("--cache=no")
.arg("--demuxer-readahead-secs=0")
.arg("--demuxer=lavf")
.arg("--demuxer-lavf-format=sdp")
.arg("--demuxer-lavf-probesize=32")
.arg("--demuxer-lavf-buffersize=4096")
.arg("--demuxer-lavf-o=protocol_whitelist=[file,udp,rtp],max_delay=0,reorder_queue_size=0")
.arg("--force-window=immediate")
.arg(format!("--title={display_name}'s stream"))
.arg("--video-sync=desync")
.arg("--framedrop=vo")
.arg("--video-timing-offset=0")
.arg("--")
.arg(sdp_path)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
player
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct StreamAudioClockDiscontinuity {
source_delta_ticks: i32,
elapsed_delta_ticks: u32,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct StreamAudioTimestamp {
local_timestamp: u32,
discontinuity: Option<StreamAudioClockDiscontinuity>,
}
#[derive(Default)]
struct LocalStreamAudioClock {
last_source_timestamp: Option<u32>,
last_local_timestamp: Option<u32>,
last_elapsed: Option<Duration>,
}
impl LocalStreamAudioClock {
fn rebase(&mut self, source_timestamp: u32, elapsed: Duration) -> StreamAudioTimestamp {
let Some(last_source_timestamp) = self.last_source_timestamp else {
let local_timestamp = elapsed_rtp_timestamp(elapsed, OPUS_RTP_CLOCK_RATE);
self.record(source_timestamp, local_timestamp, elapsed);
return StreamAudioTimestamp {
local_timestamp,
discontinuity: None,
};
};
let last_local_timestamp = self
.last_local_timestamp
.expect("an anchored audio clock has a local timestamp");
let last_elapsed = self
.last_elapsed
.expect("an anchored audio clock has an elapsed timestamp");
let elapsed_delta_ticks =
elapsed_rtp_timestamp(elapsed.saturating_sub(last_elapsed), OPUS_RTP_CLOCK_RATE);
let source_delta_ticks = source_timestamp.wrapping_sub(last_source_timestamp) as i32;
let drift_ticks = i64::from(source_delta_ticks).abs_diff(i64::from(elapsed_delta_ticks));
let source_is_continuous = source_delta_ticks > 0
&& drift_ticks <= u64::from(STREAM_AUDIO_CLOCK_DRIFT_TOLERANCE_TICKS);
let (local_timestamp, discontinuity) = if source_is_continuous {
(
last_local_timestamp.wrapping_add(source_delta_ticks as u32),
None,
)
} else {
let elapsed_timestamp = elapsed_rtp_timestamp(elapsed, OPUS_RTP_CLOCK_RATE);
let minimum_timestamp =
last_local_timestamp.wrapping_add(DISCORD_OPUS_TIMESTAMP_INCREMENT);
(
later_rtp_timestamp(elapsed_timestamp, minimum_timestamp),
Some(StreamAudioClockDiscontinuity {
source_delta_ticks,
elapsed_delta_ticks,
}),
)
};
self.record(source_timestamp, local_timestamp, elapsed);
StreamAudioTimestamp {
local_timestamp,
discontinuity,
}
}
fn timestamp_at(&self, elapsed: Duration) -> Option<u32> {
let last_local_timestamp = self.last_local_timestamp?;
let last_elapsed = self.last_elapsed?;
Some(last_local_timestamp.wrapping_add(elapsed_rtp_timestamp(
elapsed.saturating_sub(last_elapsed),
OPUS_RTP_CLOCK_RATE,
)))
}
fn record(&mut self, source_timestamp: u32, local_timestamp: u32, elapsed: Duration) {
self.last_source_timestamp = Some(source_timestamp);
self.last_local_timestamp = Some(local_timestamp);
self.last_elapsed = Some(elapsed);
}
}
#[derive(Default)]
struct LocalRtpClock {
source_origin: Option<u32>,
local_origin: u32,
}
impl LocalRtpClock {
fn anchor(&mut self, source_timestamp: u32, local_timestamp: u32) {
self.source_origin = Some(source_timestamp);
self.local_origin = local_timestamp;
}
fn rebase(&mut self, source_timestamp: u32, elapsed: Duration, clock_rate: u32) -> u32 {
let source_origin = *self.source_origin.get_or_insert_with(|| {
self.local_origin = elapsed_rtp_timestamp(elapsed, clock_rate);
source_timestamp
});
self.local_origin
.wrapping_add(source_timestamp.wrapping_sub(source_origin))
}
}
#[derive(Default)]
struct LocalStreamVideoForwarder {
sequence: u16,
packets: u32,
octets: u32,
frames: u64,
last_timestamp: Option<u32>,
clock: LocalRtpClock,
logged_first_frame: bool,
}
struct LocalStreamVideoDestination<'a> {
socket: &'a UdpSocket,
target: SocketAddrV4,
ssrc: u32,
media_started_at: Instant,
}
impl LocalStreamVideoForwarder {
async fn replay_startup(
&mut self,
startup_buffer: &mut H264StartupBuffer,
destination: &LocalStreamVideoDestination<'_>,
) {
if startup_buffer.is_empty() {
return;
}
let buffered_frames = startup_buffer.len();
let replay_origin =
elapsed_rtp_timestamp(destination.media_started_at.elapsed(), VIDEO_RTP_CLOCK_RATE);
let mut replay_anchor = None;
for (index, buffered) in startup_buffer.take().into_iter().enumerate() {
let local_timestamp = replay_origin.wrapping_add(
u32::try_from(index)
.expect("startup buffer length is bounded")
.wrapping_mul(STREAM_STARTUP_REPLAY_FRAME_TICKS),
);
self.forward_at_timestamp(destination, &buffered, local_timestamp, true)
.await;
replay_anchor = Some((buffered.source_timestamp, local_timestamp));
}
if let Some((source_timestamp, local_timestamp)) = replay_anchor {
self.clock.anchor(source_timestamp, local_timestamp);
}
logging::debug(
"stream",
format!(
"buffered H264 startup replayed: elapsed_ms={} frames={buffered_frames}",
destination.media_started_at.elapsed().as_millis()
),
);
}
async fn forward_live(
&mut self,
destination: &LocalStreamVideoDestination<'_>,
frame: &BufferedH264Frame,
) {
let local_timestamp = self.clock.rebase(
frame.source_timestamp,
destination.media_started_at.elapsed(),
VIDEO_RTP_CLOCK_RATE,
);
self.forward_at_timestamp(destination, frame, local_timestamp, false)
.await;
}
async fn forward_at_timestamp(
&mut self,
destination: &LocalStreamVideoDestination<'_>,
frame: &BufferedH264Frame,
local_timestamp: u32,
buffered: bool,
) {
let (packet_count, octet_count) = send_local_h264_frame(
destination.socket,
destination.target,
&frame.encoded,
local_timestamp,
destination.ssrc,
&mut self.sequence,
)
.await;
self.packets = self.packets.wrapping_add(packet_count);
self.octets = self.octets.wrapping_add(octet_count);
self.frames = self.frames.wrapping_add(1);
self.last_timestamp = Some(local_timestamp);
if !self.logged_first_frame {
self.logged_first_frame = true;
logging::debug(
"stream",
format!(
"first {}stream video forwarded: elapsed_ms={} source_timestamp={} local_timestamp={local_timestamp}",
if buffered { "buffered " } else { "" },
destination.media_started_at.elapsed().as_millis(),
frame.source_timestamp,
),
);
}
}
}
fn elapsed_rtp_timestamp(elapsed: Duration, clock_rate: u32) -> u32 {
let whole = elapsed.as_secs().wrapping_mul(u64::from(clock_rate));
let fractional =
u64::from(elapsed.subsec_nanos()).wrapping_mul(u64::from(clock_rate)) / 1_000_000_000;
whole.wrapping_add(fractional) as u32
}
fn later_rtp_timestamp(left: u32, right: u32) -> u32 {
if left.wrapping_sub(right) as i32 >= 0 {
left
} else {
right
}
}
fn build_rtcp_pli(sender_ssrc: u32, media_ssrc: u32) -> [u8; 12] {
let mut packet = [0u8; 12];
packet[0] = (RTP_VERSION << 6) | RTCP_PLI_FORMAT;
packet[1] = RTCP_PAYLOAD_SPECIFIC_FEEDBACK;
packet[2..4].copy_from_slice(&RTCP_PLI_LENGTH_WORDS_MINUS_ONE.to_be_bytes());
packet[4..8].copy_from_slice(&sender_ssrc.to_be_bytes());
packet[8..12].copy_from_slice(&media_ssrc.to_be_bytes());
packet
}
fn build_rtcp_nack(sender_ssrc: u32, media_ssrc: u32, missing: &[u16]) -> Vec<u8> {
let mut feedback_control = Vec::new();
let mut index = 0usize;
while let Some(&packet_id) = missing.get(index) {
index += 1;
let mut bitmask = 0u16;
while let Some(&sequence) = missing.get(index) {
let distance = sequence.wrapping_sub(packet_id);
if !(1..=16).contains(&distance) {
break;
}
bitmask |= 1 << (distance - 1);
index += 1;
}
feedback_control.extend_from_slice(&packet_id.to_be_bytes());
feedback_control.extend_from_slice(&bitmask.to_be_bytes());
}
let mut packet = Vec::with_capacity(12 + feedback_control.len());
packet.push((RTP_VERSION << 6) | RTCP_GENERIC_NACK_FORMAT);
packet.push(RTCP_TRANSPORT_LAYER_FEEDBACK);
let length_words_minus_one =
u16::try_from((12 + feedback_control.len()) / 4 - 1).expect("RTCP NACK length fits u16");
packet.extend_from_slice(&length_words_minus_one.to_be_bytes());
packet.extend_from_slice(&sender_ssrc.to_be_bytes());
packet.extend_from_slice(&media_ssrc.to_be_bytes());
packet.extend_from_slice(&feedback_control);
packet
}
async fn send_stream_rtcp_feedback(
socket: &UdpSocket,
encryptor: &VoiceRtpEncryptor,
nonce: &mut u32,
feedback: &[u8],
kind: &str,
) -> Result<(), String> {
let encrypted = encryptor.encrypt_rtcp_feedback(feedback, nonce.to_be_bytes())?;
*nonce = nonce
.checked_add(1)
.ok_or_else(|| "stream RTCP feedback nonce exhausted".to_owned())?;
socket
.send(&encrypted)
.await
.map_err(|error| format!("send stream RTCP {kind} failed: {error}"))?;
Ok(())
}
fn reset_stream_h264_pipeline(
h264: &mut H264Depacketizer,
startup: &mut H264StartupGate,
startup_buffer: &mut H264StartupBuffer,
) {
*h264 = H264Depacketizer::default();
*startup = H264StartupGate::default();
startup_buffer.clear();
}
async fn log_stream_player_output(
kind: &'static str,
output: &'static str,
stream: impl AsyncRead + Unpin,
last_error: Arc<Mutex<Option<String>>>,
player_ready: Option<StreamPlayerReadySignal>,
) {
let mut lines = BufReader::new(stream).lines();
loop {
match lines.next_line().await {
Ok(Some(line)) if !line.trim().is_empty() => {
let input_ready = stream_player_input_is_ready(&line);
logging::debug("stream", format!("mpv {kind} {output}: {line}"));
*last_error.lock().await = Some(line);
if let Some(player_ready) = player_ready.as_ref()
&& input_ready
&& !player_ready.player_ready.swap(true, Ordering::AcqRel)
{
logging::debug("stream", format!("stream {kind} mpv input ready"));
let _ = player_ready.ready_tx.send(player_ready.media_generation);
}
}
Ok(Some(_)) => {}
Ok(None) => break,
Err(error) => {
logging::debug(
"stream",
format!("read mpv {kind} {output} failed: {error}"),
);
break;
}
}
}
}
fn stream_player_ready_is_current(ready_generation: Option<u64>, media_generation: u64) -> bool {
ready_generation == Some(media_generation)
}
fn stream_player_input_is_ready(line: &str) -> bool {
line.contains("[cplayer] Opening done:")
}
struct ReservedLocalUdpPortPair {
_rtp_socket: std::net::UdpSocket,
_rtcp_socket: std::net::UdpSocket,
rtp_port: u16,
rtcp_port: u16,
}
fn reserve_local_udp_port_pair() -> Result<ReservedLocalUdpPortPair, String> {
for _ in 0..64 {
let rtp_socket = std::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))
.map_err(|error| format!("reserve local stream RTP port failed: {error}"))?;
let rtp_port = rtp_socket
.local_addr()
.map_err(|error| format!("read local stream RTP port failed: {error}"))?
.port();
let Some(rtcp_port) = rtp_port
.checked_add(1)
.filter(|_| rtp_port.is_multiple_of(2))
else {
continue;
};
let Ok(rtcp_socket) = std::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, rtcp_port)) else {
continue;
};
return Ok(ReservedLocalUdpPortPair {
_rtp_socket: rtp_socket,
_rtcp_socket: rtcp_socket,
rtp_port,
rtcp_port,
});
}
Err("reserve adjacent local stream RTP and RTCP ports failed".to_owned())
}
fn stream_sdp(
audio_port: u16,
audio_rtcp_port: u16,
video_port: u16,
video_rtcp_port: u16,
) -> String {
format!(
"v=0\r\n\
o=- 0 0 IN IP4 127.0.0.1\r\n\
s=Concord Discord Stream\r\n\
c=IN IP4 127.0.0.1\r\n\
t=0 0\r\n\
m=audio {audio_port} RTP/AVP {LOCAL_STREAM_AUDIO_PAYLOAD_TYPE}\r\n\
a=rtcp:{audio_rtcp_port} IN IP4 127.0.0.1\r\n\
a=rtpmap:{LOCAL_STREAM_AUDIO_PAYLOAD_TYPE} opus/48000/2\r\n\
a=recvonly\r\n\
m=video {video_port} RTP/AVP {LOCAL_STREAM_VIDEO_PAYLOAD_TYPE}\r\n\
a=rtcp:{video_rtcp_port} IN IP4 127.0.0.1\r\n\
a=rtpmap:{LOCAL_STREAM_VIDEO_PAYLOAD_TYPE} H264/90000\r\n\
a=fmtp:{LOCAL_STREAM_VIDEO_PAYLOAD_TYPE} packetization-mode=1\r\n\
a=recvonly\r\n"
)
}
fn build_local_rtp_packet(
payload_type: u8,
marker: bool,
sequence: u16,
timestamp: u32,
ssrc: u32,
payload: &[u8],
) -> Vec<u8> {
let mut packet = Vec::with_capacity(RTP_HEADER_MIN_LEN + payload.len());
packet.push(RTP_VERSION << 6);
packet.push((u8::from(marker) << 7) | payload_type);
packet.extend_from_slice(&sequence.to_be_bytes());
packet.extend_from_slice(×tamp.to_be_bytes());
packet.extend_from_slice(&ssrc.to_be_bytes());
packet.extend_from_slice(payload);
packet
}
#[derive(Default)]
struct H264Depacketizer {
timestamp: Option<u32>,
expected_sequence: Option<u16>,
frame: Vec<u8>,
packet_count: usize,
fragment_open: bool,
}
enum H264DepacketizerOutput {
Pending,
Frame(Vec<u8>),
BudgetExceeded,
}
enum H264AppendError {
Invalid,
BudgetExceeded,
}
#[derive(Default)]
struct H264StartupGate {
started: bool,
sps: Option<Vec<u8>>,
pps: Option<Vec<u8>>,
}
struct BufferedH264Frame {
encoded: Vec<u8>,
source_timestamp: u32,
}
#[derive(Default)]
struct H264StartupBuffer {
frames: VecDeque<BufferedH264Frame>,
bytes: usize,
}
impl H264StartupGate {
fn is_started(&self) -> bool {
self.started
}
fn accept(&mut self, frame: Vec<u8>) -> Option<Vec<u8>> {
let has_idr = {
let mut has_idr = false;
for nal in annex_b_nals(&frame) {
match nal.first().map(|byte| byte & 0x1f) {
Some(5) => has_idr = true,
Some(7) => {
self.sps = Some(nal.to_vec());
}
Some(8) => {
self.pps = Some(nal.to_vec());
}
_ => {}
}
}
has_idr
};
if self.started {
return Some(frame);
}
if !has_idr || self.sps.is_none() || self.pps.is_none() {
return None;
}
self.started = true;
let mut startup_frame = Vec::new();
append_annex_b_nal(
&mut startup_frame,
self.sps.as_deref().expect("startup requires an SPS"),
);
append_annex_b_nal(
&mut startup_frame,
self.pps.as_deref().expect("startup requires a PPS"),
);
for nal in annex_b_nals(&frame) {
if !matches!(nal.first().map(|byte| byte & 0x1f), Some(7 | 8)) {
append_annex_b_nal(&mut startup_frame, nal);
}
}
Some(startup_frame)
}
}
impl H264StartupBuffer {
fn is_empty(&self) -> bool {
self.frames.is_empty()
}
fn len(&self) -> usize {
self.frames.len()
}
fn push(&mut self, frame: BufferedH264Frame) -> bool {
let Some(next_bytes) = self.bytes.checked_add(frame.encoded.len()) else {
self.clear();
return false;
};
if self.frames.len() >= STREAM_STARTUP_BUFFER_MAX_FRAMES
|| next_bytes > STREAM_STARTUP_BUFFER_MAX_BYTES
{
self.clear();
return false;
}
self.bytes = next_bytes;
self.frames.push_back(frame);
true
}
fn take(&mut self) -> VecDeque<BufferedH264Frame> {
self.bytes = 0;
std::mem::take(&mut self.frames)
}
fn clear(&mut self) {
self.frames.clear();
self.bytes = 0;
}
}
fn accept_or_buffer_h264(
player_ready: bool,
startup: &mut H264StartupGate,
startup_buffer: &mut H264StartupBuffer,
frame: Vec<u8>,
source_timestamp: u32,
) -> Option<BufferedH264Frame> {
let frame = BufferedH264Frame {
encoded: startup.accept(frame)?,
source_timestamp,
};
if player_ready {
return Some(frame);
}
if !startup_buffer.push(frame) {
*startup = H264StartupGate::default();
}
None
}
fn append_annex_b_nal(frame: &mut Vec<u8>, nal: &[u8]) {
frame.extend_from_slice(&[0, 0, 0, 1]);
frame.extend_from_slice(nal);
}
fn h264_nal_types(frame: &[u8]) -> Vec<u8> {
annex_b_nals(frame)
.into_iter()
.filter_map(|nal| nal.first().map(|byte| byte & 0x1f))
.collect()
}
impl H264Depacketizer {
fn push(&mut self, header: &RtpHeader, payload: &[u8]) -> H264DepacketizerOutput {
if self.timestamp != Some(header.timestamp)
|| self
.expected_sequence
.is_some_and(|expected| expected != header.sequence)
{
self.reset(header.timestamp);
}
if self.packet_count >= STREAM_H264_ACCESS_UNIT_MAX_PACKETS {
self.reset(header.timestamp);
return H264DepacketizerOutput::BudgetExceeded;
}
self.packet_count += 1;
self.expected_sequence = Some(header.sequence.wrapping_add(1));
let Some(nal_type) = payload.first().map(|byte| byte & 0x1f) else {
self.reset(header.timestamp);
return H264DepacketizerOutput::Pending;
};
let accepted = match nal_type {
1..=23 => self.append_nal(payload),
24 => self.append_stap_a(payload),
28 => self.append_fu_a(payload),
_ => Err(H264AppendError::Invalid),
};
match accepted {
Ok(()) => {}
Err(H264AppendError::Invalid) => {
self.reset(header.timestamp);
return H264DepacketizerOutput::Pending;
}
Err(H264AppendError::BudgetExceeded) => {
self.reset(header.timestamp);
return H264DepacketizerOutput::BudgetExceeded;
}
}
if header.marker {
if self.fragment_open || self.frame.is_empty() {
self.reset(header.timestamp);
return H264DepacketizerOutput::Pending;
}
self.timestamp = None;
self.expected_sequence = None;
self.packet_count = 0;
return H264DepacketizerOutput::Frame(std::mem::take(&mut self.frame));
}
H264DepacketizerOutput::Pending
}
fn reset(&mut self, timestamp: u32) {
self.timestamp = Some(timestamp);
self.expected_sequence = None;
self.frame.clear();
self.packet_count = 0;
self.fragment_open = false;
}
fn append_nal(&mut self, nal: &[u8]) -> Result<(), H264AppendError> {
self.ensure_capacity(4usize.saturating_add(nal.len()))?;
self.frame.extend_from_slice(&[0, 0, 0, 1]);
self.frame.extend_from_slice(nal);
self.fragment_open = false;
Ok(())
}
fn append_stap_a(&mut self, payload: &[u8]) -> Result<(), H264AppendError> {
let mut cursor = 1usize;
let mut required_bytes = 0usize;
let mut nal_count = 0usize;
while cursor + 2 <= payload.len() {
let size = usize::from(u16::from_be_bytes([payload[cursor], payload[cursor + 1]]));
cursor += 2;
let Some(nal) = payload.get(cursor..cursor.saturating_add(size)) else {
return Err(H264AppendError::Invalid);
};
if nal.is_empty() {
return Err(H264AppendError::Invalid);
}
required_bytes = required_bytes
.checked_add(4)
.and_then(|bytes| bytes.checked_add(nal.len()))
.ok_or(H264AppendError::BudgetExceeded)?;
cursor += size;
nal_count += 1;
}
if nal_count == 0 || cursor != payload.len() {
return Err(H264AppendError::Invalid);
}
self.ensure_capacity(required_bytes)?;
cursor = 1;
while cursor + 2 <= payload.len() {
let size = usize::from(u16::from_be_bytes([payload[cursor], payload[cursor + 1]]));
cursor += 2;
let nal = &payload[cursor..cursor + size];
self.frame.extend_from_slice(&[0, 0, 0, 1]);
self.frame.extend_from_slice(nal);
cursor += size;
}
self.fragment_open = false;
Ok(())
}
fn append_fu_a(&mut self, payload: &[u8]) -> Result<(), H264AppendError> {
if payload.len() < 3 {
return Err(H264AppendError::Invalid);
}
let indicator = payload[0];
let fu_header = payload[1];
let start = fu_header & 0x80 != 0;
let end = fu_header & 0x40 != 0;
let header_bytes = if start { 5 } else { 0 };
self.ensure_capacity(header_bytes + payload.len() - 2)?;
if start {
self.frame.extend_from_slice(&[0, 0, 0, 1]);
self.frame.push((indicator & 0xe0) | (fu_header & 0x1f));
self.fragment_open = !end;
} else if !self.fragment_open {
return Err(H264AppendError::Invalid);
} else if end {
self.fragment_open = false;
}
self.frame.extend_from_slice(&payload[2..]);
Ok(())
}
fn ensure_capacity(&self, additional_bytes: usize) -> Result<(), H264AppendError> {
if self
.frame
.len()
.checked_add(additional_bytes)
.is_some_and(|bytes| bytes <= STREAM_H264_ACCESS_UNIT_MAX_BYTES)
{
Ok(())
} else {
Err(H264AppendError::BudgetExceeded)
}
}
}
fn packetize_h264_frame(
frame: &[u8],
timestamp: u32,
ssrc: u32,
sequence: &mut u16,
) -> Vec<Vec<u8>> {
let payloads = packetize_h264_payloads(frame, LOCAL_H264_MAX_PAYLOAD_BYTES);
let payload_count = payloads.len();
payloads
.into_iter()
.enumerate()
.map(|(index, payload)| {
let packet = build_local_rtp_packet(
LOCAL_STREAM_VIDEO_PAYLOAD_TYPE,
index + 1 == payload_count,
*sequence,
timestamp,
ssrc,
&payload,
);
*sequence = sequence.wrapping_add(1);
packet
})
.collect()
}
async fn send_local_h264_frame(
socket: &UdpSocket,
target: SocketAddrV4,
frame: &[u8],
timestamp: u32,
ssrc: u32,
sequence: &mut u16,
) -> (u32, u32) {
let packets = packetize_h264_frame(frame, timestamp, ssrc, sequence);
let packet_count = u32::try_from(packets.len()).expect("H264 packet count fits u32");
let octet_count = packets.iter().fold(0u32, |total, packet| {
total.wrapping_add(packet.len().saturating_sub(RTP_HEADER_MIN_LEN) as u32)
});
for packet in packets {
let _ = socket.send_to(&packet, target).await;
}
(packet_count, octet_count)
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::sync::oneshot;
struct TaskDropSignal(Option<oneshot::Sender<()>>);
impl Drop for TaskDropSignal {
fn drop(&mut self) {
if let Some(dropped) = self.0.take() {
let _ = dropped.send(());
}
}
}
fn cancellable_test_task() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Receiver<()>) {
let (started_tx, started_rx) = oneshot::channel();
let (dropped_tx, dropped_rx) = oneshot::channel();
let task = tokio::spawn(async move {
let _drop_signal = TaskDropSignal(Some(dropped_tx));
let _ = started_tx.send(());
std::future::pending::<()>().await;
});
(task, started_rx, dropped_rx)
}
#[tokio::test]
async fn dropping_stream_child_tasks_aborts_every_child() {
let (heartbeat, heartbeat_started, heartbeat_dropped) = cancellable_test_task();
let (keepalive, keepalive_started, keepalive_dropped) = cancellable_test_task();
let (media, media_started, media_dropped) = cancellable_test_task();
let mut child_tasks = GatewayChildTasks::default();
child_tasks.replace_heartbeat(heartbeat).await;
child_tasks.replace_keepalive(keepalive).await;
child_tasks.replace_media(media).await;
heartbeat_started.await.expect("heartbeat test task starts");
keepalive_started.await.expect("keepalive test task starts");
media_started.await.expect("media test task starts");
drop(child_tasks);
for dropped in [heartbeat_dropped, keepalive_dropped, media_dropped] {
timeout(Duration::from_secs(1), dropped)
.await
.expect("stream child task is aborted promptly")
.expect("stream child task drop signal is sent");
}
}
#[tokio::test]
async fn dropping_stream_player_log_tasks_aborts_both_readers() {
let (stdout, stdout_started, stdout_dropped) = cancellable_test_task();
let (stderr, stderr_started, stderr_dropped) = cancellable_test_task();
let log_tasks = StreamPlayerLogTasks::new([stdout, stderr]);
stdout_started.await.expect("stdout test task starts");
stderr_started.await.expect("stderr test task starts");
drop(log_tasks);
for dropped in [stdout_dropped, stderr_dropped] {
timeout(Duration::from_secs(1), dropped)
.await
.expect("stream player log task is aborted promptly")
.expect("stream player log task drop signal is sent");
}
}
#[test]
fn stream_player_readiness_accepts_only_the_current_media_generation() {
assert!(stream_player_ready_is_current(Some(8), 8));
assert!(!stream_player_ready_is_current(Some(7), 8));
assert!(!stream_player_ready_is_current(None, 8));
}
fn stream_request() -> StreamWatchRequest {
StreamWatchRequest {
stream_key: "guild:10:20:99".to_owned(),
scope: VoiceScope::Guild(Id::new(10)),
channel_id: Id::new(20),
owner_id: Id::new(99),
display_name: "Streamer".to_owned(),
}
}
fn current_voice_state() -> VoiceStateInfo {
VoiceStateInfo {
guild_id: Some(Id::new(10)),
channel_id: Some(Id::new(20)),
user_id: Id::new(5),
session_id: Some("parent-session".to_owned()),
member: None,
deaf: false,
mute: false,
self_deaf: false,
self_mute: false,
self_stream: false,
}
}
fn connected_stream_runtime() -> (StreamRuntimeState, StreamGatewaySession) {
let mut state = StreamRuntimeState::default();
state.apply(&VoiceRuntimeEvent::WatchStreamRequested(stream_request()));
state.apply(&VoiceRuntimeEvent::CurrentUserReady(Some(Id::new(5))));
state.apply(&VoiceRuntimeEvent::VoiceState(current_voice_state()));
state.apply(&VoiceRuntimeEvent::StreamCreate(StreamCreateInfo {
stream_key: "guild:10:20:99".to_owned(),
rtc_server_id: "400".to_owned(),
rtc_channel_id: Id::new(401),
viewer_ids: Vec::new(),
paused: false,
}));
let update = state.apply(&VoiceRuntimeEvent::StreamServer(StreamServerInfo {
stream_key: "guild:10:20:99".to_owned(),
endpoint: Some("stream.example.com".to_owned()),
token: "stream-token".to_owned(),
}));
let session = update.connect.expect("stream session should be ready");
(state, session)
}
#[test]
fn stream_gateway_session_debug_redacts_token() {
let (_, session) = connected_stream_runtime();
let debug = format!("{session:?}");
assert!(!debug.contains("stream-token"));
assert!(debug.contains("<redacted>"));
}
fn stream_connection_ended(
session: &StreamGatewaySession,
outcome: VoiceConnectionEnd,
) -> VoiceRuntimeEvent {
VoiceRuntimeEvent::StreamConnectionEnded {
connection_id: session.connection_id,
stream_key: session.request.stream_key.clone(),
outcome,
}
}
#[test]
fn h264_fu_a_round_trips_through_local_packetizer() {
let frame = [0, 0, 0, 1, 0x65]
.into_iter()
.chain(std::iter::repeat_n(0xaa, 3000))
.collect::<Vec<_>>();
let mut sequence = 7;
let packets = packetize_h264_frame(&frame, 90_000, 42, &mut sequence);
assert!(packets.len() > 1);
let mut depacketizer = H264Depacketizer::default();
let mut decoded = None;
for packet in packets {
let header = parse_rtp_header(&packet).expect("local RTP packet is valid");
if let H264DepacketizerOutput::Frame(frame) =
depacketizer.push(&header, &packet[header.payload_offset..])
{
decoded = Some(frame);
}
}
assert_eq!(decoded, Some(frame));
}
#[test]
fn h264_assembly_enforces_packet_and_memory_budgets() {
let mut oversized = H264Depacketizer::default();
let mut oversized_header =
stream_video_header(DISCORD_STREAM_VIDEO_PAYLOAD_TYPE, 1, 90_000, 42);
oversized_header.marker = false;
let mut oversized_payload = vec![0xaa; STREAM_H264_ACCESS_UNIT_MAX_BYTES + 1];
oversized_payload[0] = 0x7c;
oversized_payload[1] = 0x85;
assert!(matches!(
oversized.push(&oversized_header, &oversized_payload),
H264DepacketizerOutput::BudgetExceeded
));
let mut fragmented = H264Depacketizer::default();
let mut fragment_header =
stream_video_header(DISCORD_STREAM_VIDEO_PAYLOAD_TYPE, 1, 93_000, 42);
fragment_header.marker = false;
assert!(matches!(
fragmented.push(&fragment_header, &[0x7c, 0x85, 0xaa]),
H264DepacketizerOutput::Pending
));
for sequence in 2..=STREAM_H264_ACCESS_UNIT_MAX_PACKETS {
fragment_header.sequence = u16::try_from(sequence).expect("packet budget fits u16");
assert!(matches!(
fragmented.push(&fragment_header, &[0x7c, 0x05, 0xaa]),
H264DepacketizerOutput::Pending
));
}
fragment_header.sequence =
u16::try_from(STREAM_H264_ACCESS_UNIT_MAX_PACKETS + 1).expect("packet budget fits u16");
assert!(matches!(
fragmented.push(&fragment_header, &[0x7c, 0x45, 0xaa]),
H264DepacketizerOutput::BudgetExceeded
));
let mut startup = H264StartupBuffer::default();
for timestamp in 0..32 {
assert!(startup.push(BufferedH264Frame {
encoded: vec![0; 1024 * 1024],
source_timestamp: timestamp,
}));
}
assert!(!startup.push(BufferedH264Frame {
encoded: vec![0],
source_timestamp: 32,
}));
assert!(startup.is_empty());
assert_eq!(startup.bytes, 0);
}
fn stream_video_header(
payload_type: u8,
sequence: u16,
timestamp: u32,
ssrc: u32,
) -> RtpHeader {
RtpHeader {
has_padding: false,
marker: true,
payload_type,
sequence,
timestamp,
ssrc,
authenticated_header_len: RTP_HEADER_MIN_LEN,
encrypted_extension_body_len: 0,
payload_offset: RTP_HEADER_MIN_LEN,
}
}
#[test]
fn stream_video_reorders_primary_and_rtx_packets_before_depacketization() {
let source = StreamVideoSource {
audio_ssrc: 7,
video_ssrc: 42,
rtx_ssrc: Some(43),
};
let now = Instant::now();
let mut recovery = StreamVideoRecovery::default();
let first_payload = b"first".to_vec();
let first_payload_ptr = first_payload.as_ptr();
let first = recover_stream_video_packet(
stream_video_header(DISCORD_STREAM_VIDEO_PAYLOAD_TYPE, 10, 90_000, 42),
first_payload,
source,
)
.expect("primary video packet should be accepted");
assert_eq!(first.payload.as_ptr(), first_payload_ptr);
assert_eq!(
recovery
.push(first, now)
.ready
.into_iter()
.map(|packet| packet.header.sequence)
.collect::<Vec<_>>(),
vec![10]
);
let third = recover_stream_video_packet(
stream_video_header(DISCORD_STREAM_VIDEO_PAYLOAD_TYPE, 12, 90_000, 42),
b"third".to_vec(),
source,
)
.expect("later primary video packet should be accepted");
assert!(recovery.push(third, now).ready.is_empty());
assert_eq!(recovery.take_nack_if_due(now), Some(vec![11]));
let repaired = recover_stream_video_packet(
stream_video_header(DISCORD_STREAM_VIDEO_RTX_PAYLOAD_TYPE, 800, 90_000, 43),
vec![0, 11, b's', b'e', b'c', b'o', b'n', b'd'],
source,
)
.expect("RTX video packet should recover its original packet");
assert_eq!(repaired.header.sequence, 11);
assert_eq!(repaired.header.ssrc, source.video_ssrc);
assert_eq!(repaired.payload, b"second");
assert_eq!(
recovery
.push(repaired, now)
.ready
.into_iter()
.map(|packet| packet.header.sequence)
.collect::<Vec<_>>(),
vec![11, 12]
);
assert_eq!(recovery.take_nack_if_due(now), None);
}
#[test]
fn stream_video_recovery_expires_an_unrepaired_gap() {
let now = Instant::now();
let mut recovery = StreamVideoRecovery::default();
let first = RecoveredStreamVideoPacket {
header: stream_video_header(DISCORD_STREAM_VIDEO_PAYLOAD_TYPE, 20, 90_000, 42),
payload: vec![1],
};
let third = RecoveredStreamVideoPacket {
header: stream_video_header(DISCORD_STREAM_VIDEO_PAYLOAD_TYPE, 22, 90_000, 42),
payload: vec![3],
};
assert_eq!(recovery.push(first, now).ready.len(), 1);
assert!(recovery.push(third, now).ready.is_empty());
assert!(!recovery.gap_expired(now + STREAM_VIDEO_GAP_TIMEOUT / 2));
assert!(recovery.gap_expired(now + STREAM_VIDEO_GAP_TIMEOUT));
}
#[test]
fn stream_generic_nack_groups_missing_sequences_into_pid_and_bitmask() {
let nack = build_rtcp_nack(7, 42, &[1_000, 1_001, 1_003, 1_020]);
assert_eq!(&nack[..4], &[0x81, 205, 0, 4]);
assert_eq!(&nack[4..8], &7u32.to_be_bytes());
assert_eq!(&nack[8..12], &42u32.to_be_bytes());
assert_eq!(&nack[12..16], &[0x03, 0xe8, 0, 5]);
assert_eq!(&nack[16..20], &[0x03, 0xfc, 0, 0]);
}
#[test]
fn stream_video_starts_at_idr_with_cached_parameter_sets() {
let parameter_sets = vec![0, 0, 0, 1, 0x67, 0x11, 0, 0, 0, 1, 0x68, 0x22];
let predicted = vec![0, 0, 0, 1, 0x41, 0x33];
let idr = vec![0, 0, 0, 1, 0x65, 0x44];
let mut gate = H264StartupGate::default();
assert_eq!(gate.accept(parameter_sets), None);
assert_eq!(gate.accept(predicted.clone()), None);
let startup = gate
.accept(idr)
.expect("IDR should start local video playback");
assert_eq!(h264_nal_types(&startup), vec![7, 8, 5]);
assert!(gate.is_started());
assert_eq!(gate.accept(predicted.clone()), Some(predicted));
}
#[test]
fn stream_video_waits_for_parameter_sets_before_accepting_idr() {
let idr = vec![0, 0, 0, 1, 0x65, 0x44];
let parameter_sets = vec![0, 0, 0, 1, 0x67, 0x11, 0, 0, 0, 1, 0x68, 0x22];
let mut gate = H264StartupGate::default();
assert_eq!(gate.accept(idr.clone()), None);
assert!(!gate.is_started());
assert_eq!(gate.accept(parameter_sets), None);
let startup = gate
.accept(idr)
.expect("IDR should start after parameter sets arrive");
assert_eq!(h264_nal_types(&startup), vec![7, 8, 5]);
assert!(gate.is_started());
}
#[test]
fn stream_video_replays_the_initial_gop_after_player_readiness() {
let startup_frame = vec![
0, 0, 0, 1, 0x67, 0x11, 0, 0, 0, 1, 0x68, 0x22, 0, 0, 0, 1, 0x65, 0x33,
];
let predicted = vec![0, 0, 0, 1, 0x41, 0x44];
let mut gate = H264StartupGate::default();
let mut buffer = H264StartupBuffer::default();
assert_eq!(
accept_or_buffer_h264(false, &mut gate, &mut buffer, startup_frame.clone(), 90_000,)
.map(|frame| frame.encoded),
None
);
assert!(gate.is_started());
assert_eq!(buffer.len(), 1);
assert_eq!(
accept_or_buffer_h264(false, &mut gate, &mut buffer, predicted.clone(), 93_000,)
.map(|frame| frame.encoded),
None
);
assert_eq!(buffer.len(), 2);
assert_eq!(
buffer
.frames
.pop_front()
.map(|frame| (frame.encoded, frame.source_timestamp)),
Some((startup_frame, 90_000))
);
assert_eq!(
buffer
.frames
.pop_front()
.map(|frame| (frame.encoded, frame.source_timestamp)),
Some((predicted.clone(), 93_000))
);
assert_eq!(
accept_or_buffer_h264(true, &mut gate, &mut buffer, predicted.clone(), 96_000)
.map(|frame| (frame.encoded, frame.source_timestamp)),
Some((predicted, 96_000))
);
let mut clock = LocalRtpClock::default();
clock.anchor(93_000, 10_000);
assert_eq!(
clock.rebase(96_000, Duration::from_secs(10), VIDEO_RTP_CLOCK_RATE),
13_000
);
assert!(stream_player_input_is_ready(
"[cplayer] Opening done: /tmp/concord-video.sdp"
));
assert!(!stream_player_input_is_ready(
"[cplayer] Starting playback..."
));
}
#[tokio::test]
async fn local_video_forwarder_replays_then_continues_the_source_clock() {
let socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))
.await
.expect("test sender should bind");
let receiver = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))
.await
.expect("test receiver should bind");
let target = match receiver
.local_addr()
.expect("test receiver should have an address")
{
std::net::SocketAddr::V4(target) => target,
std::net::SocketAddr::V6(_) => panic!("test receiver should use IPv4"),
};
let media_started_at = Instant::now();
let destination = LocalStreamVideoDestination {
socket: &socket,
target,
ssrc: 42,
media_started_at,
};
let mut startup = H264StartupBuffer::default();
assert!(startup.push(BufferedH264Frame {
encoded: vec![0, 0, 0, 1, 0x65, 0x11],
source_timestamp: 90_000,
}));
assert!(startup.push(BufferedH264Frame {
encoded: vec![0, 0, 0, 1, 0x41, 0x22],
source_timestamp: 93_000,
}));
let mut forwarder = LocalStreamVideoForwarder::default();
forwarder.replay_startup(&mut startup, &destination).await;
let replay_timestamp = forwarder
.last_timestamp
.expect("startup replay should set a local timestamp");
forwarder
.forward_live(
&destination,
&BufferedH264Frame {
encoded: vec![0, 0, 0, 1, 0x41, 0x33],
source_timestamp: 96_000,
},
)
.await;
assert!(startup.is_empty());
assert_eq!(forwarder.frames, 3);
assert_eq!(
forwarder.last_timestamp,
Some(replay_timestamp.wrapping_add(3_000))
);
}
#[test]
fn stream_keyframe_request_encrypts_only_the_rtcp_feedback_body() {
let sender_ssrc = 0x0102_0304;
let media_ssrc = 0x0506_0708;
let feedback = build_rtcp_pli(sender_ssrc, media_ssrc);
assert_eq!(&feedback[..4], &[0x81, 206, 0, 2]);
assert_eq!(&feedback[4..8], &sender_ssrc.to_be_bytes());
assert_eq!(&feedback[8..], &media_ssrc.to_be_bytes());
for mode in [AEAD_AES256_GCM_RTPSIZE, AEAD_XCHACHA20_POLY1305_RTPSIZE] {
let key = [0x42; 32];
let encryptor =
VoiceRtpEncryptor::new(mode, &key).expect("feedback encryptor should initialize");
let encrypted = encryptor
.encrypt_rtcp_feedback(&feedback, 9u32.to_be_bytes())
.expect("RTCP feedback should encrypt");
assert_eq!(&encrypted[..8], &feedback[..8]);
assert_eq!(
encrypted.len(),
8 + 4 + RTP_AEAD_TAG_BYTES + RTP_AEAD_NONCE_SUFFIX_BYTES
);
let decryptor =
VoiceRtpDecryptor::new(mode, &key).expect("feedback decryptor should initialize");
let decrypted = decryptor
.decrypt_rtcp_feedback(&encrypted)
.expect("RTCP feedback body should decrypt");
assert_eq!(decrypted, feedback);
}
}
#[test]
fn stream_compound_rtcp_feedback_round_trips() {
let mut feedback = build_rtcp_pli(0x0102_0304, 0x0506_0708).to_vec();
feedback.extend_from_slice(&build_rtcp_pli(0x1112_1314, 0x1516_1718));
for mode in [AEAD_AES256_GCM_RTPSIZE, AEAD_XCHACHA20_POLY1305_RTPSIZE] {
let key = [0x42; 32];
let encryptor =
VoiceRtpEncryptor::new(mode, &key).expect("feedback encryptor should initialize");
let encrypted = encryptor
.encrypt_rtcp_feedback(&feedback, 10u32.to_be_bytes())
.expect("compound RTCP feedback should encrypt");
let decryptor =
VoiceRtpDecryptor::new(mode, &key).expect("feedback decryptor should initialize");
let decrypted = decryptor
.decrypt_rtcp_feedback(&encrypted)
.expect("compound RTCP feedback should decrypt");
assert_eq!(decrypted, feedback);
}
}
#[test]
fn local_sender_report_maps_rtp_to_a_shared_ntp_clock() {
let report = build_rtcp_sender_report(
0x0102_0304,
Duration::new(1, 500_000_000),
90_000,
30,
45_000,
);
assert_eq!(&report[..4], &[0x80, 200, 0, 6]);
assert_eq!(&report[4..8], &0x0102_0304u32.to_be_bytes());
assert_eq!(&report[8..12], &2_208_988_801u32.to_be_bytes());
assert_eq!(&report[12..16], &0x8000_0000u32.to_be_bytes());
assert_eq!(&report[16..20], &90_000u32.to_be_bytes());
assert_eq!(&report[20..24], &30u32.to_be_bytes());
assert_eq!(&report[24..28], &45_000u32.to_be_bytes());
}
#[test]
fn stream_sdp_keeps_audio_and_video_on_one_input() {
let sdp = stream_sdp(50_000, 50_001, 50_002, 50_003);
assert!(sdp.contains("m=audio 50000 RTP/AVP 111\r\n"));
assert!(sdp.contains("a=rtcp:50001 IN IP4 127.0.0.1\r\n"));
assert!(sdp.contains("m=video 50002 RTP/AVP 96\r\n"));
assert!(sdp.contains("a=rtcp:50003 IN IP4 127.0.0.1\r\n"));
}
#[test]
fn stream_video_source_prefers_highest_active_quality() {
let value = json!({
"op": 12,
"d": {
"user_id": "99",
"audio_ssrc": 10,
"video_ssrc": 20,
"streams": [
{"ssrc": 20, "rtx_ssrc": 21, "quality": 50, "active": true},
{"ssrc": 30, "rtx_ssrc": 31, "quality": 100, "active": true}
]
}
});
assert_eq!(
parse_stream_video_source(&value, Id::new(99)),
Some(StreamVideoSource {
audio_ssrc: 10,
video_ssrc: 30,
rtx_ssrc: Some(31),
})
);
}
#[test]
fn stream_runtime_waits_for_parent_voice_and_both_stream_events() {
let mut state = StreamRuntimeState::default();
assert!(
state
.apply(&VoiceRuntimeEvent::WatchStreamRequested(stream_request()))
.connect
.is_none()
);
state.apply(&VoiceRuntimeEvent::CurrentUserReady(Some(Id::new(5))));
state.apply(&VoiceRuntimeEvent::VoiceState(current_voice_state()));
state.apply(&VoiceRuntimeEvent::StreamCreate(StreamCreateInfo {
stream_key: "guild:10:20:99".to_owned(),
rtc_server_id: "400".to_owned(),
rtc_channel_id: Id::new(401),
viewer_ids: Vec::new(),
paused: false,
}));
let update = state.apply(&VoiceRuntimeEvent::StreamServer(StreamServerInfo {
stream_key: "guild:10:20:99".to_owned(),
endpoint: Some("stream.example.com".to_owned()),
token: "stream-token".to_owned(),
}));
let session = update.connect.expect("stream session should now be ready");
assert_eq!(session.session_id, "parent-session");
assert_eq!(session.rtc_server_id, "400");
assert_eq!(session.rtc_channel_id, Id::new(401));
assert_eq!(session.request.owner_id, Id::new(99));
}
#[test]
fn stream_runtime_pending_cancel_ends_playback_once() {
let request = stream_request();
let mut state = StreamRuntimeState::default();
state.apply(&VoiceRuntimeEvent::WatchStreamRequested(request.clone()));
let cancelled = state.apply(&VoiceRuntimeEvent::WatchStreamCancelled {
stream_key: request.stream_key.clone(),
});
let ended = cancelled
.playback_ended
.expect("pending stream cancellation ends preparing playback");
assert_eq!(ended.request, request);
assert!(!ended.reconnecting);
let repeated = state.apply(&VoiceRuntimeEvent::WatchStreamCancelled {
stream_key: ended.request.stream_key,
});
assert!(repeated.playback_ended.is_none());
}
#[test]
fn stream_runtime_rotates_active_stream_servers() {
let (mut state, initial) = connected_stream_runtime();
let rotated = state.apply(&VoiceRuntimeEvent::StreamServer(StreamServerInfo {
stream_key: initial.request.stream_key.clone(),
endpoint: Some("replacement.example.com".to_owned()),
token: "replacement-token".to_owned(),
}));
assert_eq!(
rotated.close_stream_key.as_deref(),
Some(initial.request.stream_key.as_str())
);
assert!(!rotated.send_delete);
assert!(
rotated
.playback_ended
.as_ref()
.is_some_and(|ended| ended.reconnecting)
);
let replacement = rotated
.connect
.expect("new stream server starts a replacement connection");
assert_eq!(replacement.endpoint, "replacement.example.com");
assert_eq!(replacement.token, "replacement-token");
let unavailable = state.apply(&VoiceRuntimeEvent::StreamServer(StreamServerInfo {
stream_key: initial.request.stream_key.clone(),
endpoint: None,
token: "pending-token".to_owned(),
}));
assert_eq!(
unavailable.close_stream_key.as_deref(),
Some(initial.request.stream_key.as_str())
);
assert!(!unavailable.send_delete);
assert!(unavailable.connect.is_none());
let reallocated = state.apply(&VoiceRuntimeEvent::StreamServer(StreamServerInfo {
stream_key: initial.request.stream_key.clone(),
endpoint: Some("reallocated.example.com".to_owned()),
token: "reallocated-token".to_owned(),
}));
let active = reallocated
.connect
.expect("reallocated stream server reconnects");
assert_ne!(active.connection_id, replacement.connection_id);
let stale_end = state.apply(&stream_connection_ended(
&replacement,
VoiceConnectionEnd::Stop,
));
assert!(stale_end.close_stream_key.is_none());
assert!(stale_end.connect.is_none());
assert_eq!(
state
.active
.as_ref()
.expect("reallocated connection remains active")
.connection_id,
active.connection_id
);
}
#[test]
fn stream_runtime_stops_after_consecutive_pre_playback_failures() {
let (mut state, mut active) = connected_stream_runtime();
for attempt in 1..=MAX_VOICE_RECONNECT_ATTEMPTS {
let update = state.apply(&stream_connection_ended(
&active,
VoiceConnectionEnd::Reconnect,
));
assert!(
update.close_stream_key.is_none(),
"retry {attempt} should keep the watch request active"
);
active = update
.connect
.expect("retry within the limit should reconnect");
}
let stopped = state.apply(&stream_connection_ended(
&active,
VoiceConnectionEnd::Reconnect,
));
assert!(stopped.connect.is_none());
assert_eq!(
stopped.close_stream_key.as_deref(),
Some(active.request.stream_key.as_str())
);
assert!(stopped.send_delete);
}
#[test]
fn stream_runtime_stops_immediately_after_terminal_failure() {
let (mut state, active) = connected_stream_runtime();
let stopped = state.apply(&stream_connection_ended(&active, VoiceConnectionEnd::Stop));
assert!(stopped.connect.is_none());
assert_eq!(
stopped.close_stream_key.as_deref(),
Some(active.request.stream_key.as_str())
);
assert!(stopped.send_delete);
}
#[test]
fn stream_runtime_resets_retries_only_after_stable_playback() {
let (mut state, initial) = connected_stream_runtime();
let first_retry = state.apply(&stream_connection_ended(
&initial,
VoiceConnectionEnd::Reconnect,
));
let mut active = first_retry
.connect
.expect("the first transport failure should reconnect");
state.apply(&VoiceRuntimeEvent::StreamConnectionEstablished {
connection_id: active.connection_id,
stream_key: active.request.stream_key.clone(),
});
for _ in 0..MAX_VOICE_RECONNECT_ATTEMPTS {
active = state
.apply(&stream_connection_ended(
&active,
VoiceConnectionEnd::Reconnect,
))
.connect
.expect("stable playback should restore the full retry budget");
}
}
#[test]
fn stream_failures_distinguish_player_and_transport_errors() {
for error in [
std::io::Error::new(std::io::ErrorKind::NotFound, "missing"),
std::io::Error::new(std::io::ErrorKind::PermissionDenied, "not executable"),
] {
let failure = stream_player_spawn_failure(error);
assert_eq!(failure.outcome, VoiceConnectionEnd::Stop);
}
assert_eq!(
StreamConnectionFailure::from("stream UDP receive failed".to_owned()).outcome,
VoiceConnectionEnd::Reconnect
);
}
#[test]
fn stream_gateway_payloads_request_stream_audio_and_h264() {
let session = StreamGatewaySession {
connection_id: 1,
request: stream_request(),
current_user_id: Id::new(5),
session_id: "parent-session".to_owned(),
rtc_server_id: "400".to_owned(),
rtc_channel_id: Id::new(401),
endpoint: "stream.example.com".to_owned(),
token: "stream-token".to_owned(),
reconnect_delay: Duration::ZERO,
};
let identify: Value =
serde_json::from_str(&stream_identify_payload(&session)).expect("valid identify json");
assert_eq!(identify["d"]["server_id"], "400");
assert_eq!(identify["d"]["channel_id"], "401");
assert_eq!(identify["d"]["video"], true);
assert!(identify["d"].get("streams").is_none());
let selected: Value = serde_json::from_str(&stream_select_protocol_payload(
&DiscoveredVoiceAddress {
address: "127.0.0.1".to_owned(),
port: 5000,
},
AEAD_XCHACHA20_POLY1305_RTPSIZE,
))
.expect("valid select protocol json");
assert_eq!(selected["d"]["codecs"][0]["name"], "opus");
assert_eq!(selected["d"]["codecs"][0]["payload_type"], 120);
assert_eq!(selected["d"]["codecs"][0]["encode"], false);
assert_eq!(selected["d"]["codecs"][0]["decode"], true);
assert_eq!(selected["d"]["codecs"][1]["name"], "H264");
assert_eq!(selected["d"]["codecs"][1]["payload_type"], 103);
assert_eq!(selected["d"]["codecs"][1]["decode"], true);
let wants: Value = serde_json::from_str(&stream_media_sink_wants_payload(800, 900))
.expect("valid media sink wants json");
assert_eq!(
wants,
json!({"op": 15, "d": {"800": 100, "900": 100, "any": 0}})
);
}
#[test]
fn stream_player_controls_the_complete_broadcast() {
let player = stream_player_command(Path::new("/tmp/concord-stream.sdp"), "neo")
.as_std()
.get_args()
.map(|argument| argument.to_string_lossy().into_owned())
.collect::<Vec<_>>();
assert_eq!(
player,
vec![
"--no-config",
"--terminal=yes",
"--load-scripts=no",
"--msg-level=all=warn,cplayer=v,lavf=v,vd=v,ad=v",
"--profile=low-latency",
"--cache=no",
"--demuxer-readahead-secs=0",
"--demuxer=lavf",
"--demuxer-lavf-format=sdp",
"--demuxer-lavf-probesize=32",
"--demuxer-lavf-buffersize=4096",
"--demuxer-lavf-o=protocol_whitelist=[file,udp,rtp],max_delay=0,reorder_queue_size=0",
"--force-window=immediate",
"--title=neo's stream",
"--video-sync=desync",
"--framedrop=vo",
"--video-timing-offset=0",
"--",
"/tmp/concord-stream.sdp",
]
);
}
#[test]
fn local_rtp_clocks_share_live_time_without_source_clock_offsets() {
let mut audio = LocalStreamAudioClock::default();
let mut video = LocalRtpClock::default();
assert_eq!(
audio
.rebase(3_000_000, Duration::from_millis(100))
.local_timestamp,
4_800
);
assert_eq!(
video.rebase(90_000_000, Duration::from_millis(125), VIDEO_RTP_CLOCK_RATE),
11_250
);
assert_eq!(
audio
.rebase(3_000_960, Duration::from_millis(120))
.local_timestamp,
5_760
);
assert_eq!(
video.rebase(90_003_000, Duration::from_millis(158), VIDEO_RTP_CLOCK_RATE),
14_250
);
}
#[test]
fn stream_audio_clock_reanchors_a_backward_source_timestamp() {
let mut clock = LocalStreamAudioClock::default();
let first = clock.rebase(3_181_287_385, Duration::ZERO);
assert_eq!(first.local_timestamp, 0);
assert!(first.discontinuity.is_none());
let before_reset = clock.rebase(3_181_475_545, Duration::from_millis(3_920));
assert_eq!(before_reset.local_timestamp, 188_160);
assert!(before_reset.discontinuity.is_none());
let reset = clock.rebase(3_181_114_584, Duration::from_millis(3_940));
assert_eq!(reset.local_timestamp, 189_120);
assert_eq!(
reset.discontinuity,
Some(StreamAudioClockDiscontinuity {
source_delta_ticks: -360_961,
elapsed_delta_ticks: 960,
})
);
let after_reset = clock.rebase(3_181_115_544, Duration::from_millis(3_960));
assert_eq!(after_reset.local_timestamp, 190_080);
assert!(after_reset.discontinuity.is_none());
assert_eq!(
clock.timestamp_at(Duration::from_millis(3_980)),
Some(191_040)
);
}
#[test]
fn stream_audio_clock_preserves_a_source_timestamp_wrap() {
let mut clock = LocalStreamAudioClock::default();
let before_wrap = clock.rebase(u32::MAX - 479, Duration::from_millis(100));
let after_wrap = clock.rebase(480, Duration::from_millis(120));
assert_eq!(before_wrap.local_timestamp, 4_800);
assert_eq!(after_wrap.local_timestamp, 5_760);
assert!(after_wrap.discontinuity.is_none());
}
}