servius/songbird
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Driver, Tracks: Cleanup of leaky types (#20)

Browse Source 
Main goal: a lot of nested future/result folding.

This mainly modifies error handling for Tracks and TrackHandles to be
more consistent, and hides the underlying channel result passing in
get_info. Errors returned should be far clearer, and are domain
specific rather than falling back to a very opaque use of the underlying
channel error. It should be clearer to users why their handle commands
failed, or why they can't make a ytdl track loop or similar.

Also fixed/cleaned up Songbird::join(_gateway) to return in a single
await, sparing the user from the underlying channel details and repeated
Errs. I was trying for some time to extend the same graces to `Call`,
but could not figure out a sane way to get a 'static version of the
first future in the chain (i.e., the gateway send) so that the whole
thing could happen after dropping the lock around the Call. I really
wanted to fix this to happen as a single folded await too, but I think
this might need some crazy hack or redesign.
This commit is contained in:
Kyle Simpson
2020-12-04 15:13:43 +00:00
committed by GitHub
parent 9fdbcd77be
commit f222ce9969
14 changed files with 276 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/tracks/error.rs Normal file
View File

@@ -0,0 +1,39 @@
use std::{error::Error, fmt};
/// Errors associated with control and manipulation of tracks.
///
/// Unless otherwise stated, these don't invalidate an existing track,
/// but do advise on valid operations and commands.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum TrackError {
/// The operation failed because the track has ended, has been removed
/// due to call closure, or some error within the driver.
Finished,
/// The supplied event listener can never be fired by a track, and should
/// be attached to the driver instead.
InvalidTrackEvent,
/// The track's underlying [`Input`] doesn't support seeking operations.
///
/// [`Input`]: crate::input::Input
SeekUnsupported,
}
impl fmt::Display for TrackError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "Failed to operate on track (handle): ")?;
match self {
TrackError::Finished => write!(f, "track ended."),
TrackError::InvalidTrackEvent =>
write!(f, "given event listener can't be fired on a track."),
TrackError::SeekUnsupported => write!(f, "track did not support seeking."),
}
}
}
impl Error for TrackError {}
/// Alias for most calls to a [`TrackHandle`].
///
/// [`TrackHandle`]: super::TrackHandle
pub type TrackResult<T> = Result<T, TrackError>;

82
src/tracks/handle.rs
View File

@@ -4,10 +4,7 @@ use crate::{
input::Metadata,
};
use std::{sync::Arc, time::Duration};
use tokio::sync::{
mpsc::{error::SendError, UnboundedSender},
oneshot,
};
use tokio::sync::{mpsc::UnboundedSender, oneshot};
use uuid::Uuid;
#[derive(Clone, Debug)]
@@ -46,12 +43,12 @@ impl TrackHandle {
}
/// Unpauses an audio track.
pub fn play(&self) -> TrackResult {
pub fn play(&self) -> TrackResult<()> {
self.send(TrackCommand::Play)
}
/// Pauses an audio track.
pub fn pause(&self) -> TrackResult {
pub fn pause(&self) -> TrackResult<()> {
self.send(TrackCommand::Pause)
}
@@ -61,12 +58,12 @@ impl TrackHandle {
/// a [`TrackEvent::End`] event.
///
/// [`TrackEvent::End`]: crate::events::TrackEvent::End
pub fn stop(&self) -> TrackResult {
pub fn stop(&self) -> TrackResult<()> {
self.send(TrackCommand::Stop)
}
/// Sets the volume of an audio track.
pub fn set_volume(&self, volume: f32) -> TrackResult {
pub fn set_volume(&self, volume: f32) -> TrackResult<()> {
self.send(TrackCommand::Volume(volume))
}
@@ -83,43 +80,43 @@ impl TrackHandle {
/// Seeks along the track to the specified position.
///
/// If the underlying [`Input`] does not support this behaviour,
/// then all calls will fail.
/// If the underlying [`Input`] does not support seeking,
/// then all calls will fail with [`TrackError::SeekUnsupported`].
///
/// [`Input`]: crate::input::Input
pub fn seek_time(&self, position: Duration) -> TrackResult {
/// [`TrackError::SeekUnsupported`]: TrackError::SeekUnsupported
pub fn seek_time(&self, position: Duration) -> TrackResult<()> {
if self.seekable {
self.send(TrackCommand::Seek(position))
} else {
Err(SendError(TrackCommand::Seek(position)))
Err(TrackError::SeekUnsupported)
}
}
/// Attach an event handler to an audio track. These will receive [`EventContext::Track`].
///
/// Users **must** ensure that no costly work or blocking occurs
/// within the supplied function or closure. *Taking excess time could prevent
/// timely sending of packets, causing audio glitches and delays*.
/// Events which can only be fired by the global context return [`TrackError::InvalidTrackEvent`]
///
/// [`Track`]: Track
/// [`EventContext::Track`]: crate::events::EventContext::Track
pub fn add_event<F: EventHandler + 'static>(&self, event: Event, action: F) -> TrackResult {
/// [`TrackError::InvalidTrackEvent`]: TrackError::InvalidTrackEvent
pub fn add_event<F: EventHandler + 'static>(&self, event: Event, action: F) -> TrackResult<()> {
let cmd = TrackCommand::AddEvent(EventData::new(event, action));
if event.is_global_only() {
Err(SendError(cmd))
Err(TrackError::InvalidTrackEvent)
} else {
self.send(cmd)
}
}
/// Perform an arbitrary action on a raw [`Track`] object.
/// Perform an arbitrary synchronous action on a raw [`Track`] object.
///
/// Users **must** ensure that no costly work or blocking occurs
/// within the supplied function or closure. *Taking excess time could prevent
/// timely sending of packets, causing audio glitches and delays*.
///
/// [`Track`]: Track
pub fn action<F>(&self, action: F) -> TrackResult
pub fn action<F>(&self, action: F) -> TrackResult<()>
where
F: FnOnce(&mut Track) + Send + Sync + 'static,
{
@@ -127,38 +124,55 @@ impl TrackHandle {
}
/// Request playback information and state from the audio context.
///
/// Crucially, the audio thread will respond *at a later time*:
/// It is up to the user when or how this should be read from the returned channel.
pub fn get_info(&self) -> TrackQueryResult {
pub async fn get_info(&self) -> TrackResult<Box<TrackState>> {
let (tx, rx) = oneshot::channel();
self.send(TrackCommand::Request(tx)).map(move |_| rx)
self.send(TrackCommand::Request(tx))?;
rx.await.map_err(|_| TrackError::Finished)
}
/// Set an audio track to loop indefinitely.
pub fn enable_loop(&self) -> TrackResult {
///
/// If the underlying [`Input`] does not support seeking,
/// then all calls will fail with [`TrackError::SeekUnsupported`].
///
/// [`Input`]: crate::input::Input
/// [`TrackError::SeekUnsupported`]: TrackError::SeekUnsupported
pub fn enable_loop(&self) -> TrackResult<()> {
if self.seekable {
self.send(TrackCommand::Loop(LoopState::Infinite))
} else {
Err(SendError(TrackCommand::Loop(LoopState::Infinite)))
Err(TrackError::SeekUnsupported)
}
}
/// Set an audio track to no longer loop.
pub fn disable_loop(&self) -> TrackResult {
///
/// If the underlying [`Input`] does not support seeking,
/// then all calls will fail with [`TrackError::SeekUnsupported`].
///
/// [`Input`]: crate::input::Input
/// [`TrackError::SeekUnsupported`]: TrackError::SeekUnsupported
pub fn disable_loop(&self) -> TrackResult<()> {
if self.seekable {
self.send(TrackCommand::Loop(LoopState::Finite(0)))
} else {
Err(SendError(TrackCommand::Loop(LoopState::Finite(0))))
Err(TrackError::SeekUnsupported)
}
}
/// Set an audio track to loop a set number of times.
pub fn loop_for(&self, count: usize) -> TrackResult {
///
/// If the underlying [`Input`] does not support seeking,
/// then all calls will fail with [`TrackError::SeekUnsupported`].
///
/// [`Input`]: crate::input::Input
/// [`TrackError::SeekUnsupported`]: TrackError::SeekUnsupported
pub fn loop_for(&self, count: usize) -> TrackResult<()> {
if self.seekable {
self.send(TrackCommand::Loop(LoopState::Finite(count)))
} else {
Err(SendError(TrackCommand::Loop(LoopState::Finite(count))))
Err(TrackError::SeekUnsupported)
}
}
@@ -182,7 +196,11 @@ impl TrackHandle {
/// Send a raw command to the [`Track`] object.
///
/// [`Track`]: Track
pub fn send(&self, cmd: TrackCommand) -> TrackResult {
self.command_channel.send(cmd)
pub fn send(&self, cmd: TrackCommand) -> TrackResult<()> {
// As the send channels are unbounded, we can be reasonably certain
// that send failure == cancellation.
self.command_channel
.send(cmd)
.map_err(|_e| TrackError::Finished)
}
}

116
src/tracks/mod.rs
View File

@@ -15,24 +15,18 @@
//! [`create_player`]: fn.create_player.html
mod command;
mod error;
mod handle;
mod looping;
mod mode;
mod queue;
mod state;
pub use self::{command::*, handle::*, looping::*, mode::*, queue::*, state::*};
pub use self::{command::*, error::*, handle::*, looping::*, mode::*, queue::*, state::*};
use crate::{constants::*, driver::tasks::message::*, events::EventStore, input::Input};
use std::time::Duration;
use tokio::sync::{
mpsc::{
self,
error::{SendError, TryRecvError},
UnboundedReceiver,
},
oneshot::Receiver as OneshotReceiver,
};
use tokio::sync::mpsc::{self, error::TryRecvError, UnboundedReceiver};
use uuid::Uuid;
/// Control object for audio playback.
@@ -63,18 +57,18 @@ use uuid::Uuid;
/// # };
/// ```
///
/// [`Driver::play_only`]: ../struct.Driver.html#method.play_only
/// [`Driver::play`]: ../struct.Driver.html#method.play
/// [`TrackHandle`]: struct.TrackHandle.html
/// [`create_player`]: fn.create_player.html
/// [`Driver::play_only`]: crate::driver::Driver::play_only
/// [`Driver::play`]: crate::driver::Driver::play
/// [`TrackHandle`]: TrackHandle
/// [`create_player`]: create_player
#[derive(Debug)]
pub struct Track {
/// Whether or not this sound is currently playing.
///
/// Can be controlled with [`play`] or [`pause`] if chaining is desired.
///
/// [`play`]: #method.play
/// [`pause`]: #method.pause
/// [`play`]: Track::play
/// [`pause`]: Track::pause
pub(crate) playing: PlayMode,
/// The desired volume for playback.
@@ -83,7 +77,7 @@ pub struct Track {
///
/// Can be controlled with [`volume`] if chaining is desired.
///
/// [`volume`]: #method.volume
/// [`volume`]: Track::volume
pub(crate) volume: f32,
/// Underlying data access object.
@@ -187,7 +181,7 @@ impl Track {
/// Sets [`volume`] in a manner that allows method chaining.
///
/// [`volume`]: #structfield.volume
/// [`volume`]: Track::volume
pub fn set_volume(&mut self, volume: f32) -> &mut Self {
self.volume = volume;
@@ -209,12 +203,20 @@ impl Track {
self.play_time
}
/// Sets [`loops`] in a manner that allows method chaining.
/// Set an audio track to loop a set number of times.
///
/// [`loops`]: #structfield.loops
pub fn set_loops(&mut self, loops: LoopState) -> &mut Self {
self.loops = loops;
self
/// If the underlying [`Input`] does not support seeking,
/// then all calls will fail with [`TrackError::SeekUnsupported`].
///
/// [`Input`]: crate::input::Input
/// [`TrackError::SeekUnsupported`]: TrackError::SeekUnsupported
pub fn set_loops(&mut self, loops: LoopState) -> TrackResult<()> {
if self.source.is_seekable() {
self.loops = loops;
Ok(())
} else {
Err(TrackError::SeekUnsupported)
}
}
pub(crate) fn do_loop(&mut self) -> bool {
@@ -234,11 +236,9 @@ impl Track {
self.play_time += TIMESTEP_LENGTH;
}
/// Receives and acts upon any commands forwarded by [`TrackHandle`]s.
/// Receives and acts upon any commands forwarded by TrackHandles.
///
/// *Used internally*, this should not be exposed to users.
///
/// [`TrackHandle`]: struct.TrackHandle.html
pub(crate) fn process_commands(&mut self, index: usize, ic: &Interconnect) {
// Note: disconnection and an empty channel are both valid,
// and should allow the audio object to keep running as intended.
@@ -280,13 +280,13 @@ impl Track {
TrackStateChange::Volume(self.volume),
));
},
Seek(time) => {
self.seek_time(time);
let _ = ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Position(self.position),
));
},
Seek(time) =>
if let Ok(new_time) = self.seek_time(time) {
let _ = ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Position(new_time),
));
},
AddEvent(evt) => {
let _ = ic.events.send(EventMessage::AddTrackEvent(index, evt));
},
@@ -300,13 +300,13 @@ impl Track {
Request(tx) => {
let _ = tx.send(Box::new(self.state()));
},
Loop(loops) => {
self.set_loops(loops);
let _ = ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Loops(self.loops, true),
));
},
Loop(loops) =>
if self.set_loops(loops).is_ok() {
let _ = ic.events.send(EventMessage::ChangeState(
index,
TrackStateChange::Loops(self.loops, true),
));
},
}
},
Err(TryRecvError::Closed) => {
@@ -325,7 +325,7 @@ impl Track {
/// The primary use-case of this is sending information across
/// threads in response to a [`TrackHandle`].
///
/// [`TrackHandle`]: struct.TrackHandle.html
/// [`TrackHandle`]: TrackHandle
pub fn state(&self) -> TrackState {
TrackState {
playing: self.playing,
@@ -338,15 +338,18 @@ impl Track {
/// Seek to a specific point in the track.
///
/// Returns `None` if unsupported.
pub fn seek_time(&mut self, pos: Duration) -> Option<Duration> {
let out = self.source.seek_time(pos);
if let Some(t) = out {
/// If the underlying [`Input`] does not support seeking,
/// then all calls will fail with [`TrackError::SeekUnsupported`].
///
/// [`Input`]: crate::input::Input
/// [`TrackError::SeekUnsupported`]: TrackError::SeekUnsupported
pub fn seek_time(&mut self, pos: Duration) -> TrackResult<Duration> {
if let Some(t) = self.source.seek_time(pos) {
self.position = t;
Ok(t)
} else {
Err(TrackError::SeekUnsupported)
}
out
}
/// Returns this track's unique identifier.
@@ -373,22 +376,3 @@ pub fn create_player(source: Input) -> (Track, TrackHandle) {
(player, handle)
}
/// Alias for most result-free calls to a [`TrackHandle`].
///
/// Failure indicates that the accessed audio object has been
/// removed or deleted by the audio context.
///
/// [`TrackHandle`]: TrackHandle
pub type TrackResult = Result<(), SendError<TrackCommand>>;
/// Alias for return value from calls to [`TrackHandle::get_info`].
///
/// Crucially, the audio thread will respond *at a later time*:
/// It is up to the user when or how this should be read from the returned channel.
///
/// Failure indicates that the accessed audio object has been
/// removed or deleted by the audio context.
///
/// [`TrackHandle::get_info`]: TrackHandle::get_info
pub type TrackQueryResult = Result<OneshotReceiver<Box<TrackState>>, SendError<TrackCommand>>;

8
src/tracks/queue.rs
View File

@@ -240,7 +240,7 @@ impl TrackQueue {
}
/// Pause the track at the head of the queue.
pub fn pause(&self) -> TrackResult {
pub fn pause(&self) -> TrackResult<()> {
let inner = self.inner.lock();
if let Some(handle) = inner.tracks.front() {
@@ -251,7 +251,7 @@ impl TrackQueue {
}
/// Resume the track at the head of the queue.
pub fn resume(&self) -> TrackResult {
pub fn resume(&self) -> TrackResult<()> {
let inner = self.inner.lock();
if let Some(handle) = inner.tracks.front() {
@@ -273,7 +273,7 @@ impl TrackQueue {
}
/// Skip to the next track in the queue, if it exists.
pub fn skip(&self) -> TrackResult {
pub fn skip(&self) -> TrackResult<()> {
let inner = self.inner.lock();
inner.stop_current()
@@ -295,7 +295,7 @@ impl TrackQueue {
impl TrackQueueCore {
/// Skip to the next track in the queue, if it exists.
fn stop_current(&self) -> TrackResult {
fn stop_current(&self) -> TrackResult<()> {
if let Some(handle) = self.tracks.front() {
handle.stop()
} else {