Driver/Input: Migrate audio backend to Symphonia (#89)

This extensive PR rewrites the internal mixing logic of the driver to use symphonia for parsing and decoding audio data, and rubato to resample audio. Existing logic to decode DCA and Opus formats/data have been reworked as plugins for symphonia. The main benefit is that we no longer need to keep yt-dlp and ffmpeg processes alive, saving a lot of memory and CPU: all decoding can be done in Rust! In exchange, we now need to do a lot of the HTTP handling and resumption ourselves, but this is still a huge net positive.

`Input`s have been completely reworked such that all default (non-cached) sources are lazy by default, and are no longer covered by a special-case `Restartable`. These now span a gamut from a `Compose` (lazy), to a live source, to a fully `Parsed` source. As mixing is still sync, this includes adapters for `AsyncRead`/`AsyncSeek`, and HTTP streams.

`Track`s have been reworked so that they only contain initialisation state for each track. `TrackHandles` are only created once a `Track`/`Input` has been handed over to the driver, replacing `create_player` and related functions. `TrackHandle::action` now acts on a `View` of (im)mutable state, and can request seeks/readying via `Action`.

Per-track event handling has also been improved -- we can now determine and propagate the reason behind individual track errors due to the new backend. Some `TrackHandle` commands (seek etc.) benefit from this, and now use internal callbacks to signal completion.

Due to associated PRs on felixmcfelix/songbird from avid testers, this includes general clippy tweaks, API additions, and other repo-wide cleanup. Thanks go out to the below co-authors.

Co-authored-by: Gnome! <45660393+GnomedDev@users.noreply.github.com>
Co-authored-by: Alakh <36898190+alakhpc@users.noreply.github.com>
This commit is contained in:
Kyle Simpson
2022-07-23 23:29:02 +01:00
parent 6c6ffa7ca8
commit 8cc7a22b0b
136 changed files with 9761 additions and 4891 deletions

View File

@@ -1,8 +1,19 @@
//! Various driver internals which need to be exported for benchmarking.
//! Various driver internals which need to be exported for benchmarking.
//!
//! Included if using the `"internals"` feature flag.
//! You should not and/or cannot use these as part of a normal application.
#![allow(missing_docs)]
pub use super::tasks::{message as task_message, mixer};
pub use super::crypto::CryptoState;
use crate::{
driver::tasks::message::TrackContext,
tracks::{Track, TrackHandle},
};
pub fn track_context(t: Track) -> (TrackHandle, TrackContext) {
t.into_context()
}

View File

@@ -94,21 +94,21 @@ impl From<Elapsed> for Error {
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "failed to connect to Discord RTP server: ")?;
use Error::*;
match self {
AttemptDiscarded => write!(f, "connection attempt was aborted/discarded"),
Crypto(e) => e.fmt(f),
CryptoModeInvalid => write!(f, "server changed negotiated encryption mode"),
CryptoModeUnavailable => write!(f, "server did not offer chosen encryption mode"),
EndpointUrl => write!(f, "endpoint URL received from gateway was invalid"),
ExpectedHandshake => write!(f, "voice initialisation protocol was violated"),
IllegalDiscoveryResponse => write!(f, "IP discovery/NAT punching response was invalid"),
IllegalIp => write!(f, "IP discovery/NAT punching response had bad IP value"),
Io(e) => e.fmt(f),
Json(e) => e.fmt(f),
InterconnectFailure(e) => write!(f, "failed to contact other task ({:?})", e),
Ws(e) => write!(f, "websocket issue ({:?}).", e),
TimedOut => write!(f, "connection attempt timed out"),
Self::AttemptDiscarded => write!(f, "connection attempt was aborted/discarded"),
Self::Crypto(e) => e.fmt(f),
Self::CryptoModeInvalid => write!(f, "server changed negotiated encryption mode"),
Self::CryptoModeUnavailable => write!(f, "server did not offer chosen encryption mode"),
Self::EndpointUrl => write!(f, "endpoint URL received from gateway was invalid"),
Self::ExpectedHandshake => write!(f, "voice initialisation protocol was violated"),
Self::IllegalDiscoveryResponse =>
write!(f, "IP discovery/NAT punching response was invalid"),
Self::IllegalIp => write!(f, "IP discovery/NAT punching response had bad IP value"),
Self::Io(e) => e.fmt(f),
Self::Json(e) => e.fmt(f),
Self::InterconnectFailure(e) => write!(f, "failed to contact other task ({:?})", e),
Self::Ws(e) => write!(f, "websocket issue ({:?}).", e),
Self::TimedOut => write!(f, "connection attempt timed out"),
}
}
}
@@ -116,19 +116,19 @@ impl fmt::Display for Error {
impl StdError for Error {
fn source(&self) -> Option<&(dyn StdError + 'static)> {
match self {
Error::AttemptDiscarded => None,
Error::AttemptDiscarded
| Error::CryptoModeInvalid
| Error::CryptoModeUnavailable
| Error::EndpointUrl
| Error::ExpectedHandshake
| Error::IllegalDiscoveryResponse
| Error::IllegalIp
| Error::InterconnectFailure(_)
| Error::Ws(_)
| Error::TimedOut => None,
Error::Crypto(e) => e.source(),
Error::CryptoModeInvalid => None,
Error::CryptoModeUnavailable => None,
Error::EndpointUrl => None,
Error::ExpectedHandshake => None,
Error::IllegalDiscoveryResponse => None,
Error::IllegalIp => None,
Error::Io(e) => e.source(),
Error::Json(e) => e.source(),
Error::InterconnectFailure(_) => None,
Error::Ws(_) => None,
Error::TimedOut => None,
}
}
}

View File

@@ -12,7 +12,7 @@ use xsalsa20poly1305::{
TAG_SIZE,
};
/// Variants of the XSalsa20Poly1305 encryption scheme.
/// Variants of the `XSalsa20Poly1305` encryption scheme.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum CryptoMode {
@@ -35,57 +35,58 @@ pub enum CryptoMode {
impl From<CryptoState> for CryptoMode {
fn from(val: CryptoState) -> Self {
use CryptoState::*;
match val {
Normal => CryptoMode::Normal,
Suffix => CryptoMode::Suffix,
Lite(_) => CryptoMode::Lite,
CryptoState::Normal => Self::Normal,
CryptoState::Suffix => Self::Suffix,
CryptoState::Lite(_) => Self::Lite,
}
}
}
impl CryptoMode {
/// Returns the name of a mode as it will appear during negotiation.
#[must_use]
pub fn to_request_str(self) -> &'static str {
use CryptoMode::*;
match self {
Normal => "xsalsa20_poly1305",
Suffix => "xsalsa20_poly1305_suffix",
Lite => "xsalsa20_poly1305_lite",
Self::Normal => "xsalsa20_poly1305",
Self::Suffix => "xsalsa20_poly1305_suffix",
Self::Lite => "xsalsa20_poly1305_lite",
}
}
/// Returns the number of bytes each nonce is stored as within
/// a packet.
#[must_use]
pub fn nonce_size(self) -> usize {
use CryptoMode::*;
match self {
Normal => RtpPacket::minimum_packet_size(),
Suffix => NONCE_SIZE,
Lite => 4,
Self::Normal => RtpPacket::minimum_packet_size(),
Self::Suffix => NONCE_SIZE,
Self::Lite => 4,
}
}
/// Returns the number of bytes occupied by the encryption scheme
/// which fall before the payload.
pub fn payload_prefix_len(self) -> usize {
#[must_use]
pub fn payload_prefix_len() -> usize {
TAG_SIZE
}
/// Returns the number of bytes occupied by the encryption scheme
/// which fall after the payload.
#[must_use]
pub fn payload_suffix_len(self) -> usize {
use CryptoMode::*;
match self {
Normal => 0,
Suffix | Lite => self.nonce_size(),
Self::Normal => 0,
Self::Suffix | Self::Lite => self.nonce_size(),
}
}
/// Calculates the number of additional bytes required compared
/// to an unencrypted payload.
#[must_use]
pub fn payload_overhead(self) -> usize {
self.payload_prefix_len() + self.payload_suffix_len()
Self::payload_prefix_len() + self.payload_suffix_len()
}
/// Extracts the byte slice in a packet used as the nonce, and the remaining mutable
@@ -95,10 +96,9 @@ impl CryptoMode {
header: &'a [u8],
body: &'a mut [u8],
) -> Result<(&'a [u8], &'a mut [u8]), CryptoError> {
use CryptoMode::*;
match self {
Normal => Ok((header, body)),
Suffix | Lite => {
Self::Normal => Ok((header, body)),
Self::Suffix | Self::Lite => {
let len = body.len();
if len < self.payload_suffix_len() {
Err(CryptoError)
@@ -135,7 +135,7 @@ impl CryptoMode {
&nonce
};
let body_start = self.payload_prefix_len();
let body_start = Self::payload_prefix_len();
let body_tail = self.payload_suffix_len();
if body_start > body_remaining.len() {
@@ -183,22 +183,33 @@ impl CryptoMode {
}
}
#[allow(missing_docs)]
/// State used in nonce generation for the `XSalsa20Poly1305` encryption variants
/// in [`CryptoMode`].
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum CryptoState {
/// The RTP header is used as the source of nonce bytes for the packet.
///
/// No state is required.
Normal,
/// An additional random 24B suffix is used as the source of nonce bytes for the packet.
/// This is regenerated randomly for each packet.
///
/// No state is required.
Suffix,
/// An additional random 4B suffix is used as the source of nonce bytes for the packet.
/// This nonce value increments by `1` with each packet.
///
/// The last used nonce is stored.
Lite(Wrapping<u32>),
}
impl From<CryptoMode> for CryptoState {
fn from(val: CryptoMode) -> Self {
use CryptoMode::*;
match val {
Normal => CryptoState::Normal,
Suffix => CryptoState::Suffix,
Lite => CryptoState::Lite(Wrapping(rand::random::<u32>())),
CryptoMode::Normal => CryptoState::Normal,
CryptoMode::Suffix => CryptoState::Suffix,
CryptoMode::Lite => CryptoState::Lite(Wrapping(rand::random::<u32>())),
}
}
}
@@ -213,12 +224,11 @@ impl CryptoState {
let mode = self.kind();
let endpoint = payload_end + mode.payload_suffix_len();
use CryptoState::*;
match self {
Suffix => {
Self::Suffix => {
rand::thread_rng().fill(&mut packet.payload_mut()[payload_end..endpoint]);
},
Lite(mut i) => {
Self::Lite(mut i) => {
(&mut packet.payload_mut()[payload_end..endpoint])
.write_u32::<NetworkEndian>(i.0)
.expect(
@@ -233,8 +243,8 @@ impl CryptoState {
}
/// Returns the underlying (stateless) type of the active crypto mode.
pub fn kind(&self) -> CryptoMode {
CryptoMode::from(*self)
pub fn kind(self) -> CryptoMode {
CryptoMode::from(self)
}
}
@@ -246,7 +256,7 @@ mod test {
#[test]
fn small_packet_decrypts_error() {
let mut buf = [0u8; MutableRtpPacket::minimum_packet_size() + 0];
let mut buf = [0u8; MutableRtpPacket::minimum_packet_size()];
let modes = [CryptoMode::Normal, CryptoMode::Suffix, CryptoMode::Lite];
let mut pkt = MutableRtpPacket::new(&mut buf[..]).unwrap();

View File

@@ -26,6 +26,7 @@ pub enum DecodeMode {
impl DecodeMode {
/// Returns whether this mode will decrypt received packets.
#[must_use]
pub fn should_decrypt(self) -> bool {
self != DecodeMode::Pass
}

59
src/driver/mix_mode.rs Normal file
View File

@@ -0,0 +1,59 @@
use audiopus::Channels;
use symphonia_core::audio::Layout;
use crate::constants::{MONO_FRAME_SIZE, STEREO_FRAME_SIZE};
/// Mixing behaviour for sent audio sources processed within the driver.
///
/// This has no impact on Opus packet passthrough, which will pass packets
/// irrespective of their channel count.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum MixMode {
/// Audio sources will be downmixed into a mono buffer.
Mono,
/// Audio sources will be mixed into into a stereo buffer, where mono sources
/// will be duplicated into both channels.
Stereo,
}
impl MixMode {
pub(crate) const fn to_opus(self) -> Channels {
match self {
Self::Mono => Channels::Mono,
Self::Stereo => Channels::Stereo,
}
}
pub(crate) const fn sample_count_in_frame(self) -> usize {
match self {
Self::Mono => MONO_FRAME_SIZE,
Self::Stereo => STEREO_FRAME_SIZE,
}
}
pub(crate) const fn channels(self) -> usize {
match self {
Self::Mono => 1,
Self::Stereo => 2,
}
}
pub(crate) const fn symph_layout(self) -> Layout {
match self {
Self::Mono => Layout::Mono,
Self::Stereo => Layout::Stereo,
}
}
}
impl From<MixMode> for Layout {
fn from(val: MixMode) -> Self {
val.symph_layout()
}
}
impl From<MixMode> for Channels {
fn from(val: MixMode) -> Self {
val.to_opus()
}
}

View File

@@ -14,20 +14,26 @@ pub mod bench_internals;
pub(crate) mod connection;
mod crypto;
mod decode_mode;
mod mix_mode;
pub mod retry;
pub(crate) mod tasks;
#[cfg(test)]
pub(crate) mod test_config;
use connection::error::{Error, Result};
pub use crypto::CryptoMode;
pub(crate) use crypto::CryptoState;
pub use decode_mode::DecodeMode;
pub use mix_mode::MixMode;
#[cfg(test)]
pub use test_config::*;
#[cfg(feature = "builtin-queue")]
use crate::tracks::TrackQueue;
use crate::{
events::EventData,
input::Input,
tracks::{self, Track, TrackHandle},
tracks::{Track, TrackHandle},
Config,
ConnectionInfo,
Event,
@@ -41,6 +47,8 @@ use core::{
task::{Context, Poll},
};
use flume::{r#async::RecvFut, SendError, Sender};
#[cfg(feature = "builtin-queue")]
use std::time::Duration;
use tasks::message::CoreMessage;
use tracing::instrument;
@@ -54,8 +62,13 @@ pub struct Driver {
config: Config,
self_mute: bool,
sender: Sender<CoreMessage>,
// Making this an Option is an abhorrent hack to coerce the borrow checker
// into letting us have an &TrackQueue at the same time as an &mut Driver.
// This is probably preferable to cloning the driver: Arc<...> should be nonzero
// and if the compiler's smart we'll just codegen a pointer swap. It definitely makes
// use of NonZero.
#[cfg(feature = "builtin-queue")]
queue: TrackQueue,
queue: Option<TrackQueue>,
}
impl Driver {
@@ -63,6 +76,7 @@ impl Driver {
///
/// This will create the core voice tasks in the background.
#[inline]
#[must_use]
pub fn new(config: Config) -> Self {
let sender = Self::start_inner(config.clone());
@@ -71,7 +85,7 @@ impl Driver {
self_mute: false,
sender,
#[cfg(feature = "builtin-queue")]
queue: Default::default(),
queue: Some(TrackQueue::default()),
}
}
@@ -136,63 +150,45 @@ impl Driver {
self.self_mute
}
/// Plays audio from a source, returning a handle for further control.
///
/// This can be a source created via [`ffmpeg`] or [`ytdl`].
///
/// [`ffmpeg`]: crate::input::ffmpeg
/// [`ytdl`]: crate::input::ytdl
#[instrument(skip(self))]
pub fn play_source(&mut self, source: Input) -> TrackHandle {
let (player, handle) = super::create_player(source);
self.send(CoreMessage::AddTrack(player));
handle
/// Plays audio from an input, returning a handle for further control.
#[instrument(skip(self, input))]
pub fn play_input(&mut self, input: Input) -> TrackHandle {
self.play(input.into())
}
/// Plays audio from a source, returning a handle for further control.
/// Plays audio from an input, returning a handle for further control.
///
/// Unlike [`play_source`], this stops all other sources attached
/// Unlike [`Self::play_input`], this stops all other inputs attached
/// to the channel.
///
/// [`play_source`]: Driver::play_source
#[instrument(skip(self))]
pub fn play_only_source(&mut self, source: Input) -> TrackHandle {
let (player, handle) = super::create_player(source);
self.send(CoreMessage::SetTrack(Some(player)));
handle
#[instrument(skip(self, input))]
pub fn play_only_input(&mut self, input: Input) -> TrackHandle {
self.play_only(input.into())
}
/// Plays audio from a [`Track`] object.
///
/// This will be one half of the return value of [`create_player`].
/// The main difference between this function and [`play_source`] is
/// The main difference between this function and [`Self::play_input`] is
/// that this allows for direct manipulation of the [`Track`] object
/// before it is passed over to the voice and mixing contexts.
///
/// [`create_player`]: crate::tracks::create_player
/// [`create_player`]: crate::tracks::Track
/// [`play_source`]: Driver::play_source
#[instrument(skip(self))]
pub fn play(&mut self, track: Track) {
self.send(CoreMessage::AddTrack(track));
#[instrument(skip(self, track))]
pub fn play(&mut self, track: Track) -> TrackHandle {
let (handle, ctx) = track.into_context();
self.send(CoreMessage::AddTrack(ctx));
handle
}
/// Exclusively plays audio from a [`Track`] object.
///
/// This will be one half of the return value of [`create_player`].
/// As in [`play_only_source`], this stops all other sources attached to the
/// channel. Like [`play`], however, this allows for direct manipulation of the
/// As in [`Self::play_only_input`], this stops all other sources attached to the
/// channel. Like [`Self::play`], however, this allows for direct manipulation of the
/// [`Track`] object before it is passed over to the voice and mixing contexts.
///
/// [`create_player`]: crate::tracks::create_player
/// [`Track`]: crate::tracks::Track
/// [`play_only_source`]: Driver::play_only_source
/// [`play`]: Driver::play
#[instrument(skip(self))]
pub fn play_only(&mut self, track: Track) {
self.send(CoreMessage::SetTrack(Some(track)));
#[instrument(skip(self, track))]
pub fn play_only(&mut self, track: Track) -> TrackHandle {
let (handle, ctx) = track.into_context();
self.send(CoreMessage::SetTrack(Some(ctx)));
handle
}
/// Sets the bitrate for encoding Opus packets sent along
@@ -204,20 +200,20 @@ impl Driver {
/// Alternatively, `Auto` and `Max` remain available.
#[instrument(skip(self))]
pub fn set_bitrate(&mut self, bitrate: Bitrate) {
self.send(CoreMessage::SetBitrate(bitrate))
self.send(CoreMessage::SetBitrate(bitrate));
}
/// Stops playing audio from all sources, if any are set.
#[instrument(skip(self))]
pub fn stop(&mut self) {
self.send(CoreMessage::SetTrack(None))
self.send(CoreMessage::SetTrack(None));
}
/// Sets the configuration for this driver (and parent `Call`, if applicable).
#[instrument(skip(self))]
pub fn set_config(&mut self, config: Config) {
self.config = config.clone();
self.send(CoreMessage::SetConfig(config))
self.send(CoreMessage::SetConfig(config));
}
/// Returns a view of this driver's configuration.
@@ -237,7 +233,6 @@ impl Driver {
/// within the supplied function or closure. *Taking excess time could prevent
/// timely sending of packets, causing audio glitches and delays*.
///
/// [`Track`]: crate::tracks::Track
/// [`TrackEvent`]: crate::events::TrackEvent
/// [`EventContext`]: crate::events::EventContext
#[instrument(skip(self, action))]
@@ -267,41 +262,53 @@ impl Driver {
/// Returns a reference to this driver's built-in queue.
///
/// Requires the `"builtin-queue"` feature.
/// Queue additions should be made via [`enqueue`] and
/// [`enqueue_source`].
///
/// [`enqueue`]: Driver::enqueue
/// [`enqueue_source`]: Driver::enqueue_source
/// Queue additions should be made via [`Driver::enqueue`] and
/// [`Driver::enqueue_input`].
#[must_use]
pub fn queue(&self) -> &TrackQueue {
&self.queue
self.queue
.as_ref()
.expect("Queue: The only case this can fail is if a previous queue operation panicked.")
}
/// Adds an audio [`Input`] to this driver's built-in queue.
///
/// Requires the `"builtin-queue"` feature.
///
/// [`Input`]: crate::input::Input
pub fn enqueue_source(&mut self, source: Input) -> TrackHandle {
let (track, handle) = tracks::create_player(source);
self.enqueue(track);
handle
pub async fn enqueue_input(&mut self, input: Input) -> TrackHandle {
self.enqueue(input.into()).await
}
/// Adds an existing [`Track`] to this driver's built-in queue.
///
/// Requires the `"builtin-queue"` feature.
pub async fn enqueue(&mut self, mut track: Track) -> TrackHandle {
let preload_time = TrackQueue::get_preload_time(&mut track).await;
self.enqueue_with_preload(track, preload_time)
}
/// Add an existing [`Track`] to the queue, using a known time to preload the next track.
///
/// [`Track`]: crate::tracks::Track
pub fn enqueue(&mut self, mut track: Track) {
self.queue.add_raw(&mut track);
self.play(track);
/// See [`TrackQueue::add_with_preload`] for how `preload_time` is used.
///
/// Requires the `"builtin-queue"` feature.
pub fn enqueue_with_preload(
&mut self,
track: Track,
preload_time: Option<Duration>,
) -> TrackHandle {
let queue = self.queue.take().expect(
"Enqueue: The only case this can fail is if a previous queue operation panicked.",
);
let handle = queue.add_with_preload(track, self, preload_time);
self.queue = Some(queue);
handle
}
}
impl Default for Driver {
fn default() -> Self {
Self::new(Default::default())
Self::new(Config::default())
}
}
@@ -309,7 +316,7 @@ impl Drop for Driver {
/// Leaves the current connected voice channel, if connected to one, and
/// forgets all configurations relevant to this Handler.
fn drop(&mut self) {
let _ = self.sender.send(CoreMessage::Poison);
drop(self.sender.send(CoreMessage::Poison));
}
}
@@ -317,8 +324,6 @@ impl Drop for Driver {
///
/// This future awaits the *result* of a connection; the driver
/// is messaged at the time of the call.
///
/// [`Driver::connect`]: Driver::connect
pub struct Connect {
inner: RecvFut<'static, Result<()>>,
}

View File

@@ -28,7 +28,7 @@ pub struct Retry {
impl Default for Retry {
fn default() -> Self {
Self {
strategy: Strategy::Backoff(Default::default()),
strategy: Strategy::Backoff(ExponentialBackoff::default()),
retry_limit: Some(5),
}
}
@@ -40,7 +40,7 @@ impl Retry {
last_wait: Option<Duration>,
attempts: usize,
) -> Option<Duration> {
if self.retry_limit.map(|a| attempts < a).unwrap_or(true) {
if self.retry_limit.map_or(true, |a| attempts < a) {
Some(self.strategy.retry_in(last_wait))
} else {
None

View File

@@ -58,7 +58,7 @@ impl Default for ExponentialBackoff {
impl ExponentialBackoff {
pub(crate) fn retry_in(&self, last_wait: Option<Duration>) -> Duration {
let attempt = last_wait.map(|t| 2 * t).unwrap_or(self.min);
let attempt = last_wait.map_or(self.min, |t| 2 * t);
let perturb = (1.0 - (self.jitter * 2.0 * (random::<f32>() - 1.0)))
.max(0.0)
.min(2.0);

View File

@@ -9,10 +9,5 @@ use tracing::instrument;
/// to prevent deadline misses.
#[instrument(skip(mix_rx))]
pub(crate) fn runner(mix_rx: Receiver<DisposalMessage>) {
loop {
match mix_rx.recv() {
Err(_) | Ok(DisposalMessage::Poison) => break,
_ => {},
}
}
while mix_rx.recv().is_ok() {}
}

View File

@@ -32,9 +32,7 @@ impl Error {
pub(crate) fn should_trigger_connect(&self) -> bool {
matches!(
self,
Error::InterconnectFailure(Recipient::AuxNetwork)
| Error::InterconnectFailure(Recipient::UdpRx)
| Error::InterconnectFailure(Recipient::UdpTx)
Error::InterconnectFailure(Recipient::AuxNetwork | Recipient::UdpRx | Recipient::UdpTx)
)
}

View File

@@ -1,7 +1,7 @@
use super::message::*;
use crate::{
events::{EventStore, GlobalEvents, TrackEvent},
tracks::{TrackHandle, TrackState},
tracks::{ReadyState, TrackHandle, TrackState},
};
use flume::Receiver;
use tracing::{debug, info, instrument, trace};
@@ -14,14 +14,13 @@ pub(crate) async fn runner(_interconnect: Interconnect, evt_rx: Receiver<EventMe
let mut states: Vec<TrackState> = vec![];
let mut handles: Vec<TrackHandle> = vec![];
loop {
use EventMessage::*;
match evt_rx.recv_async().await {
Ok(AddGlobalEvent(data)) => {
while let Ok(msg) = evt_rx.recv_async().await {
match msg {
EventMessage::AddGlobalEvent(data) => {
info!("Global event added.");
global.add_event(data);
},
Ok(AddTrackEvent(i, data)) => {
EventMessage::AddTrackEvent(i, data) => {
info!("Adding event to track {}.", i);
let event_store = events
@@ -33,7 +32,7 @@ pub(crate) async fn runner(_interconnect: Interconnect, evt_rx: Receiver<EventMe
event_store.add_event(data, state.position);
},
Ok(FireCoreEvent(ctx)) => {
EventMessage::FireCoreEvent(ctx) => {
let ctx = ctx.to_user_context();
let evt = ctx
.to_core_event()
@@ -43,19 +42,17 @@ pub(crate) async fn runner(_interconnect: Interconnect, evt_rx: Receiver<EventMe
global.fire_core_event(evt, ctx).await;
},
Ok(RemoveGlobalEvents) => {
EventMessage::RemoveGlobalEvents => {
global.remove_handlers();
},
Ok(AddTrack(store, state, handle)) => {
EventMessage::AddTrack(store, state, handle) => {
events.push(store);
states.push(state);
handles.push(handle);
info!("Event state for track {} added", events.len());
},
Ok(ChangeState(i, change)) => {
use TrackStateChange::*;
EventMessage::ChangeState(i, change) => {
let max_states = states.len();
debug!(
"Changing state for track {} of {}: {:?}",
@@ -67,53 +64,74 @@ pub(crate) async fn runner(_interconnect: Interconnect, evt_rx: Receiver<EventMe
.expect("Event thread was given an illegal state index for ChangeState.");
match change {
Mode(mode) => {
let old = state.playing;
state.playing = mode;
if old != mode {
global.fire_track_event(mode.as_track_event(), i);
TrackStateChange::Mode(mut mode) => {
std::mem::swap(&mut state.playing, &mut mode);
if state.playing != mode {
global.fire_track_event(state.playing.as_track_event(), i);
if let Some(other_evts) = state.playing.also_fired_track_events() {
for evt in other_evts {
global.fire_track_event(evt, i);
}
}
}
},
Volume(vol) => {
TrackStateChange::Volume(vol) => {
state.volume = vol;
},
Position(pos) => {
TrackStateChange::Position(pos) => {
// Currently, only Tick should fire time events.
state.position = pos;
},
Loops(loops, user_set) => {
TrackStateChange::Loops(loops, user_set) => {
state.loops = loops;
if !user_set {
global.fire_track_event(TrackEvent::Loop, i);
}
},
Total(new) => {
TrackStateChange::Total(new) => {
// Massive, unprecedented state changes.
*state = new;
},
TrackStateChange::Ready(ready_state) => {
state.ready = ready_state;
match ready_state {
ReadyState::Playable => {
global.fire_track_event(TrackEvent::Playable, i);
},
ReadyState::Preparing => {
global.fire_track_event(TrackEvent::Preparing, i);
},
ReadyState::Uninitialised => {},
}
},
}
},
Ok(RemoveTrack(i)) => {
info!("Event state for track {} of {} removed.", i, events.len());
events.swap_remove(i);
states.swap_remove(i);
handles.swap_remove(i);
},
Ok(RemoveAllTracks) => {
EventMessage::RemoveAllTracks => {
info!("Event state for all tracks removed.");
events.clear();
states.clear();
handles.clear();
},
Ok(Tick) => {
EventMessage::Tick => {
// NOTE: this should fire saved up blocks of state change evts.
global.tick(&mut events, &mut states, &mut handles).await;
let mut i = 0;
while i < states.len() {
if states[i].playing.is_done() {
info!("Event state for track {} of {} removed.", i, events.len());
events.swap_remove(i);
states.swap_remove(i);
handles.swap_remove(i);
} else {
i += 1;
}
}
},
Err(_) | Ok(Poison) => {
break;
},
EventMessage::Poison => break,
}
}

View File

@@ -3,20 +3,18 @@
use crate::{
driver::{connection::error::Error, Bitrate, Config},
events::{context_data::DisconnectReason, EventData},
tracks::Track,
tracks::{Track, TrackCommand, TrackHandle},
ConnectionInfo,
};
use flume::Sender;
use flume::{Receiver, Sender};
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
pub enum CoreMessage {
ConnectWithResult(ConnectionInfo, Sender<Result<(), Error>>),
RetryConnect(usize),
SignalWsClosure(usize, ConnectionInfo, Option<DisconnectReason>),
Disconnect,
SetTrack(Option<Track>),
AddTrack(Track),
SetTrack(Option<TrackContext>),
AddTrack(TrackContext),
SetBitrate(Bitrate),
AddEvent(EventData),
RemoveGlobalEvents,
@@ -27,3 +25,9 @@ pub enum CoreMessage {
RebuildInterconnect,
Poison,
}
pub struct TrackContext {
pub track: Track,
pub handle: TrackHandle,
pub receiver: Receiver<TrackCommand>,
}

View File

@@ -1,9 +1,8 @@
#![allow(missing_docs)]
use crate::tracks::Track;
use crate::{driver::tasks::mixer::InternalTrack, tracks::TrackHandle};
pub enum DisposalMessage {
Track(Track),
Poison,
Track(Box<InternalTrack>),
Handle(TrackHandle),
}

View File

@@ -2,7 +2,7 @@
use crate::{
events::{CoreContext, EventData, EventStore},
tracks::{LoopState, PlayMode, TrackHandle, TrackState},
tracks::{LoopState, PlayMode, ReadyState, TrackHandle, TrackState},
};
use std::time::Duration;
@@ -16,7 +16,6 @@ pub enum EventMessage {
AddTrack(EventStore, TrackState, TrackHandle),
ChangeState(usize, TrackStateChange),
RemoveTrack(usize),
RemoveAllTracks,
Tick,
@@ -31,4 +30,5 @@ pub enum TrackStateChange {
// Bool indicates user-set.
Loops(LoopState, bool),
Total(TrackState),
Ready(ReadyState),
}

View File

@@ -1,12 +1,14 @@
#![allow(missing_docs)]
use super::{Interconnect, UdpRxMessage, UdpTxMessage, WsMessage};
use super::{Interconnect, TrackContext, UdpRxMessage, UdpTxMessage, WsMessage};
use crate::{
driver::{Bitrate, Config, CryptoState},
tracks::Track,
input::{AudioStreamError, Compose, Parsed},
};
use flume::Sender;
use std::sync::Arc;
use symphonia_core::{errors::Error as SymphoniaError, formats::SeekedTo};
use xsalsa20poly1305::XSalsa20Poly1305 as Cipher;
pub struct MixerConnection {
@@ -16,16 +18,9 @@ pub struct MixerConnection {
pub udp_tx: Sender<UdpTxMessage>,
}
impl Drop for MixerConnection {
fn drop(&mut self) {
let _ = self.udp_rx.send(UdpRxMessage::Poison);
let _ = self.udp_tx.send(UdpTxMessage::Poison);
}
}
pub enum MixerMessage {
AddTrack(Track),
SetTrack(Option<Track>),
AddTrack(TrackContext),
SetTrack(Option<TrackContext>),
SetBitrate(Bitrate),
SetConfig(Config),
@@ -40,3 +35,14 @@ pub enum MixerMessage {
Poison,
}
pub enum MixerInputResultMessage {
CreateErr(Arc<AudioStreamError>),
ParseErr(Arc<SymphoniaError>),
Seek(
Parsed,
Option<Box<dyn Compose>>,
Result<SeekedTo, Arc<SymphoniaError>>,
),
Built(Parsed, Option<Box<dyn Compose>>),
}

View File

@@ -23,11 +23,11 @@ pub struct Interconnect {
impl Interconnect {
pub fn poison(&self) {
let _ = self.events.send(EventMessage::Poison);
drop(self.events.send(EventMessage::Poison));
}
pub fn poison_all(&self) {
let _ = self.mixer.send(MixerMessage::Poison);
drop(self.mixer.send(MixerMessage::Poison));
self.poison();
}
@@ -46,8 +46,9 @@ impl Interconnect {
});
// Make mixer aware of new targets...
let _ = self
.mixer
.send(MixerMessage::ReplaceInterconnect(self.clone()));
drop(
self.mixer
.send(MixerMessage::ReplaceInterconnect(self.clone())),
);
}
}

View File

@@ -6,6 +6,4 @@ use crate::driver::Config;
pub enum UdpRxMessage {
SetConfig(Config),
ReplaceInterconnect(Interconnect),
Poison,
}

View File

@@ -1,6 +1,4 @@
#![allow(missing_docs)]
pub enum UdpTxMessage {
Packet(Vec<u8>), // TODO: do something cheaper.
Poison,
}
// TODO: do something cheaper.
pub type UdpTxMessage = Vec<u8>;

View File

@@ -3,12 +3,9 @@
use super::Interconnect;
use crate::ws::WsStream;
#[allow(dead_code)]
pub enum WsMessage {
Ws(Box<WsStream>),
ReplaceInterconnect(Interconnect),
SetKeepalive(f64),
Speaking(bool),
Poison,
}

View File

@@ -1,628 +0,0 @@
use super::{disposal, error::Result, message::*};
use crate::{
constants::*,
tracks::{PlayMode, Track},
Config,
};
use audiopus::{
coder::Encoder as OpusEncoder,
softclip::SoftClip,
Application as CodingMode,
Bitrate,
Channels,
};
use discortp::{
rtp::{MutableRtpPacket, RtpPacket},
MutablePacket,
};
use flume::{Receiver, Sender, TryRecvError};
use rand::random;
use std::{convert::TryInto, time::Instant};
use tokio::runtime::Handle;
use tracing::{debug, error, instrument};
use xsalsa20poly1305::TAG_SIZE;
pub struct Mixer {
pub async_handle: Handle,
pub bitrate: Bitrate,
pub config: Config,
pub conn_active: Option<MixerConnection>,
pub deadline: Instant,
pub disposer: Sender<DisposalMessage>,
pub encoder: OpusEncoder,
pub interconnect: Interconnect,
pub mix_rx: Receiver<MixerMessage>,
pub muted: bool,
pub packet: [u8; VOICE_PACKET_MAX],
pub prevent_events: bool,
pub silence_frames: u8,
pub skip_sleep: bool,
pub soft_clip: SoftClip,
pub tracks: Vec<Track>,
pub ws: Option<Sender<WsMessage>>,
}
fn new_encoder(bitrate: Bitrate) -> Result<OpusEncoder> {
let mut encoder = OpusEncoder::new(SAMPLE_RATE, Channels::Stereo, CodingMode::Audio)?;
encoder.set_bitrate(bitrate)?;
Ok(encoder)
}
impl Mixer {
pub fn new(
mix_rx: Receiver<MixerMessage>,
async_handle: Handle,
interconnect: Interconnect,
config: Config,
) -> Self {
let bitrate = DEFAULT_BITRATE;
let encoder = new_encoder(bitrate)
.expect("Failed to create encoder in mixing thread with known-good values.");
let soft_clip = SoftClip::new(Channels::Stereo);
let mut packet = [0u8; VOICE_PACKET_MAX];
let mut rtp = MutableRtpPacket::new(&mut packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
rtp.set_version(RTP_VERSION);
rtp.set_payload_type(RTP_PROFILE_TYPE);
rtp.set_sequence(random::<u16>().into());
rtp.set_timestamp(random::<u32>().into());
let tracks = Vec::with_capacity(1.max(config.preallocated_tracks));
// Create an object disposal thread here.
let (disposer, disposal_rx) = flume::unbounded();
std::thread::spawn(move || disposal::runner(disposal_rx));
Self {
async_handle,
bitrate,
config,
conn_active: None,
deadline: Instant::now(),
disposer,
encoder,
interconnect,
mix_rx,
muted: false,
packet,
prevent_events: false,
silence_frames: 0,
skip_sleep: false,
soft_clip,
tracks,
ws: None,
}
}
fn run(&mut self) {
let mut events_failure = false;
let mut conn_failure = false;
'runner: loop {
if self.conn_active.is_some() {
loop {
match self.mix_rx.try_recv() {
Ok(m) => {
let (events, conn, should_exit) = self.handle_message(m);
events_failure |= events;
conn_failure |= conn;
if should_exit {
break 'runner;
}
},
Err(TryRecvError::Disconnected) => {
break 'runner;
},
Err(TryRecvError::Empty) => {
break;
},
};
}
// The above action may have invalidated the connection; need to re-check!
if self.conn_active.is_some() {
if let Err(e) = self.cycle().and_then(|_| self.audio_commands_events()) {
events_failure |= e.should_trigger_interconnect_rebuild();
conn_failure |= e.should_trigger_connect();
debug!("Mixer thread cycle: {:?}", e);
}
}
} else {
match self.mix_rx.recv() {
Ok(m) => {
let (events, conn, should_exit) = self.handle_message(m);
events_failure |= events;
conn_failure |= conn;
if should_exit {
break 'runner;
}
},
Err(_) => {
break 'runner;
},
}
}
// event failure? rebuild interconnect.
// ws or udp failure? full connect
// (soft reconnect is covered by the ws task.)
//
// in both cases, send failure is fatal,
// but will only occur on disconnect.
// expecting this is fairly noisy, so exit silently.
if events_failure {
self.prevent_events = true;
let sent = self
.interconnect
.core
.send(CoreMessage::RebuildInterconnect);
events_failure = false;
if sent.is_err() {
break;
}
}
if conn_failure {
self.conn_active = None;
let sent = self.interconnect.core.send(CoreMessage::FullReconnect);
conn_failure = false;
if sent.is_err() {
break;
}
}
}
}
#[inline]
fn handle_message(&mut self, msg: MixerMessage) -> (bool, bool, bool) {
let mut events_failure = false;
let mut conn_failure = false;
let mut should_exit = false;
use MixerMessage::*;
let error = match msg {
AddTrack(mut t) => {
t.source.prep_with_handle(self.async_handle.clone());
self.add_track(t)
},
SetTrack(t) => {
self.tracks.clear();
let mut out = self.fire_event(EventMessage::RemoveAllTracks);
if let Some(mut t) = t {
t.source.prep_with_handle(self.async_handle.clone());
// Do this unconditionally: this affects local state infallibly,
// with the event installation being the remote part.
if let Err(e) = self.add_track(t) {
out = Err(e);
}
}
out
},
SetBitrate(b) => {
self.bitrate = b;
if let Err(e) = self.set_bitrate(b) {
error!("Failed to update bitrate {:?}", e);
}
Ok(())
},
SetMute(m) => {
self.muted = m;
Ok(())
},
SetConn(conn, ssrc) => {
self.conn_active = Some(conn);
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
rtp.set_ssrc(ssrc);
rtp.set_sequence(random::<u16>().into());
rtp.set_timestamp(random::<u32>().into());
self.deadline = Instant::now();
Ok(())
},
DropConn => {
self.conn_active = None;
Ok(())
},
ReplaceInterconnect(i) => {
self.prevent_events = false;
if let Some(ws) = &self.ws {
conn_failure |= ws.send(WsMessage::ReplaceInterconnect(i.clone())).is_err();
}
if let Some(conn) = &self.conn_active {
conn_failure |= conn
.udp_rx
.send(UdpRxMessage::ReplaceInterconnect(i.clone()))
.is_err();
}
self.interconnect = i;
self.rebuild_tracks()
},
SetConfig(new_config) => {
self.config = new_config.clone();
if self.tracks.capacity() < self.config.preallocated_tracks {
self.tracks
.reserve(self.config.preallocated_tracks - self.tracks.len());
}
if let Some(conn) = &self.conn_active {
conn_failure |= conn
.udp_rx
.send(UdpRxMessage::SetConfig(new_config))
.is_err();
}
Ok(())
},
RebuildEncoder => match new_encoder(self.bitrate) {
Ok(encoder) => {
self.encoder = encoder;
Ok(())
},
Err(e) => {
error!("Failed to rebuild encoder. Resetting bitrate. {:?}", e);
self.bitrate = DEFAULT_BITRATE;
self.encoder = new_encoder(self.bitrate)
.expect("Failed fallback rebuild of OpusEncoder with safe inputs.");
Ok(())
},
},
Ws(new_ws_handle) => {
self.ws = new_ws_handle;
Ok(())
},
Poison => {
should_exit = true;
Ok(())
},
};
if let Err(e) = error {
events_failure |= e.should_trigger_interconnect_rebuild();
conn_failure |= e.should_trigger_connect();
}
(events_failure, conn_failure, should_exit)
}
#[inline]
fn fire_event(&self, event: EventMessage) -> Result<()> {
// As this task is responsible for noticing the potential death of an event context,
// it's responsible for not forcibly recreating said context repeatedly.
if !self.prevent_events {
self.interconnect.events.send(event)?;
Ok(())
} else {
Ok(())
}
}
#[inline]
fn add_track(&mut self, mut track: Track) -> Result<()> {
let evts = track.events.take().unwrap_or_default();
let state = track.state();
let handle = track.handle.clone();
self.tracks.push(track);
self.interconnect
.events
.send(EventMessage::AddTrack(evts, state, handle))?;
Ok(())
}
// rebuilds the event thread's view of each track, in event of a full rebuild.
#[inline]
fn rebuild_tracks(&mut self) -> Result<()> {
for track in self.tracks.iter_mut() {
let evts = track.events.take().unwrap_or_default();
let state = track.state();
let handle = track.handle.clone();
self.interconnect
.events
.send(EventMessage::AddTrack(evts, state, handle))?;
}
Ok(())
}
#[inline]
fn audio_commands_events(&mut self) -> Result<()> {
// Apply user commands.
for (i, track) in self.tracks.iter_mut().enumerate() {
// This causes fallible event system changes,
// but if the event thread has died then we'll certainly
// detect that on the tick later.
// Changes to play state etc. MUST all be handled.
track.process_commands(i, &self.interconnect);
}
// TODO: do without vec?
let mut i = 0;
let mut to_remove = Vec::with_capacity(self.tracks.len());
while i < self.tracks.len() {
let track = self
.tracks
.get_mut(i)
.expect("Tried to remove an illegal track index.");
if track.playing.is_done() {
let p_state = track.playing();
let to_drop = self.tracks.swap_remove(i);
to_remove.push(i);
self.fire_event(EventMessage::ChangeState(
i,
TrackStateChange::Mode(p_state),
))?;
let _ = self.disposer.send(DisposalMessage::Track(to_drop));
} else {
i += 1;
}
}
// Tick
self.fire_event(EventMessage::Tick)?;
// Then do removals.
for i in &to_remove[..] {
self.fire_event(EventMessage::RemoveTrack(*i))?;
}
Ok(())
}
#[inline]
fn march_deadline(&mut self) {
if self.skip_sleep {
return;
}
std::thread::sleep(self.deadline.saturating_duration_since(Instant::now()));
self.deadline += TIMESTEP_LENGTH;
}
pub fn cycle(&mut self) -> Result<()> {
let mut mix_buffer = [0f32; STEREO_FRAME_SIZE];
// Walk over all the audio files, combining into one audio frame according
// to volume, play state, etc.
let mut mix_len = {
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload_mut();
// self.mix_tracks(&mut payload[TAG_SIZE..], &mut mix_buffer)
mix_tracks(
&mut payload[TAG_SIZE..],
&mut mix_buffer,
&mut self.tracks,
&self.interconnect,
self.prevent_events,
)
};
self.soft_clip.apply((&mut mix_buffer[..]).try_into()?)?;
if self.muted {
mix_len = MixType::MixedPcm(0);
}
if mix_len == MixType::MixedPcm(0) {
if self.silence_frames > 0 {
self.silence_frames -= 1;
// Explicit "Silence" frame.
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload_mut();
(&mut payload[TAG_SIZE..TAG_SIZE + SILENT_FRAME.len()])
.copy_from_slice(&SILENT_FRAME[..]);
mix_len = MixType::Passthrough(SILENT_FRAME.len());
} else {
// Per official guidelines, send 5x silence BEFORE we stop speaking.
if let Some(ws) = &self.ws {
// NOTE: this should prevent a catastrophic thread pileup.
// A full reconnect might cause an inner closed connection.
// It's safer to leave the central task to clean this up and
// pass the mixer a new channel.
let _ = ws.send(WsMessage::Speaking(false));
}
self.march_deadline();
return Ok(());
}
} else {
self.silence_frames = 5;
}
if let Some(ws) = &self.ws {
ws.send(WsMessage::Speaking(true))?;
}
self.march_deadline();
self.prep_and_send_packet(mix_buffer, mix_len)?;
Ok(())
}
fn set_bitrate(&mut self, bitrate: Bitrate) -> Result<()> {
self.encoder.set_bitrate(bitrate).map_err(Into::into)
}
#[inline]
fn prep_and_send_packet(&mut self, buffer: [f32; 1920], mix_len: MixType) -> Result<()> {
let conn = self
.conn_active
.as_mut()
.expect("Shouldn't be mixing packets without access to a cipher + UDP dest.");
let index = {
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload_mut();
let crypto_mode = conn.crypto_state.kind();
let payload_len = match mix_len {
MixType::Passthrough(opus_len) => opus_len,
MixType::MixedPcm(_samples) => {
let total_payload_space = payload.len() - crypto_mode.payload_suffix_len();
self.encoder.encode_float(
&buffer[..STEREO_FRAME_SIZE],
&mut payload[TAG_SIZE..total_payload_space],
)?
},
};
let final_payload_size = conn
.crypto_state
.write_packet_nonce(&mut rtp, TAG_SIZE + payload_len);
conn.crypto_state.kind().encrypt_in_place(
&mut rtp,
&conn.cipher,
final_payload_size,
)?;
RtpPacket::minimum_packet_size() + final_payload_size
};
// TODO: This is dog slow, don't do this.
// Can we replace this with a shared ring buffer + semaphore?
// i.e., do something like double/triple buffering in graphics.
conn.udp_tx
.send(UdpTxMessage::Packet(self.packet[..index].to_vec()))?;
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
rtp.set_sequence(rtp.get_sequence() + 1);
rtp.set_timestamp(rtp.get_timestamp() + MONO_FRAME_SIZE as u32);
Ok(())
}
}
#[derive(Debug, Eq, PartialEq)]
enum MixType {
Passthrough(usize),
MixedPcm(usize),
}
#[inline]
fn mix_tracks<'a>(
opus_frame: &'a mut [u8],
mix_buffer: &mut [f32; STEREO_FRAME_SIZE],
tracks: &mut Vec<Track>,
interconnect: &Interconnect,
prevent_events: bool,
) -> MixType {
let mut len = 0;
// Opus frame passthrough.
// This requires that we have only one track, who has volume 1.0, and an
// Opus codec type.
let do_passthrough = tracks.len() == 1 && {
let track = &tracks[0];
(track.volume - 1.0).abs() < f32::EPSILON && track.source.supports_passthrough()
};
for (i, track) in tracks.iter_mut().enumerate() {
let vol = track.volume;
let stream = &mut track.source;
if track.playing != PlayMode::Play {
continue;
}
let (temp_len, opus_len) = if do_passthrough {
(0, track.source.read_opus_frame(opus_frame).ok())
} else {
(stream.mix(mix_buffer, vol), None)
};
len = len.max(temp_len);
if temp_len > 0 || opus_len.is_some() {
track.step_frame();
} else if track.do_loop() {
if let Ok(time) = track.seek_time(Default::default()) {
// have to reproduce self.fire_event here
// to circumvent the borrow checker's lack of knowledge.
//
// In event of error, one of the later event calls will
// trigger the event thread rebuild: it is more prudent that
// the mixer works as normal right now.
if !prevent_events {
let _ = interconnect.events.send(EventMessage::ChangeState(
i,
TrackStateChange::Position(time),
));
let _ = interconnect.events.send(EventMessage::ChangeState(
i,
TrackStateChange::Loops(track.loops, false),
));
}
}
} else {
track.end();
}
if let Some(opus_len) = opus_len {
return MixType::Passthrough(opus_len);
}
}
MixType::MixedPcm(len)
}
/// The mixing thread is a synchronous context due to its compute-bound nature.
///
/// We pass in an async handle for the benefit of some Input classes (e.g., restartables)
/// who need to run their restart code elsewhere and return blank data until such time.
#[instrument(skip(interconnect, mix_rx, async_handle))]
pub(crate) fn runner(
interconnect: Interconnect,
mix_rx: Receiver<MixerMessage>,
async_handle: Handle,
config: Config,
) {
let mut mixer = Mixer::new(mix_rx, async_handle, interconnect, config);
mixer.run();
let _ = mixer.disposer.send(DisposalMessage::Poison);
}

View File

@@ -0,0 +1,444 @@
use super::*;
/// Mix a track's audio stream into either the shared mixing buffer, or directly into the output
/// packet ("passthrough") when possible.
///
/// Passthrough is highest performance, but the source MUST be opus, have 20ms frames, and be the only
/// live track. In this case we copy the opus-encoded data with no changes. Otherwise, we fall back to
/// below.
///
/// There are a few functional requirements here for non-passthrough mixing that make it tricky:
/// * Input frame lengths are not congruent with what we need to send (i.e., 26.12ms in MP3 vs
/// needed 20ms).
/// * Input audio arrives at a different sample rate from required (i.e., 44.1 vs needed 48 kHz).
/// * Input data may not be `f32`s.
/// * Input data may not match stereo/mono of desired output.
///
/// All of the above challenges often happen at once. The rough pipeline in processing is:
///
/// until source end or 20 ms taken:
/// (use previous frame 'til empty / get new frame) -> [resample] -> [audio += vol * (sample as f32)]
///
/// Typically, we mix between a subset of the input packet and the output buf because the 20ms window
/// straddles packet boundaries. If there's enough space AND 48kHz AND receive f32s, then we use a fast
/// path.
///
/// In the mono -> stereo case, we duplicate across all target channels. In stereo -> mono, we average
/// the samples from each channel.
///
/// To avoid needing to hold onto resampled data longer than one mix cycle, we take enough input samples
/// to fill a chunk of the mixer (e.g., 10ms == 20ms / 2) so that they will all be used.
///
/// This is a fairly annoying piece of code to reason about, mainly because you need to hold so many
/// internal positions into: the mix buffer, resample buffers, and previous/current packets
/// for a stream.
#[inline]
pub fn mix_symph_indiv(
// shared buffer to mix into.
symph_mix: &mut AudioBuffer<f32>,
// buffer to hold built up packet
resample_scratch: &mut AudioBuffer<f32>,
// the input stream to use
input: &mut Parsed,
// resampler state and positions into partially read packets
local_state: &mut DecodeState,
// volume of this source
volume: f32,
// window into the output UDP buffer to copy opus frames into.
// This is set to `Some` IF passthrough is possible (i.e., one live source).
mut opus_slot: Option<&mut [u8]>,
) -> (MixType, MixStatus) {
let mut samples_written = 0;
let mut resample_in_progress = false;
let mut track_status = MixStatus::Live;
let codec_type = input.decoder.codec_params().codec;
resample_scratch.clear();
while samples_written != MONO_FRAME_SIZE {
// fetch a packet: either in progress, passthrough (early exit), or
let source_packet = if local_state.inner_pos != 0 {
Some(input.decoder.last_decoded())
} else if let Ok(pkt) = input.format.next_packet() {
if pkt.track_id() != input.track_id {
continue;
}
let buf = pkt.buf();
// Opus packet passthrough special case.
if codec_type == CODEC_TYPE_OPUS && local_state.passthrough != Passthrough::Block {
if let Some(slot) = opus_slot.as_mut() {
let sample_ct = buf
.try_into()
.and_then(|buf| audiopus::packet::nb_samples(buf, SAMPLE_RATE));
// We don't actually block passthrough until a few violations are
// seen. The main one is that most Opus tracks end on a sub-20ms
// frame, particularly on Youtube.
// However, a frame that's bigger than the target buffer is an instant block.
let buf_size_fatal = buf.len() <= slot.len();
if match sample_ct {
Ok(MONO_FRAME_SIZE) => true,
_ => !local_state.record_and_check_passthrough_strike_final(buf_size_fatal),
} {
slot.write_all(buf)
.expect("Bounds check performed, and failure will block passthrough.");
return (MixType::Passthrough(buf.len()), MixStatus::Live);
}
}
}
input
.decoder
.decode(&pkt)
.map_err(|e| {
track_status = e.into();
})
.ok()
} else {
track_status = MixStatus::Ended;
None
};
// Cleanup: failed to get the next packet, but still have to convert and mix scratch.
if source_packet.is_none() {
if resample_in_progress {
// fill up remainder of buf with zeroes, resample, mix
let (chan_c, resampler, rs_out_buf) = local_state.resampler.as_mut().unwrap();
let in_len = resample_scratch.frames();
let to_render = resampler.input_frames_next().saturating_sub(in_len);
if to_render != 0 {
resample_scratch.render_reserved(Some(to_render));
for plane in resample_scratch.planes_mut().planes() {
for val in &mut plane[in_len..] {
*val = 0.0f32;
}
}
}
// Luckily, we make use of the WHOLE input buffer here.
resampler
.process_into_buffer(
&resample_scratch.planes().planes()[..*chan_c],
rs_out_buf,
None,
)
.unwrap();
// Calculate true end position using sample rate math
let ratio = (rs_out_buf[0].len() as f32) / (resample_scratch.frames() as f32);
let out_samples = (ratio * (in_len as f32)).round() as usize;
mix_resampled(rs_out_buf, symph_mix, samples_written, volume);
samples_written += out_samples;
}
break;
}
let source_packet = source_packet.unwrap();
let in_rate = source_packet.spec().rate;
if in_rate == SAMPLE_RATE_RAW as u32 {
// No need to resample: mix as standard.
let samples_marched = mix_over_ref(
&source_packet,
symph_mix,
local_state.inner_pos,
samples_written,
volume,
);
samples_written += samples_marched;
local_state.inner_pos += samples_marched;
local_state.inner_pos %= source_packet.frames();
} else {
// NOTE: this should NEVER change in one stream.
let chan_c = source_packet.spec().channels.count();
let (_, resampler, rs_out_buf) = local_state.resampler.get_or_insert_with(|| {
// TODO: integ. error handling here.
let resampler = FftFixedOut::new(
in_rate as usize,
SAMPLE_RATE_RAW,
RESAMPLE_OUTPUT_FRAME_SIZE,
4,
chan_c,
)
.expect("Failed to create resampler.");
let out_buf = resampler.output_buffer_allocate();
(chan_c, resampler, out_buf)
});
let inner_pos = local_state.inner_pos;
let pkt_frames = source_packet.frames();
if pkt_frames == 0 {
continue;
}
let needed_in_frames = resampler.input_frames_next();
let available_frames = pkt_frames - inner_pos;
let force_copy = resample_in_progress || needed_in_frames > available_frames;
if (!force_copy) && matches!(source_packet, AudioBufferRef::F32(_)) {
// This is the only case where we can pull off a straight resample...
// I would really like if this could be a slice of slices,
// but the technology just isn't there yet. And I don't feel like
// writing unsafe transformations to do so.
// NOTE: if let needed as if-let && {bool} is nightly only.
if let AudioBufferRef::F32(s_pkt) = source_packet {
let refs: Vec<&[f32]> = s_pkt
.planes()
.planes()
.iter()
.map(|s| &s[inner_pos..][..needed_in_frames])
.collect();
local_state.inner_pos += needed_in_frames;
local_state.inner_pos %= pkt_frames;
resampler
.process_into_buffer(&refs, rs_out_buf, None)
.unwrap();
} else {
unreachable!()
}
} else {
// We either lack enough samples, or have the wrong data format, forcing
// a conversion/copy into the buffer.
let old_scratch_len = resample_scratch.frames();
let missing_frames = needed_in_frames - old_scratch_len;
let frames_to_take = available_frames.min(missing_frames);
resample_scratch.render_reserved(Some(frames_to_take));
copy_into_resampler(
&source_packet,
resample_scratch,
inner_pos,
old_scratch_len,
frames_to_take,
);
local_state.inner_pos += frames_to_take;
local_state.inner_pos %= pkt_frames;
if resample_scratch.frames() == needed_in_frames {
resampler
.process_into_buffer(
&resample_scratch.planes().planes()[..chan_c],
rs_out_buf,
None,
)
.unwrap();
resample_scratch.clear();
resample_in_progress = false;
} else {
// Not enough data to fill the resampler: fetch more.
resample_in_progress = true;
continue;
}
};
let samples_marched = mix_resampled(rs_out_buf, symph_mix, samples_written, volume);
samples_written += samples_marched;
}
}
(MixType::MixedPcm(samples_written), track_status)
}
#[inline]
fn mix_over_ref(
source: &AudioBufferRef,
target: &mut AudioBuffer<f32>,
source_pos: usize,
dest_pos: usize,
volume: f32,
) -> usize {
match source {
AudioBufferRef::U8(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::U16(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::U24(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::U32(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::S8(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::S16(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::S24(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::S32(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::F32(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
AudioBufferRef::F64(v) => mix_symph_buffer(v, target, source_pos, dest_pos, volume),
}
}
#[inline]
fn mix_symph_buffer<S>(
source: &AudioBuffer<S>,
target: &mut AudioBuffer<f32>,
source_pos: usize,
dest_pos: usize,
volume: f32,
) -> usize
where
S: Sample + IntoSample<f32>,
{
// mix in source_packet[inner_pos..] til end of EITHER buffer.
let src_usable = source.frames() - source_pos;
let tgt_usable = target.frames() - dest_pos;
let mix_ct = src_usable.min(tgt_usable);
let target_chans = target.spec().channels.count();
let target_mono = target_chans == 1;
let source_chans = source.spec().channels.count();
let source_mono = source_chans == 1;
let source_planes = source.planes();
let source_raw_planes = source_planes.planes();
if source_mono {
// mix this signal into *all* output channels at req'd volume.
let source_plane = source_raw_planes[0];
for d_plane in (&mut *target.planes_mut().planes()).iter_mut() {
for (d, s) in d_plane[dest_pos..dest_pos + mix_ct]
.iter_mut()
.zip(source_plane[source_pos..source_pos + mix_ct].iter())
{
*d += volume * (*s).into_sample();
}
}
} else if target_mono {
// mix all signals into the one target channel: reduce aggregate volume
// by n_channels.
let vol_adj = 1.0 / (source_chans as f32);
let mut t_planes = target.planes_mut();
let d_plane = &mut *t_planes.planes()[0];
for s_plane in source_raw_planes[..].iter() {
for (d, s) in d_plane[dest_pos..dest_pos + mix_ct]
.iter_mut()
.zip(s_plane[source_pos..source_pos + mix_ct].iter())
{
*d += volume * vol_adj * (*s).into_sample();
}
}
} else {
// stereo -> stereo: don't change volume, map input -> output channels w/ no duplication
for (d_plane, s_plane) in (&mut *target.planes_mut().planes())
.iter_mut()
.zip(source_raw_planes[..].iter())
{
for (d, s) in d_plane[dest_pos..dest_pos + mix_ct]
.iter_mut()
.zip(s_plane[source_pos..source_pos + mix_ct].iter())
{
*d += volume * (*s).into_sample();
}
}
}
mix_ct
}
#[inline]
fn mix_resampled(
source: &[Vec<f32>],
target: &mut AudioBuffer<f32>,
dest_pos: usize,
volume: f32,
) -> usize {
let mix_ct = source[0].len();
let target_chans = target.spec().channels.count();
let target_mono = target_chans == 1;
let source_chans = source.len();
let source_mono = source_chans == 1;
// see `mix_symph_buffer` for explanations of stereo<->mono logic.
if source_mono {
let source_plane = &source[0];
for d_plane in (&mut *target.planes_mut().planes()).iter_mut() {
for (d, s) in d_plane[dest_pos..dest_pos + mix_ct]
.iter_mut()
.zip(source_plane)
{
*d += volume * s;
}
}
} else if target_mono {
let vol_adj = 1.0 / (source_chans as f32);
let mut t_planes = target.planes_mut();
let d_plane = &mut *t_planes.planes()[0];
for s_plane in source[..].iter() {
for (d, s) in d_plane[dest_pos..dest_pos + mix_ct].iter_mut().zip(s_plane) {
*d += volume * vol_adj * s;
}
}
} else {
for (d_plane, s_plane) in (&mut *target.planes_mut().planes())
.iter_mut()
.zip(source[..].iter())
{
for (d, s) in d_plane[dest_pos..dest_pos + mix_ct].iter_mut().zip(s_plane) {
*d += volume * (*s);
}
}
}
mix_ct
}
#[inline]
pub(crate) fn copy_into_resampler(
source: &AudioBufferRef,
target: &mut AudioBuffer<f32>,
source_pos: usize,
dest_pos: usize,
len: usize,
) -> usize {
match source {
AudioBufferRef::U8(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::U16(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::U24(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::U32(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::S8(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::S16(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::S24(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::S32(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::F32(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
AudioBufferRef::F64(v) => copy_symph_buffer(v, target, source_pos, dest_pos, len),
}
}
#[inline]
fn copy_symph_buffer<S>(
source: &AudioBuffer<S>,
target: &mut AudioBuffer<f32>,
source_pos: usize,
dest_pos: usize,
len: usize,
) -> usize
where
S: Sample + IntoSample<f32>,
{
for (d_plane, s_plane) in (&mut *target.planes_mut().planes())
.iter_mut()
.zip(source.planes().planes()[..].iter())
{
for (d, s) in d_plane[dest_pos..dest_pos + len]
.iter_mut()
.zip(s_plane[source_pos..source_pos + len].iter())
{
*d = (*s).into_sample();
}
}
len
}

View File

@@ -0,0 +1,924 @@
pub mod mix_logic;
mod pool;
mod result;
pub mod state;
pub mod track;
mod util;
use pool::*;
use result::*;
use state::*;
pub use track::*;
use super::{disposal, error::Result, message::*};
use crate::{
constants::*,
driver::MixMode,
events::EventStore,
input::{Input, Parsed},
tracks::{Action, LoopState, PlayError, PlayMode, TrackCommand, TrackHandle, TrackState, View},
Config,
};
use audiopus::{
coder::Encoder as OpusEncoder,
softclip::SoftClip,
Application as CodingMode,
Bitrate,
};
use discortp::{
rtp::{MutableRtpPacket, RtpPacket},
MutablePacket,
};
use flume::{Receiver, Sender, TryRecvError};
use rand::random;
use rubato::{FftFixedOut, Resampler};
use std::{
io::Write,
sync::Arc,
time::{Duration, Instant},
};
use symphonia_core::{
audio::{AudioBuffer, AudioBufferRef, Layout, SampleBuffer, Signal, SignalSpec},
codecs::CODEC_TYPE_OPUS,
conv::IntoSample,
formats::SeekTo,
sample::Sample,
units::Time,
};
use tokio::runtime::Handle;
use tracing::{debug, error, instrument, warn};
use xsalsa20poly1305::TAG_SIZE;
#[cfg(test)]
use crate::driver::test_config::{OutputMessage, OutputMode, TickStyle};
#[cfg(test)]
use discortp::Packet as _;
pub struct Mixer {
pub bitrate: Bitrate,
pub config: Arc<Config>,
pub conn_active: Option<MixerConnection>,
pub content_prep_sequence: u64,
pub deadline: Instant,
pub disposer: Sender<DisposalMessage>,
pub encoder: OpusEncoder,
pub interconnect: Interconnect,
pub mix_rx: Receiver<MixerMessage>,
pub muted: bool,
pub packet: [u8; VOICE_PACKET_MAX],
pub prevent_events: bool,
pub silence_frames: u8,
pub skip_sleep: bool,
pub soft_clip: SoftClip,
thread_pool: BlockyTaskPool,
pub ws: Option<Sender<WsMessage>>,
pub tracks: Vec<InternalTrack>,
track_handles: Vec<TrackHandle>,
sample_buffer: SampleBuffer<f32>,
symph_mix: AudioBuffer<f32>,
resample_scratch: AudioBuffer<f32>,
#[cfg(test)]
remaining_loops: Option<u64>,
}
fn new_encoder(bitrate: Bitrate, mix_mode: MixMode) -> Result<OpusEncoder> {
let mut encoder = OpusEncoder::new(SAMPLE_RATE, mix_mode.to_opus(), CodingMode::Audio)?;
encoder.set_bitrate(bitrate)?;
Ok(encoder)
}
impl Mixer {
pub fn new(
mix_rx: Receiver<MixerMessage>,
async_handle: Handle,
interconnect: Interconnect,
config: Config,
) -> Self {
let bitrate = DEFAULT_BITRATE;
let encoder = new_encoder(bitrate, config.mix_mode)
.expect("Failed to create encoder in mixing thread with known-good values.");
let soft_clip = SoftClip::new(config.mix_mode.to_opus());
let mut packet = [0u8; VOICE_PACKET_MAX];
let mut rtp = MutableRtpPacket::new(&mut packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
rtp.set_version(RTP_VERSION);
rtp.set_payload_type(RTP_PROFILE_TYPE);
rtp.set_sequence(random::<u16>().into());
rtp.set_timestamp(random::<u32>().into());
let tracks = Vec::with_capacity(1.max(config.preallocated_tracks));
let track_handles = Vec::with_capacity(1.max(config.preallocated_tracks));
// Create an object disposal thread here.
let (disposer, disposal_rx) = flume::unbounded();
std::thread::spawn(move || disposal::runner(disposal_rx));
let thread_pool = BlockyTaskPool::new(async_handle);
let symph_layout = config.mix_mode.symph_layout();
let config = config.into();
let sample_buffer = SampleBuffer::<f32>::new(
MONO_FRAME_SIZE as u64,
symphonia_core::audio::SignalSpec::new_with_layout(
SAMPLE_RATE_RAW as u32,
symph_layout,
),
);
let symph_mix = AudioBuffer::<f32>::new(
MONO_FRAME_SIZE as u64,
symphonia_core::audio::SignalSpec::new_with_layout(
SAMPLE_RATE_RAW as u32,
symph_layout,
),
);
let resample_scratch = AudioBuffer::<f32>::new(
MONO_FRAME_SIZE as u64,
SignalSpec::new_with_layout(SAMPLE_RATE_RAW as u32, Layout::Stereo),
);
Self {
bitrate,
config,
conn_active: None,
content_prep_sequence: 0,
deadline: Instant::now(),
disposer,
encoder,
interconnect,
mix_rx,
muted: false,
packet,
prevent_events: false,
silence_frames: 0,
skip_sleep: false,
soft_clip,
thread_pool,
ws: None,
tracks,
track_handles,
sample_buffer,
symph_mix,
resample_scratch,
#[cfg(test)]
remaining_loops: None,
}
}
fn run(&mut self) {
let mut events_failure = false;
let mut conn_failure = false;
#[cfg(test)]
let ignore_check = self.config.override_connection.is_some();
#[cfg(not(test))]
let ignore_check = false;
'runner: loop {
if self.conn_active.is_some() || ignore_check {
loop {
match self.mix_rx.try_recv() {
Ok(m) => {
let (events, conn, should_exit) = self.handle_message(m);
events_failure |= events;
conn_failure |= conn;
if should_exit {
break 'runner;
}
},
Err(TryRecvError::Disconnected) => {
break 'runner;
},
Err(TryRecvError::Empty) => {
break;
},
};
}
// The above action may have invalidated the connection; need to re-check!
// Also, if we're in a test mode we should unconditionally run packet mixing code.
if self.conn_active.is_some() || ignore_check {
if let Err(e) = self.cycle().and_then(|_| self.audio_commands_events()) {
events_failure |= e.should_trigger_interconnect_rebuild();
conn_failure |= e.should_trigger_connect();
debug!("Mixer thread cycle: {:?}", e);
}
}
} else {
match self.mix_rx.recv() {
Ok(m) => {
let (events, conn, should_exit) = self.handle_message(m);
events_failure |= events;
conn_failure |= conn;
if should_exit {
break 'runner;
}
},
Err(_) => {
break 'runner;
},
}
}
// event failure? rebuild interconnect.
// ws or udp failure? full connect
// (soft reconnect is covered by the ws task.)
//
// in both cases, send failure is fatal,
// but will only occur on disconnect.
// expecting this is fairly noisy, so exit silently.
if events_failure {
self.prevent_events = true;
let sent = self
.interconnect
.core
.send(CoreMessage::RebuildInterconnect);
events_failure = false;
if sent.is_err() {
break;
}
}
if conn_failure {
self.conn_active = None;
let sent = self.interconnect.core.send(CoreMessage::FullReconnect);
conn_failure = false;
if sent.is_err() {
break;
}
}
}
}
#[inline]
fn handle_message(&mut self, msg: MixerMessage) -> (bool, bool, bool) {
let mut events_failure = false;
let mut conn_failure = false;
let mut should_exit = false;
let error = match msg {
MixerMessage::AddTrack(t) => self.add_track(t),
MixerMessage::SetTrack(t) => {
self.tracks.clear();
let mut out = self.fire_event(EventMessage::RemoveAllTracks);
if let Some(t) = t {
// Do this unconditionally: this affects local state infallibly,
// with the event installation being the remote part.
if let Err(e) = self.add_track(t) {
out = Err(e);
}
}
out
},
MixerMessage::SetBitrate(b) => {
self.bitrate = b;
if let Err(e) = self.set_bitrate(b) {
error!("Failed to update bitrate {:?}", e);
}
Ok(())
},
MixerMessage::SetMute(m) => {
self.muted = m;
Ok(())
},
MixerMessage::SetConn(conn, ssrc) => {
self.conn_active = Some(conn);
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
rtp.set_ssrc(ssrc);
rtp.set_sequence(random::<u16>().into());
rtp.set_timestamp(random::<u32>().into());
self.deadline = Instant::now();
Ok(())
},
MixerMessage::DropConn => {
self.conn_active = None;
Ok(())
},
MixerMessage::ReplaceInterconnect(i) => {
self.prevent_events = false;
if let Some(ws) = &self.ws {
conn_failure |= ws.send(WsMessage::ReplaceInterconnect(i.clone())).is_err();
}
if let Some(conn) = &self.conn_active {
conn_failure |= conn
.udp_rx
.send(UdpRxMessage::ReplaceInterconnect(i.clone()))
.is_err();
}
self.interconnect = i;
self.rebuild_tracks()
},
MixerMessage::SetConfig(new_config) => {
if new_config.mix_mode != self.config.mix_mode {
self.soft_clip = SoftClip::new(new_config.mix_mode.to_opus());
if let Ok(enc) = new_encoder(self.bitrate, new_config.mix_mode) {
self.encoder = enc;
} else {
self.bitrate = DEFAULT_BITRATE;
self.encoder = new_encoder(self.bitrate, new_config.mix_mode)
.expect("Failed fallback rebuild of OpusEncoder with safe inputs.");
}
let sl = new_config.mix_mode.symph_layout();
self.sample_buffer = SampleBuffer::<f32>::new(
MONO_FRAME_SIZE as u64,
SignalSpec::new_with_layout(SAMPLE_RATE_RAW as u32, sl),
);
self.symph_mix = AudioBuffer::<f32>::new(
MONO_FRAME_SIZE as u64,
SignalSpec::new_with_layout(SAMPLE_RATE_RAW as u32, sl),
);
}
self.config = Arc::new(new_config.clone());
if self.tracks.capacity() < self.config.preallocated_tracks {
self.tracks
.reserve(self.config.preallocated_tracks - self.tracks.len());
}
if let Some(conn) = &self.conn_active {
conn_failure |= conn
.udp_rx
.send(UdpRxMessage::SetConfig(new_config))
.is_err();
}
Ok(())
},
MixerMessage::RebuildEncoder => match new_encoder(self.bitrate, self.config.mix_mode) {
Ok(encoder) => {
self.encoder = encoder;
Ok(())
},
Err(e) => {
error!("Failed to rebuild encoder. Resetting bitrate. {:?}", e);
self.bitrate = DEFAULT_BITRATE;
self.encoder = new_encoder(self.bitrate, self.config.mix_mode)
.expect("Failed fallback rebuild of OpusEncoder with safe inputs.");
Ok(())
},
},
MixerMessage::Ws(new_ws_handle) => {
self.ws = new_ws_handle;
Ok(())
},
MixerMessage::Poison => {
should_exit = true;
Ok(())
},
};
if let Err(e) = error {
events_failure |= e.should_trigger_interconnect_rebuild();
conn_failure |= e.should_trigger_connect();
}
(events_failure, conn_failure, should_exit)
}
#[inline]
fn fire_event(&self, event: EventMessage) -> Result<()> {
// As this task is responsible for noticing the potential death of an event context,
// it's responsible for not forcibly recreating said context repeatedly.
if !self.prevent_events {
self.interconnect.events.send(event)?;
}
Ok(())
}
#[inline]
pub fn add_track(&mut self, track: TrackContext) -> Result<()> {
let (track, evts, state, handle) = InternalTrack::decompose_track(track);
self.tracks.push(track);
self.track_handles.push(handle.clone());
self.interconnect
.events
.send(EventMessage::AddTrack(evts, state, handle))?;
Ok(())
}
// rebuilds the event thread's view of each track, in event of a full rebuild.
#[inline]
fn rebuild_tracks(&mut self) -> Result<()> {
for (track, handle) in self.tracks.iter().zip(self.track_handles.iter()) {
let evts = EventStore::default();
let state = track.state();
let handle = handle.clone();
self.interconnect
.events
.send(EventMessage::AddTrack(evts, state, handle))?;
}
Ok(())
}
#[inline]
fn audio_commands_events(&mut self) -> Result<()> {
// Apply user commands.
for (i, track) in self.tracks.iter_mut().enumerate() {
// This causes fallible event system changes,
// but if the event thread has died then we'll certainly
// detect that on the tick later.
// Changes to play state etc. MUST all be handled.
let action = track.process_commands(i, &self.interconnect);
if let Some(req) = action.seek_point {
track.seek(
i,
req,
&self.interconnect,
&self.thread_pool,
&self.config,
self.prevent_events,
);
}
if let Some(callback) = action.make_playable {
if let Err(e) = track.get_or_ready_input(
i,
&self.interconnect,
&self.thread_pool,
&self.config,
self.prevent_events,
) {
track.callbacks.make_playable = Some(callback);
if let Some(fail) = e.as_user() {
track.playing = PlayMode::Errored(fail);
}
if let Some(req) = e.into_seek_request() {
track.seek(
i,
req,
&self.interconnect,
&self.thread_pool,
&self.config,
self.prevent_events,
);
}
} else {
// Track is already ready: don't register callback and just act.
drop(callback.send(Ok(())));
}
}
}
let mut i = 0;
while i < self.tracks.len() {
let track = self
.tracks
.get_mut(i)
.expect("Tried to remove an illegal track index.");
if track.playing.is_done() {
let p_state = track.playing.clone();
let to_drop = self.tracks.swap_remove(i);
drop(
self.disposer
.send(DisposalMessage::Track(Box::new(to_drop))),
);
let to_drop = self.track_handles.swap_remove(i);
drop(self.disposer.send(DisposalMessage::Handle(to_drop)));
self.fire_event(EventMessage::ChangeState(
i,
TrackStateChange::Mode(p_state),
))?;
} else {
i += 1;
}
}
// Tick -- receive side also handles removals in same manner after it increments
// times etc.
self.fire_event(EventMessage::Tick)?;
Ok(())
}
#[cfg(test)]
fn _march_deadline(&mut self) {
match &self.config.tick_style {
TickStyle::Timed => {
std::thread::sleep(self.deadline.saturating_duration_since(Instant::now()));
self.deadline += TIMESTEP_LENGTH;
},
TickStyle::UntimedWithExecLimit(rx) => {
if self.remaining_loops.is_none() {
if let Ok(new_val) = rx.recv() {
self.remaining_loops = Some(new_val.wrapping_sub(1));
}
}
if let Some(cnt) = self.remaining_loops.as_mut() {
if *cnt == 0 {
self.remaining_loops = None;
} else {
*cnt = cnt.wrapping_sub(1);
}
}
},
}
}
#[cfg(not(test))]
#[inline(always)]
#[allow(clippy::inline_always)] // Justified, this is a very very hot path
fn _march_deadline(&mut self) {
std::thread::sleep(self.deadline.saturating_duration_since(Instant::now()));
self.deadline += TIMESTEP_LENGTH;
}
#[inline]
fn march_deadline(&mut self) {
if self.skip_sleep {
return;
}
self._march_deadline();
}
pub fn cycle(&mut self) -> Result<()> {
let mut mix_buffer = [0f32; STEREO_FRAME_SIZE];
// symph_mix is an `AudioBuffer` (planar format), we need to convert this
// later into an interleaved `SampleBuffer` for libopus.
self.symph_mix.clear();
self.symph_mix.render_reserved(Some(MONO_FRAME_SIZE));
self.resample_scratch.clear();
// Walk over all the audio files, combining into one audio frame according
// to volume, play state, etc.
let mut mix_len = {
let out = self.mix_tracks();
self.sample_buffer.copy_interleaved_typed(&self.symph_mix);
out
};
if self.muted {
mix_len = MixType::MixedPcm(0);
}
// Explicit "Silence" frame handling: if there is no mixed data, we must send
// ~5 frames of silence (unless another good audio frame appears) before we
// stop sending RTP frames.
if mix_len == MixType::MixedPcm(0) {
if self.silence_frames > 0 {
self.silence_frames -= 1;
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload_mut();
(&mut payload[TAG_SIZE..TAG_SIZE + SILENT_FRAME.len()])
.copy_from_slice(&SILENT_FRAME[..]);
mix_len = MixType::Passthrough(SILENT_FRAME.len());
} else {
// Per official guidelines, send 5x silence BEFORE we stop speaking.
if let Some(ws) = &self.ws {
// NOTE: this explicit `drop` should prevent a catastrophic thread pileup.
// A full reconnect might cause an inner closed connection.
// It's safer to leave the central task to clean this up and
// pass the mixer a new channel.
drop(ws.send(WsMessage::Speaking(false)));
}
self.march_deadline();
#[cfg(test)]
match &self.config.override_connection {
Some(OutputMode::Raw(tx)) =>
drop(tx.send(crate::driver::test_config::TickMessage::NoEl)),
Some(OutputMode::Rtp(tx)) =>
drop(tx.send(crate::driver::test_config::TickMessage::NoEl)),
None => {},
}
return Ok(());
}
} else {
self.silence_frames = 5;
if let MixType::MixedPcm(n) = mix_len {
// FIXME: When impling #134, prevent this copy from happening if softclip disabled.
// Offer sample_buffer.samples() to prep_and_send_packet.
// to apply soft_clip, we need this to be in a normal f32 buffer.
// unfortunately, SampleBuffer does not expose a `.samples_mut()`.
// hence, an extra copy...
let samples_to_copy = self.config.mix_mode.channels() * n;
(&mut mix_buffer[..samples_to_copy])
.copy_from_slice(&self.sample_buffer.samples()[..samples_to_copy]);
self.soft_clip.apply(
(&mut mix_buffer[..])
.try_into()
.expect("Mix buffer is known to have a valid sample count (softclip)."),
)?;
}
}
if let Some(ws) = &self.ws {
ws.send(WsMessage::Speaking(true))?;
}
// Wait till the right time to send this packet:
// usually a 20ms tick, in test modes this is either a finite number of runs or user input.
self.march_deadline();
#[cfg(test)]
if let Some(OutputMode::Raw(tx)) = &self.config.override_connection {
let msg = match mix_len {
MixType::Passthrough(len) if len == SILENT_FRAME.len() => OutputMessage::Silent,
MixType::Passthrough(len) => {
let rtp = RtpPacket::new(&self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload();
let opus_frame = (&payload[TAG_SIZE..][..len]).to_vec();
OutputMessage::Passthrough(opus_frame)
},
MixType::MixedPcm(_) => OutputMessage::Mixed(
mix_buffer[..self.config.mix_mode.sample_count_in_frame()].to_vec(),
),
};
drop(tx.send(msg.into()));
} else {
self.prep_and_send_packet(&mix_buffer, mix_len)?;
}
#[cfg(not(test))]
self.prep_and_send_packet(&mix_buffer, mix_len)?;
// Zero out all planes of the mix buffer if any audio was written.
if matches!(mix_len, MixType::MixedPcm(a) if a > 0) {
for plane in self.symph_mix.planes_mut().planes() {
plane.fill(0.0);
}
}
Ok(())
}
fn set_bitrate(&mut self, bitrate: Bitrate) -> Result<()> {
self.encoder.set_bitrate(bitrate).map_err(Into::into)
}
#[inline]
fn prep_and_send_packet(&mut self, buffer: &[f32; 1920], mix_len: MixType) -> Result<()> {
let conn = self
.conn_active
.as_mut()
.expect("Shouldn't be mixing packets without access to a cipher + UDP dest.");
let index = {
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload_mut();
let crypto_mode = conn.crypto_state.kind();
// If passthrough, Opus payload in place already.
// Else encode into buffer with space for AEAD encryption headers.
let payload_len = match mix_len {
MixType::Passthrough(opus_len) => opus_len,
MixType::MixedPcm(_samples) => {
let total_payload_space = payload.len() - crypto_mode.payload_suffix_len();
self.encoder.encode_float(
&buffer[..self.config.mix_mode.sample_count_in_frame()],
&mut payload[TAG_SIZE..total_payload_space],
)?
},
};
let final_payload_size = conn
.crypto_state
.write_packet_nonce(&mut rtp, TAG_SIZE + payload_len);
// Packet encryption ignored in test modes.
#[cfg(not(test))]
let encrypt = true;
#[cfg(test)]
let encrypt = self.config.override_connection.is_none();
if encrypt {
conn.crypto_state.kind().encrypt_in_place(
&mut rtp,
&conn.cipher,
final_payload_size,
)?;
}
RtpPacket::minimum_packet_size() + final_payload_size
};
#[cfg(test)]
if let Some(OutputMode::Rtp(tx)) = &self.config.override_connection {
// Test mode: send unencrypted (compressed) packets to local receiver.
drop(tx.send(self.packet[..index].to_vec().into()));
} else {
conn.udp_tx.send(self.packet[..index].to_vec())?;
}
#[cfg(not(test))]
{
// Normal operation: send encrypted payload to UDP Tx task.
// TODO: This is dog slow, don't do this.
// Can we replace this with a shared ring buffer + semaphore?
// or the BBQueue crate?
conn.udp_tx.send(self.packet[..index].to_vec())?;
}
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
rtp.set_sequence(rtp.get_sequence() + 1);
rtp.set_timestamp(rtp.get_timestamp() + MONO_FRAME_SIZE as u32);
Ok(())
}
#[inline]
fn mix_tracks(&mut self) -> MixType {
// Get a slice of bytes to write in data for Opus packet passthrough.
let mut rtp = MutableRtpPacket::new(&mut self.packet[..]).expect(
"FATAL: Too few bytes in self.packet for RTP header.\
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload_mut();
let opus_frame = &mut payload[TAG_SIZE..];
// Opus frame passthrough.
// This requires that we have only one PLAYING track, who has volume 1.0, and an
// Opus codec type (verified later in mix_symph_indiv).
//
// We *could* cache the number of live tracks separately, but that makes this
// quite fragile given all the ways a user can alter the PlayMode.
let mut num_live = 0;
let mut last_live_vol = 1.0;
for track in &self.tracks {
if track.playing.is_playing() {
num_live += 1;
last_live_vol = track.volume;
}
}
let do_passthrough = num_live == 1 && (last_live_vol - 1.0).abs() < f32::EPSILON;
let mut len = 0;
for (i, track) in self.tracks.iter_mut().enumerate() {
let vol = track.volume;
// This specifically tries to get tracks who are "preparing",
// so that event handlers and the like can all be fired without
// the track being in a `Play` state.
if !track.should_check_input() {
continue;
}
let should_play = track.playing.is_playing();
let input = track.get_or_ready_input(
i,
&self.interconnect,
&self.thread_pool,
&self.config,
self.prevent_events,
);
let (input, mix_state) = match input {
Ok(i) => i,
Err(InputReadyingError::Waiting) => continue,
Err(InputReadyingError::NeedsSeek(req)) => {
track.seek(
i,
req,
&self.interconnect,
&self.thread_pool,
&self.config,
self.prevent_events,
);
continue;
},
// TODO: allow for retry in given time.
Err(e) => {
if let Some(fail) = e.as_user() {
track.playing = PlayMode::Errored(fail);
}
continue;
},
};
// Now that we have dealt with potential errors in preparing tracks,
// only do any mixing if the track is to be played!
if !should_play {
continue;
}
let (mix_type, status) = mix_logic::mix_symph_indiv(
&mut self.symph_mix,
&mut self.resample_scratch,
input,
mix_state,
vol,
do_passthrough.then(|| &mut *opus_frame),
);
let return_here = if let MixType::MixedPcm(pcm_len) = mix_type {
len = len.max(pcm_len);
false
} else {
if mix_state.passthrough == Passthrough::Inactive {
input.decoder.reset();
}
mix_state.passthrough = Passthrough::Active;
true
};
// FIXME: allow Ended to trigger a seek/loop/revisit in the same mix cycle?
// Would this be possible with special-casing to mark some inputs as fast
// to recreate? Probably not doable in the general case.
match status {
MixStatus::Live => track.step_frame(),
MixStatus::Errored(e) =>
track.playing = PlayMode::Errored(PlayError::Decode(e.into())),
MixStatus::Ended if track.do_loop() => {
drop(self.track_handles[i].seek(Duration::default()));
if !self.prevent_events {
// position update is sent out later, when the seek concludes.
drop(self.interconnect.events.send(EventMessage::ChangeState(
i,
TrackStateChange::Loops(track.loops, false),
)));
}
},
MixStatus::Ended => {
track.end();
},
}
// This needs to happen here due to borrow checker shenanigans.
if return_here {
return mix_type;
}
}
MixType::MixedPcm(len)
}
}
/// The mixing thread is a synchronous context due to its compute-bound nature.
///
/// We pass in an async handle for the benefit of some Input classes (e.g., restartables)
/// who need to run their restart code elsewhere and return blank data until such time.
#[instrument(skip(interconnect, mix_rx, async_handle))]
pub(crate) fn runner(
interconnect: Interconnect,
mix_rx: Receiver<MixerMessage>,
async_handle: Handle,
config: Config,
) {
Mixer::new(mix_rx, async_handle, interconnect, config).run();
}

View File

@@ -0,0 +1,148 @@
use super::util::copy_seek_to;
use crate::{
driver::tasks::message::MixerInputResultMessage,
input::{AudioStream, AudioStreamError, Compose, Input, LiveInput, Parsed},
Config,
};
use flume::Sender;
use parking_lot::RwLock;
use std::{result::Result as StdResult, sync::Arc, time::Duration};
use symphonia_core::{
formats::{SeekMode, SeekTo},
io::MediaSource,
};
use tokio::runtime::Handle;
#[derive(Clone)]
pub struct BlockyTaskPool {
pool: Arc<RwLock<rusty_pool::ThreadPool>>,
handle: Handle,
}
impl BlockyTaskPool {
pub fn new(handle: Handle) -> Self {
Self {
pool: Arc::new(RwLock::new(rusty_pool::ThreadPool::new(
1,
64,
Duration::from_secs(300),
))),
handle,
}
}
pub fn create(
&self,
callback: Sender<MixerInputResultMessage>,
input: Input,
seek_time: Option<SeekTo>,
config: Arc<Config>,
) {
// Moves an Input from Lazy -> Live.
// We either do this on this pool, or move it to the tokio executor as the source requires.
// This takes a seek_time to pass on and execute *after* parsing (i.e., back-seek on
// read-only stream).
match input {
Input::Lazy(mut lazy) => {
let far_pool = self.clone();
if lazy.should_create_async() {
self.handle.spawn(async move {
let out = lazy.create_async().await;
far_pool.send_to_parse(out, lazy, callback, seek_time, config);
});
} else {
let pool = self.pool.read();
pool.execute(move || {
let out = lazy.create();
far_pool.send_to_parse(out, lazy, callback, seek_time, config);
});
}
},
Input::Live(live, maybe_create) =>
self.parse(config, callback, live, maybe_create, seek_time),
}
}
pub fn send_to_parse(
&self,
create_res: StdResult<AudioStream<Box<dyn MediaSource>>, AudioStreamError>,
rec: Box<dyn Compose>,
callback: Sender<MixerInputResultMessage>,
seek_time: Option<SeekTo>,
config: Arc<Config>,
) {
match create_res {
Ok(o) => {
self.parse(config, callback, LiveInput::Raw(o), Some(rec), seek_time);
},
Err(e) => {
drop(callback.send(MixerInputResultMessage::CreateErr(e.into())));
},
}
}
pub fn parse(
&self,
config: Arc<Config>,
callback: Sender<MixerInputResultMessage>,
input: LiveInput,
rec: Option<Box<dyn Compose>>,
seek_time: Option<SeekTo>,
) {
let pool_clone = self.clone();
let pool = self.pool.read();
pool.execute(
move || match input.promote(config.codec_registry, config.format_registry) {
Ok(LiveInput::Parsed(parsed)) => match seek_time {
// If seek time is zero, then wipe it out.
// Some formats (MKV) make SeekTo(0) require a backseek to realign with the
// current page.
Some(seek_time) if !super::util::seek_to_is_zero(&seek_time) => {
pool_clone.seek(callback, parsed, rec, seek_time, false, config);
},
_ => {
drop(callback.send(MixerInputResultMessage::Built(parsed, rec)));
},
},
Ok(_) => unreachable!(),
Err(e) => {
drop(callback.send(MixerInputResultMessage::ParseErr(e.into())));
},
},
);
}
pub fn seek(
&self,
callback: Sender<MixerInputResultMessage>,
mut input: Parsed,
rec: Option<Box<dyn Compose>>,
seek_time: SeekTo,
// Not all of symphonia's formats bother to return SeekErrorKind::ForwardOnly.
// So, we need *this* flag.
backseek_needed: bool,
config: Arc<Config>,
) {
let pool_clone = self.clone();
let pool = self.pool.read();
pool.execute(move || match rec {
Some(rec) if (!input.supports_backseek) && backseek_needed => {
pool_clone.create(callback, Input::Lazy(rec), Some(seek_time), config);
},
_ => {
let seek_result = input
.format
.seek(SeekMode::Accurate, copy_seek_to(&seek_time));
input.decoder.reset();
drop(callback.send(MixerInputResultMessage::Seek(
input,
rec,
seek_result.map_err(Arc::new),
)));
},
});
}
}

View File

@@ -0,0 +1,55 @@
use crate::{
input::AudioStreamError,
tracks::{PlayError, SeekRequest},
};
use std::sync::Arc;
use symphonia_core::errors::Error as SymphoniaError;
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum MixType {
Passthrough(usize),
MixedPcm(usize),
}
pub enum MixStatus {
Live,
Ended,
Errored(SymphoniaError),
}
impl From<SymphoniaError> for MixStatus {
fn from(e: SymphoniaError) -> Self {
Self::Errored(e)
}
}
// The Symph errors are Arc'd here since if they come up, they will always
// be Arc'd anyway via into_user.
#[derive(Clone, Debug)]
pub enum InputReadyingError {
Parsing(Arc<SymphoniaError>),
Creation(Arc<AudioStreamError>),
Seeking(Arc<SymphoniaError>),
Dropped,
Waiting,
NeedsSeek(SeekRequest),
}
impl InputReadyingError {
pub fn as_user(&self) -> Option<PlayError> {
match self {
Self::Parsing(e) => Some(PlayError::Parse(e.clone())),
Self::Creation(e) => Some(PlayError::Create(e.clone())),
Self::Seeking(e) => Some(PlayError::Seek(e.clone())),
_ => None,
}
}
pub fn into_seek_request(self) -> Option<SeekRequest> {
if let Self::NeedsSeek(a) = self {
Some(a)
} else {
None
}
}
}

View File

@@ -0,0 +1,104 @@
use crate::{
constants::OPUS_PASSTHROUGH_STRIKE_LIMIT,
driver::tasks::message::*,
input::{Compose, Input, LiveInput, Metadata, Parsed},
tracks::{ReadyState, SeekRequest},
};
use flume::Receiver;
use rubato::FftFixedOut;
use std::time::Instant;
pub enum InputState {
NotReady(Input),
Preparing(PreparingInfo),
Ready(Parsed, Option<Box<dyn Compose>>),
}
impl InputState {
pub fn metadata(&mut self) -> Option<Metadata> {
if let Self::Ready(parsed, _) = self {
Some(parsed.into())
} else {
None
}
}
pub fn ready_state(&self) -> ReadyState {
match self {
Self::NotReady(_) => ReadyState::Uninitialised,
Self::Preparing(_) => ReadyState::Preparing,
Self::Ready(_, _) => ReadyState::Playable,
}
}
}
impl From<Input> for InputState {
fn from(val: Input) -> Self {
match val {
a @ Input::Lazy(_) => Self::NotReady(a),
Input::Live(live, rec) => match live {
LiveInput::Parsed(p) => Self::Ready(p, rec),
other => Self::NotReady(Input::Live(other, rec)),
},
}
}
}
pub struct PreparingInfo {
#[allow(dead_code)]
/// Time this request was fired.
pub time: Instant,
/// Used to handle seek requests fired while a track was being created (or a seek was in progress).
pub queued_seek: Option<SeekRequest>,
/// Callback from the thread pool to indicate the result of creating/parsing this track.
pub callback: Receiver<MixerInputResultMessage>,
}
pub struct DecodeState {
pub inner_pos: usize,
pub resampler: Option<(usize, FftFixedOut<f32>, Vec<Vec<f32>>)>,
pub passthrough: Passthrough,
pub passthrough_violations: u8,
}
impl DecodeState {
pub fn reset(&mut self) {
self.inner_pos = 0;
self.resampler = None;
}
pub fn record_and_check_passthrough_strike_final(&mut self, fatal: bool) -> bool {
self.passthrough_violations = self.passthrough_violations.saturating_add(1);
let blocked = fatal || self.passthrough_violations > OPUS_PASSTHROUGH_STRIKE_LIMIT;
if blocked {
self.passthrough = Passthrough::Block;
}
blocked
}
}
impl Default for DecodeState {
fn default() -> Self {
Self {
inner_pos: 0,
resampler: None,
passthrough: Passthrough::Inactive,
passthrough_violations: 0,
}
}
}
/// Simple state to manage decoder resets etc.
///
/// Inactive->Active transitions should trigger a reset.
///
/// Block should be used if a source contains known-bad packets:
/// it's unlikely that packet sizes will vary, but if they do then
/// we can't passthrough (and every attempt will trigger a codec reset,
/// which probably won't sound too smooth).
#[derive(Clone, Copy, Eq, PartialEq)]
pub enum Passthrough {
Active,
Inactive,
Block,
}

View File

@@ -0,0 +1,400 @@
use crate::tracks::{ReadyState, SeekRequest};
use std::result::Result as StdResult;
use symphonia_core::errors::Error as SymphError;
use super::*;
pub struct InternalTrack {
pub(crate) playing: PlayMode,
pub(crate) volume: f32,
pub(crate) input: InputState,
pub(crate) mix_state: DecodeState,
pub(crate) position: Duration,
pub(crate) play_time: Duration,
pub(crate) commands: Receiver<TrackCommand>,
pub(crate) loops: LoopState,
pub(crate) callbacks: Callbacks,
}
impl<'a> InternalTrack {
pub(crate) fn decompose_track(
val: TrackContext,
) -> (Self, EventStore, TrackState, TrackHandle) {
let TrackContext {
handle,
track,
receiver,
} = val;
let out = InternalTrack {
playing: track.playing,
volume: track.volume,
input: InputState::from(track.input),
mix_state: DecodeState::default(),
position: Duration::default(),
play_time: Duration::default(),
commands: receiver,
loops: track.loops,
callbacks: Callbacks::default(),
};
let state = out.state();
(out, track.events, state, handle)
}
pub(crate) fn state(&self) -> TrackState {
let ready = self.input.ready_state();
TrackState {
playing: self.playing.clone(),
volume: self.volume,
position: self.position,
play_time: self.play_time,
loops: self.loops,
ready,
}
}
pub(crate) fn view(&'a mut self) -> View<'a> {
let ready = self.input.ready_state();
View {
position: &self.position,
play_time: &self.play_time,
volume: &mut self.volume,
meta: self.input.metadata(),
ready,
playing: &mut self.playing,
loops: &mut self.loops,
}
}
pub(crate) fn process_commands(&mut self, index: usize, ic: &Interconnect) -> Action {
// Note: disconnection and an empty channel are both valid,
// and should allow the audio object to keep running as intended.
// We also need to export a target seek point to the mixer, if known.
let mut action = Action::default();
// Note that interconnect failures are not currently errors.
// In correct operation, the event thread should never panic,
// but it receiving status updates is secondary do actually
// doing the work.
while let Ok(cmd) = self.commands.try_recv() {
match cmd {
TrackCommand::Play => {
self.playing.change_to(PlayMode::Play);
drop(ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Mode(self.playing.clone()),
)));
},
TrackCommand::Pause => {
self.playing.change_to(PlayMode::Pause);
drop(ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Mode(self.playing.clone()),
)));
},
TrackCommand::Stop => {
self.playing.change_to(PlayMode::Stop);
drop(ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Mode(self.playing.clone()),
)));
},
TrackCommand::Volume(vol) => {
self.volume = vol;
drop(ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Volume(self.volume),
)));
},
TrackCommand::Seek(req) => action.seek_point = Some(req),
TrackCommand::AddEvent(evt) => {
drop(ic.events.send(EventMessage::AddTrackEvent(index, evt)));
},
TrackCommand::Do(func) => {
if let Some(indiv_action) = func(self.view()) {
action.combine(indiv_action);
}
drop(ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Total(self.state()),
)));
},
TrackCommand::Request(tx) => {
drop(tx.send(self.state()));
},
TrackCommand::Loop(loops) => {
self.loops = loops;
drop(ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Loops(self.loops, true),
)));
},
TrackCommand::MakePlayable(callback) => action.make_playable = Some(callback),
}
}
action
}
pub(crate) fn do_loop(&mut self) -> bool {
match self.loops {
LoopState::Infinite => true,
LoopState::Finite(0) => false,
LoopState::Finite(ref mut n) => {
*n -= 1;
true
},
}
}
/// Steps playback location forward by one frame.
pub(crate) fn step_frame(&mut self) {
self.position += TIMESTEP_LENGTH;
self.play_time += TIMESTEP_LENGTH;
}
pub(crate) fn should_check_input(&self) -> bool {
self.playing.is_playing() || matches!(self.input, InputState::Preparing(_))
}
pub(crate) fn end(&mut self) -> &mut Self {
self.playing.change_to(PlayMode::End);
self
}
/// Readies the requested input state.
///
/// Returns the usable version of the audio if available, and whether the track should be deleted.
pub(crate) fn get_or_ready_input(
&'a mut self,
id: usize,
interconnect: &Interconnect,
pool: &BlockyTaskPool,
config: &Arc<Config>,
prevent_events: bool,
) -> StdResult<(&'a mut Parsed, &'a mut DecodeState), InputReadyingError> {
let input = &mut self.input;
let mix_state = &mut self.mix_state;
let (out, queued_seek) = match input {
InputState::NotReady(_) => {
let (tx, rx) = flume::bounded(1);
let mut state = InputState::Preparing(PreparingInfo {
time: Instant::now(),
queued_seek: None,
callback: rx,
});
std::mem::swap(&mut state, input);
match state {
InputState::NotReady(a @ Input::Lazy(_)) => {
pool.create(tx, a, None, config.clone());
},
InputState::NotReady(Input::Live(audio, rec)) => {
pool.parse(config.clone(), tx, audio, rec, None);
},
_ => unreachable!(),
}
if !prevent_events {
drop(interconnect.events.send(EventMessage::ChangeState(
id,
TrackStateChange::Ready(ReadyState::Preparing),
)));
}
(Err(InputReadyingError::Waiting), None)
},
InputState::Preparing(info) => {
let queued_seek = info.queued_seek.take();
let orig_out = match info.callback.try_recv() {
Ok(MixerInputResultMessage::Built(parsed, rec)) => {
*input = InputState::Ready(parsed, rec);
mix_state.reset();
// possible TODO: set position to the true track position here?
// ISSUE: need to get next_packet to see its `ts`, but inner_pos==0
// will trigger next packet to be taken at mix time.
if !prevent_events {
drop(interconnect.events.send(EventMessage::ChangeState(
id,
TrackStateChange::Ready(ReadyState::Playable),
)));
}
self.callbacks.playable();
if let InputState::Ready(ref mut parsed, _) = input {
Ok(parsed)
} else {
unreachable!()
}
},
Ok(MixerInputResultMessage::Seek(parsed, rec, seek_res)) => {
match seek_res {
Ok(pos) =>
if let Some(time_base) = parsed.decoder.codec_params().time_base {
// Update track's position to match the actual timestamp the
// seek landed at.
let new_time = time_base.calc_time(pos.actual_ts);
let time_in_float = new_time.seconds as f64 + new_time.frac;
self.position =
std::time::Duration::from_secs_f64(time_in_float);
self.callbacks.seeked(self.position);
self.callbacks.playable();
if !prevent_events {
drop(interconnect.events.send(EventMessage::ChangeState(
id,
TrackStateChange::Position(self.position),
)));
drop(interconnect.events.send(EventMessage::ChangeState(
id,
TrackStateChange::Ready(ReadyState::Playable),
)));
}
// Our decoder state etc. must be reset.
// (Symphonia decoder state reset in the thread pool during
// the operation.)
mix_state.reset();
*input = InputState::Ready(parsed, rec);
if let InputState::Ready(ref mut parsed, _) = input {
Ok(parsed)
} else {
unreachable!()
}
} else {
Err(InputReadyingError::Seeking(
SymphError::Unsupported("Track had no recorded time base.")
.into(),
))
},
Err(e) => Err(InputReadyingError::Seeking(e)),
}
},
Ok(MixerInputResultMessage::CreateErr(e)) =>
Err(InputReadyingError::Creation(e)),
Ok(MixerInputResultMessage::ParseErr(e)) => Err(InputReadyingError::Parsing(e)),
Err(TryRecvError::Disconnected) => Err(InputReadyingError::Dropped),
Err(TryRecvError::Empty) => Err(InputReadyingError::Waiting),
};
let orig_out = orig_out.map(|a| (a, mix_state));
if let Err(ref e) = orig_out {
if let Some(e) = e.as_user() {
self.callbacks.readying_error(e);
}
}
(orig_out, queued_seek)
},
InputState::Ready(ref mut parsed, _) => (Ok((parsed, mix_state)), None),
};
match (out, queued_seek) {
(Ok(_), Some(request)) => Err(InputReadyingError::NeedsSeek(request)),
(a, _) => a,
}
}
pub(crate) fn seek(
&mut self,
id: usize,
request: SeekRequest,
interconnect: &Interconnect,
pool: &BlockyTaskPool,
config: &Arc<Config>,
prevent_events: bool,
) {
if let InputState::Preparing(p) = &mut self.input {
p.queued_seek = Some(request);
return;
}
// might be a little topsy turvy: rethink me.
let SeekRequest { time, callback } = request;
self.callbacks.seek = Some(callback);
if !prevent_events {
drop(interconnect.events.send(EventMessage::ChangeState(
id,
TrackStateChange::Ready(ReadyState::Preparing),
)));
}
let backseek_needed = time < self.position;
let time = Time::from(time.as_secs_f64());
let mut ts = SeekTo::Time {
time,
track_id: None,
};
let (tx, rx) = flume::bounded(1);
let state = std::mem::replace(
&mut self.input,
InputState::Preparing(PreparingInfo {
time: Instant::now(),
callback: rx,
queued_seek: None,
}),
);
match state {
InputState::Ready(p, r) => {
if let SeekTo::Time { time: _, track_id } = &mut ts {
*track_id = Some(p.track_id);
}
pool.seek(tx, p, r, ts, backseek_needed, config.clone());
},
InputState::NotReady(lazy) => pool.create(tx, lazy, Some(ts), config.clone()),
InputState::Preparing(_) => unreachable!(), // Covered above.
}
}
}
#[derive(Debug, Default)]
pub struct Callbacks {
pub seek: Option<Sender<StdResult<Duration, PlayError>>>,
pub make_playable: Option<Sender<StdResult<(), PlayError>>>,
}
impl Callbacks {
fn readying_error(&mut self, err: PlayError) {
if let Some(callback) = self.seek.take() {
drop(callback.send(Err(err.clone())));
}
if let Some(callback) = self.make_playable.take() {
drop(callback.send(Err(err)));
}
}
fn playable(&mut self) {
if let Some(callback) = self.make_playable.take() {
drop(callback.send(Ok(())));
}
}
fn seeked(&mut self, time: Duration) {
if let Some(callback) = self.seek.take() {
drop(callback.send(Ok(time)));
}
}
}

View File

@@ -0,0 +1,20 @@
use symphonia_core::{formats::SeekTo, units::Time};
// SeekTo lacks Copy and Clone... somehow.
pub fn copy_seek_to(pos: &SeekTo) -> SeekTo {
match *pos {
SeekTo::Time { time, track_id } => SeekTo::Time { time, track_id },
SeekTo::TimeStamp { ts, track_id } => SeekTo::TimeStamp { ts, track_id },
}
}
pub fn seek_to_is_zero(pos: &SeekTo) -> bool {
match *pos {
SeekTo::Time { time, .. } =>
time == Time {
seconds: 0,
frac: 0.0,
},
SeekTo::TimeStamp { ts, .. } => ts == 0,
}
}

View File

@@ -21,7 +21,7 @@ use crate::{
Config,
ConnectionInfo,
};
use flume::{Receiver, RecvError, Sender};
use flume::{Receiver, Sender};
use message::*;
use tokio::{runtime::Handle, spawn, time::sleep as tsleep};
use tracing::{debug, instrument, trace};
@@ -70,23 +70,21 @@ async fn runner(mut config: Config, rx: Receiver<CoreMessage>, tx: Sender<CoreMe
let mut retrying = None;
let mut attempt_idx = 0;
loop {
match rx.recv_async().await {
Ok(CoreMessage::ConnectWithResult(info, tx)) => {
while let Ok(msg) = rx.recv_async().await {
match msg {
CoreMessage::ConnectWithResult(info, tx) => {
config = if let Some(new_config) = next_config.take() {
let _ = interconnect
.mixer
.send(MixerMessage::SetConfig(new_config.clone()));
drop(
interconnect
.mixer
.send(MixerMessage::SetConfig(new_config.clone())),
);
new_config
} else {
config
};
if connection
.as_ref()
.map(|conn| conn.info != info)
.unwrap_or(true)
{
if connection.as_ref().map_or(true, |conn| conn.info != info) {
// Only *actually* reconnect if the conn info changed, or we don't have an
// active connection.
// This allows the gateway component to keep sending join requests independent
@@ -97,10 +95,10 @@ async fn runner(mut config: Config, rx: Receiver<CoreMessage>, tx: Sender<CoreMe
} else {
// No reconnection was attempted as there's a valid, identical connection;
// tell the outside listener that the operation was a success.
let _ = tx.send(Ok(()));
drop(tx.send(Ok(())));
}
},
Ok(CoreMessage::RetryConnect(retry_idx)) => {
CoreMessage::RetryConnect(retry_idx) => {
debug!("Retrying idx: {} (vs. {})", retry_idx, attempt_idx);
if retry_idx == attempt_idx {
if let Some(progress) = retrying.take() {
@@ -110,68 +108,68 @@ async fn runner(mut config: Config, rx: Receiver<CoreMessage>, tx: Sender<CoreMe
}
}
},
Ok(CoreMessage::Disconnect) => {
CoreMessage::Disconnect => {
let last_conn = connection.take();
let _ = interconnect.mixer.send(MixerMessage::DropConn);
let _ = interconnect.mixer.send(MixerMessage::RebuildEncoder);
drop(interconnect.mixer.send(MixerMessage::DropConn));
drop(interconnect.mixer.send(MixerMessage::RebuildEncoder));
if let Some(conn) = last_conn {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::DriverDisconnect(InternalDisconnect {
kind: DisconnectKind::Runtime,
reason: None,
info: conn.info.clone(),
}),
));
)));
}
},
Ok(CoreMessage::SignalWsClosure(ws_idx, ws_info, mut reason)) => {
CoreMessage::SignalWsClosure(ws_idx, ws_info, mut reason) => {
// if idx is not a match, quash reason
// (i.e., prevent users from mistakenly trying to reconnect for an *old* dead conn).
// if it *is* a match, the conn needs to die!
// (as the WS channel has truly given up the ghost).
if ws_idx != attempt_idx {
reason = None;
} else {
if ws_idx == attempt_idx {
connection = None;
let _ = interconnect.mixer.send(MixerMessage::DropConn);
let _ = interconnect.mixer.send(MixerMessage::RebuildEncoder);
drop(interconnect.mixer.send(MixerMessage::DropConn));
drop(interconnect.mixer.send(MixerMessage::RebuildEncoder));
} else {
reason = None;
}
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::DriverDisconnect(InternalDisconnect {
kind: DisconnectKind::Runtime,
reason,
info: ws_info,
}),
));
)));
},
Ok(CoreMessage::SetTrack(s)) => {
let _ = interconnect.mixer.send(MixerMessage::SetTrack(s));
CoreMessage::SetTrack(s) => {
drop(interconnect.mixer.send(MixerMessage::SetTrack(s)));
},
Ok(CoreMessage::AddTrack(s)) => {
let _ = interconnect.mixer.send(MixerMessage::AddTrack(s));
CoreMessage::AddTrack(s) => {
drop(interconnect.mixer.send(MixerMessage::AddTrack(s)));
},
Ok(CoreMessage::SetBitrate(b)) => {
let _ = interconnect.mixer.send(MixerMessage::SetBitrate(b));
CoreMessage::SetBitrate(b) => {
drop(interconnect.mixer.send(MixerMessage::SetBitrate(b)));
},
Ok(CoreMessage::SetConfig(mut new_config)) => {
CoreMessage::SetConfig(mut new_config) => {
next_config = Some(new_config.clone());
new_config.make_safe(&config, connection.is_some());
let _ = interconnect.mixer.send(MixerMessage::SetConfig(new_config));
drop(interconnect.mixer.send(MixerMessage::SetConfig(new_config)));
},
Ok(CoreMessage::AddEvent(evt)) => {
let _ = interconnect.events.send(EventMessage::AddGlobalEvent(evt));
CoreMessage::AddEvent(evt) => {
drop(interconnect.events.send(EventMessage::AddGlobalEvent(evt)));
},
Ok(CoreMessage::RemoveGlobalEvents) => {
let _ = interconnect.events.send(EventMessage::RemoveGlobalEvents);
CoreMessage::RemoveGlobalEvents => {
drop(interconnect.events.send(EventMessage::RemoveGlobalEvents));
},
Ok(CoreMessage::Mute(m)) => {
let _ = interconnect.mixer.send(MixerMessage::SetMute(m));
CoreMessage::Mute(m) => {
drop(interconnect.mixer.send(MixerMessage::SetMute(m)));
},
Ok(CoreMessage::Reconnect) => {
CoreMessage::Reconnect => {
if let Some(mut conn) = connection.take() {
// try once: if interconnect, try again.
// if still issue, full connect.
@@ -201,16 +199,16 @@ async fn runner(mut config: Config, rx: Receiver<CoreMessage>, tx: Sender<CoreMe
.attempt(&mut retrying, &interconnect, &config)
.await;
} else if let Some(ref connection) = &connection {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::DriverReconnect(InternalConnect {
info: connection.info.clone(),
ssrc: connection.ssrc,
}),
));
)));
}
}
},
Ok(CoreMessage::FullReconnect) =>
CoreMessage::FullReconnect =>
if let Some(conn) = connection.take() {
let info = conn.info.clone();
@@ -218,12 +216,10 @@ async fn runner(mut config: Config, rx: Receiver<CoreMessage>, tx: Sender<CoreMe
.attempt(&mut retrying, &interconnect, &config)
.await;
},
Ok(CoreMessage::RebuildInterconnect) => {
CoreMessage::RebuildInterconnect => {
interconnect.restart_volatile_internals();
},
Err(RecvError::Disconnected) | Ok(CoreMessage::Poison) => {
break;
},
CoreMessage::Poison => break,
}
}
@@ -275,22 +271,22 @@ impl ConnectionRetryData {
match self.flavour {
ConnectionFlavour::Connect(tx) => {
// Other side may not be listening: this is fine.
let _ = tx.send(Ok(()));
drop(tx.send(Ok(())));
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::DriverConnect(InternalConnect {
info: connection.info.clone(),
ssrc: connection.ssrc,
}),
));
)));
},
ConnectionFlavour::Reconnect => {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::DriverReconnect(InternalConnect {
info: connection.info.clone(),
ssrc: connection.ssrc,
}),
));
)));
},
}
@@ -304,7 +300,7 @@ impl ConnectionRetryData {
spawn(async move {
tsleep(t).await;
let _ = remote_ic.core.send(CoreMessage::RetryConnect(idx));
drop(remote_ic.core.send(CoreMessage::RetryConnect(idx)));
});
self.attempts += 1;
@@ -325,24 +321,24 @@ impl ConnectionRetryData {
match self.flavour {
ConnectionFlavour::Connect(tx) => {
// See above.
let _ = tx.send(Err(why));
drop(tx.send(Err(why)));
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::DriverDisconnect(InternalDisconnect {
kind: DisconnectKind::Connect,
reason,
info: self.info,
}),
));
)));
},
ConnectionFlavour::Reconnect => {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::DriverDisconnect(InternalDisconnect {
kind: DisconnectKind::Reconnect,
reason,
info: self.info,
}),
));
)));
},
}
}

View File

@@ -5,7 +5,7 @@ use super::{
};
use crate::{
constants::*,
driver::DecodeMode,
driver::{CryptoMode, DecodeMode},
events::{internal_data::*, CoreContext},
};
use audiopus::{
@@ -53,27 +53,25 @@ enum PacketDecodeSize {
impl PacketDecodeSize {
fn bump_up(self) -> Self {
use PacketDecodeSize::*;
match self {
TwentyMillis => ThirtyMillis,
ThirtyMillis => FortyMillis,
FortyMillis => SixtyMillis,
SixtyMillis | Max => Max,
Self::TwentyMillis => Self::ThirtyMillis,
Self::ThirtyMillis => Self::FortyMillis,
Self::FortyMillis => Self::SixtyMillis,
Self::SixtyMillis | Self::Max => Self::Max,
}
}
fn can_bump_up(self) -> bool {
self != PacketDecodeSize::Max
self != Self::Max
}
fn len(self) -> usize {
use PacketDecodeSize::*;
match self {
TwentyMillis => STEREO_FRAME_SIZE,
ThirtyMillis => (STEREO_FRAME_SIZE / 2) * 3,
FortyMillis => 2 * STEREO_FRAME_SIZE,
SixtyMillis => 3 * STEREO_FRAME_SIZE,
Max => 6 * STEREO_FRAME_SIZE,
Self::TwentyMillis => STEREO_FRAME_SIZE,
Self::ThirtyMillis => (STEREO_FRAME_SIZE / 2) * 3,
Self::FortyMillis => 2 * STEREO_FRAME_SIZE,
Self::SixtyMillis => 3 * STEREO_FRAME_SIZE,
Self::Max => 6 * STEREO_FRAME_SIZE,
}
}
}
@@ -86,7 +84,7 @@ enum SpeakingDelta {
}
impl SsrcState {
fn new(pkt: RtpPacket<'_>) -> Self {
fn new(pkt: &RtpPacket<'_>) -> Self {
Self {
silent_frame_count: 5, // We do this to make the first speech packet fire an event.
decoder: OpusDecoder::new(SAMPLE_RATE, Channels::Stereo)
@@ -98,7 +96,7 @@ impl SsrcState {
fn process(
&mut self,
pkt: RtpPacket<'_>,
pkt: &RtpPacket<'_>,
data_offset: usize,
data_trailer: usize,
decode_mode: DecodeMode,
@@ -198,11 +196,10 @@ impl SsrcState {
// and then remember that.
loop {
let tried_audio_len = self.decoder.decode(
Some((&data[start..]).try_into()?),
Some(data[start..].try_into()?),
(&mut out[..]).try_into()?,
false,
);
match tried_audio_len {
Ok(audio_len) => {
// Decoding to stereo: audio_len refers to sample count irrespective of channel count.
@@ -243,7 +240,6 @@ struct UdpRx {
config: Config,
packet_buffer: [u8; VOICE_PACKET_MAX],
rx: Receiver<UdpRxMessage>,
udp_socket: Arc<UdpSocket>,
}
@@ -256,15 +252,14 @@ impl UdpRx {
self.process_udp_message(interconnect, len);
}
msg = self.rx.recv_async() => {
use UdpRxMessage::*;
match msg {
Ok(ReplaceInterconnect(i)) => {
Ok(UdpRxMessage::ReplaceInterconnect(i)) => {
*interconnect = i;
},
Ok(SetConfig(c)) => {
Ok(UdpRxMessage::SetConfig(c)) => {
self.config = c;
},
Ok(Poison) | Err(_) => break,
Err(flume::RecvError::Disconnected) => break,
}
}
}
@@ -284,7 +279,7 @@ impl UdpRx {
match demux::demux_mut(packet) {
DemuxedMut::Rtp(mut rtp) => {
if !rtp_valid(rtp.to_immutable()) {
if !rtp_valid(&rtp.to_immutable()) {
error!("Illegal RTP message received.");
return;
}
@@ -303,9 +298,10 @@ impl UdpRx {
None
};
let rtp = rtp.to_immutable();
let (rtp_body_start, rtp_body_tail, decrypted) = packet_data.unwrap_or_else(|| {
(
crypto_mode.payload_prefix_len(),
CryptoMode::payload_prefix_len(),
crypto_mode.payload_suffix_len(),
false,
)
@@ -314,10 +310,10 @@ impl UdpRx {
let entry = self
.decoder_map
.entry(rtp.get_ssrc())
.or_insert_with(|| SsrcState::new(rtp.to_immutable()));
.or_insert_with(|| SsrcState::new(&rtp));
if let Ok((delta, audio)) = entry.process(
rtp.to_immutable(),
&rtp,
rtp_body_start,
rtp_body_tail,
self.config.decode_mode,
@@ -325,32 +321,32 @@ impl UdpRx {
) {
match delta {
SpeakingDelta::Start => {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::SpeakingUpdate(InternalSpeakingUpdate {
ssrc: rtp.get_ssrc(),
speaking: true,
}),
));
)));
},
SpeakingDelta::Stop => {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::SpeakingUpdate(InternalSpeakingUpdate {
ssrc: rtp.get_ssrc(),
speaking: false,
}),
));
)));
},
_ => {},
SpeakingDelta::Same => {},
}
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::VoicePacket(InternalVoicePacket {
audio,
packet: rtp.from_packet(),
payload_offset: rtp_body_start,
payload_end_pad: rtp_body_tail,
}),
));
)));
} else {
warn!("RTP decoding/processing failed.");
}
@@ -370,26 +366,23 @@ impl UdpRx {
let (start, tail) = packet_data.unwrap_or_else(|| {
(
crypto_mode.payload_prefix_len(),
CryptoMode::payload_prefix_len(),
crypto_mode.payload_suffix_len(),
)
});
let _ =
interconnect
.events
.send(EventMessage::FireCoreEvent(CoreContext::RtcpPacket(
InternalRtcpPacket {
packet: rtcp.from_packet(),
payload_offset: start,
payload_end_pad: tail,
},
)));
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::RtcpPacket(InternalRtcpPacket {
packet: rtcp.from_packet(),
payload_offset: start,
payload_end_pad: tail,
}),
)));
},
DemuxedMut::FailedParse(t) => {
warn!("Failed to parse message of type {:?}.", t);
},
_ => {
DemuxedMut::TooSmall => {
warn!("Illegal UDP packet from voice server.");
},
}
@@ -408,7 +401,7 @@ pub(crate) async fn runner(
let mut state = UdpRx {
cipher,
decoder_map: Default::default(),
decoder_map: HashMap::new(),
config,
packet_buffer: [0u8; VOICE_PACKET_MAX],
rx,
@@ -421,6 +414,6 @@ pub(crate) async fn runner(
}
#[inline]
fn rtp_valid(packet: RtpPacket<'_>) -> bool {
fn rtp_valid(packet: &RtpPacket<'_>) -> bool {
packet.get_version() == RTP_VERSION && packet.get_payload_type() == RTP_PROFILE_TYPE
}

View File

@@ -12,7 +12,6 @@ use tracing::{error, instrument, trace};
struct UdpTx {
ssrc: u32,
rx: Receiver<UdpTxMessage>,
udp_tx: Arc<UdpSocket>,
}
@@ -26,7 +25,6 @@ impl UdpTx {
let mut ka_time = Instant::now() + UDP_KEEPALIVE_GAP;
loop {
use UdpTxMessage::*;
match timeout_at(ka_time, self.rx.recv_async()).await {
Err(_) => {
trace!("Sending UDP Keepalive.");
@@ -36,16 +34,12 @@ impl UdpTx {
}
ka_time += UDP_KEEPALIVE_GAP;
},
Ok(Ok(Packet(p))) =>
Ok(Ok(p)) =>
if let Err(e) = self.udp_tx.send(&p[..]).await {
error!("Fatal UDP packet send error: {:?}.", e);
break;
},
Ok(Err(e)) => {
error!("Fatal UDP packet receive error: {:?}.", e);
break;
},
Ok(Ok(Poison)) => {
Ok(Err(flume::RecvError::Disconnected)) => {
break;
},
}

View File

@@ -140,7 +140,7 @@ impl AuxNetwork {
}
}
},
Err(_) | Ok(WsMessage::Poison) => {
Err(flume::RecvError::Disconnected) => {
break;
},
}
@@ -151,13 +151,13 @@ impl AuxNetwork {
self.dont_send = true;
if should_reconnect {
let _ = interconnect.core.send(CoreMessage::Reconnect);
drop(interconnect.core.send(CoreMessage::Reconnect));
} else {
let _ = interconnect.core.send(CoreMessage::SignalWsClosure(
drop(interconnect.core.send(CoreMessage::SignalWsClosure(
self.attempt_idx,
self.info.clone(),
ws_reason,
));
)));
break;
}
}
@@ -186,17 +186,17 @@ impl AuxNetwork {
fn process_ws(&mut self, interconnect: &Interconnect, value: GatewayEvent) {
match value {
GatewayEvent::Speaking(ev) => {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::SpeakingStateUpdate(ev),
));
)));
},
GatewayEvent::ClientConnect(ev) => {
debug!("Received discontinued ClientConnect: {:?}", ev);
},
GatewayEvent::ClientDisconnect(ev) => {
let _ = interconnect.events.send(EventMessage::FireCoreEvent(
drop(interconnect.events.send(EventMessage::FireCoreEvent(
CoreContext::ClientDisconnect(ev),
));
)));
},
GatewayEvent::HeartbeatAck(ev) => {
if let Some(nonce) = self.last_heartbeat_nonce.take() {

251
src/driver/test_config.rs Normal file
View File

@@ -0,0 +1,251 @@
#![allow(missing_docs)]
use flume::{Receiver, Sender};
use crate::{
tracks::{PlayMode, TrackHandle, TrackState},
Event,
EventContext,
EventHandler,
TrackEvent,
};
use std::time::Duration;
#[allow(dead_code)]
#[derive(Clone, Debug)]
pub enum TickStyle {
Timed,
UntimedWithExecLimit(Receiver<u64>),
}
#[derive(Clone, Debug, PartialEq)]
pub enum OutputMessage {
Passthrough(Vec<u8>),
Mixed(Vec<f32>),
Silent,
}
#[allow(dead_code)]
impl OutputMessage {
pub fn is_passthrough(&self) -> bool {
matches!(self, Self::Passthrough(_))
}
pub fn is_mixed(&self) -> bool {
matches!(self, Self::Mixed(_))
}
pub fn is_mixed_with_nonzero_signal(&self) -> bool {
if let Self::Mixed(data) = self {
data.iter().any(|v| *v != 0.0f32)
} else {
false
}
}
pub fn is_explicit_silence(&self) -> bool {
*self == Self::Silent
}
}
#[allow(dead_code)]
#[derive(Clone, Debug)]
pub enum OutputMode {
Raw(Sender<TickMessage<OutputMessage>>),
Rtp(Sender<TickMessage<Vec<u8>>>),
}
#[allow(dead_code)]
#[derive(Clone, Debug)]
pub enum TickMessage<T> {
El(T),
NoEl,
}
impl<T> From<T> for TickMessage<T> {
fn from(val: T) -> Self {
TickMessage::El(val)
}
}
impl From<TickMessage<OutputMessage>> for OutputPacket {
fn from(val: TickMessage<OutputMessage>) -> Self {
match val {
TickMessage::El(e) => OutputPacket::Raw(e),
TickMessage::NoEl => OutputPacket::Empty,
}
}
}
impl From<TickMessage<Vec<u8>>> for OutputPacket {
fn from(val: TickMessage<Vec<u8>>) -> Self {
match val {
TickMessage::El(e) => OutputPacket::Rtp(e),
TickMessage::NoEl => OutputPacket::Empty,
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub enum OutputPacket {
Raw(OutputMessage),
Rtp(Vec<u8>),
Empty,
}
impl OutputPacket {
pub fn raw(&self) -> Option<&OutputMessage> {
if let Self::Raw(o) = self {
Some(o)
} else {
None
}
}
}
#[derive(Clone, Debug)]
pub enum OutputReceiver {
Raw(Receiver<TickMessage<OutputMessage>>),
Rtp(Receiver<TickMessage<Vec<u8>>>),
}
#[derive(Clone)]
pub struct DriverTestHandle {
pub rx: OutputReceiver,
pub tx: Sender<u64>,
}
impl DriverTestHandle {
pub fn recv(&self) -> OutputPacket {
match &self.rx {
OutputReceiver::Raw(rx) => rx.recv().unwrap().into(),
OutputReceiver::Rtp(rx) => rx.recv().unwrap().into(),
}
}
pub async fn recv_async(&self) -> OutputPacket {
match &self.rx {
OutputReceiver::Raw(rx) => rx.recv_async().await.unwrap().into(),
OutputReceiver::Rtp(rx) => rx.recv_async().await.unwrap().into(),
}
}
pub fn len(&self) -> usize {
match &self.rx {
OutputReceiver::Raw(rx) => rx.len(),
OutputReceiver::Rtp(rx) => rx.len(),
}
}
pub fn wait(&self, n_ticks: u64) {
for _i in 0..n_ticks {
drop(self.recv());
}
}
pub async fn wait_async(&self, n_ticks: u64) {
for _i in 0..n_ticks {
drop(self.recv_async().await);
}
}
pub async fn spawn_ticker(&self) {
let remote = self.clone();
tokio::spawn(async move {
loop {
remote.skip(1).await;
tokio::time::sleep(Duration::from_millis(1)).await;
}
});
}
pub fn wait_noisy(&self, n_ticks: u64) {
for _i in 0..n_ticks {
match self.recv() {
OutputPacket::Empty => eprintln!("pkt: Nothing"),
OutputPacket::Rtp(p) => eprintln!("pkt: RTP[{}B]", p.len()),
OutputPacket::Raw(OutputMessage::Silent) => eprintln!("pkt: Raw-Silent"),
OutputPacket::Raw(OutputMessage::Passthrough(p)) =>
eprintln!("pkt: Raw-Passthrough[{}B]", p.len()),
OutputPacket::Raw(OutputMessage::Mixed(p)) =>
eprintln!("pkt: Raw-Mixed[{}B]", p.len()),
}
}
}
pub async fn skip(&self, n_ticks: u64) {
self.tick(n_ticks);
self.wait_async(n_ticks).await;
}
pub fn tick(&self, n_ticks: u64) {
if n_ticks == 0 {
panic!("Number of ticks to advance driver/mixer must be >= 1.");
}
self.tx.send(n_ticks).unwrap();
}
pub async fn ready_track(
&self,
handle: &TrackHandle,
tick_wait: Option<Duration>,
) -> TrackState {
let (tx, rx) = flume::bounded(1);
let (err_tx, err_rx) = flume::bounded(1);
struct SongPlayable {
tx: Sender<TrackState>,
}
#[async_trait::async_trait]
impl EventHandler for SongPlayable {
async fn act(&self, ctx: &crate::EventContext<'_>) -> Option<Event> {
if let EventContext::Track(&[(state, _)]) = ctx {
drop(self.tx.send(state.clone()));
}
Some(Event::Cancel)
}
}
struct SongErred {
tx: Sender<PlayMode>,
}
#[async_trait::async_trait]
impl EventHandler for SongErred {
async fn act(&self, ctx: &crate::EventContext<'_>) -> Option<Event> {
if let EventContext::Track(&[(state, _)]) = ctx {
drop(self.tx.send(state.playing.clone()));
}
Some(Event::Cancel)
}
}
handle
.add_event(Event::Track(TrackEvent::Playable), SongPlayable { tx })
.expect("Adding track evt should not fail before any ticks.");
handle
.add_event(Event::Track(TrackEvent::Error), SongErred { tx: err_tx })
.expect("Adding track evt should not fail before any ticks.");
loop {
self.tick(1);
tokio::time::sleep(tick_wait.unwrap_or_else(|| Duration::from_millis(20))).await;
self.wait_async(1).await;
match err_rx.try_recv() {
Ok(e) => panic!("Error reported on track: {:?}", e),
Err(flume::TryRecvError::Empty | flume::TryRecvError::Disconnected) => {},
}
match rx.try_recv() {
Ok(val) => return val,
Err(flume::TryRecvError::Disconnected) => panic!(),
Err(flume::TryRecvError::Empty) => {},
}
}
}
}