librespot/playback/src/player.rs

1597 lines
54 KiB
Rust
Raw Normal View History

2018-02-23 19:08:20 +00:00
use byteorder::{LittleEndian, ReadBytesExt};
2018-02-23 19:16:03 +00:00
use futures;
use futures::{future, Async, Future, Poll, Stream};
2018-02-10 15:26:08 +00:00
use std;
use std::borrow::Cow;
2019-11-11 07:22:41 +00:00
use std::cmp::max;
2018-02-26 01:50:41 +00:00
use std::io::{Read, Result, Seek, SeekFrom};
2017-01-29 14:11:20 +00:00
use std::mem;
2017-01-20 12:56:42 +00:00
use std::thread;
use std::time::{Duration, Instant};
2015-06-23 14:38:29 +00:00
use crate::config::{Bitrate, PlayerConfig};
2019-09-16 19:00:09 +00:00
use librespot_core::session::Session;
use librespot_core::spotify_id::SpotifyId;
2017-08-03 18:58:44 +00:00
use librespot_core::util::SeqGenerator;
use crate::audio::{AudioDecrypt, AudioFile, StreamLoaderController};
use crate::audio::{VorbisDecoder, VorbisPacket};
use crate::audio::{
2019-11-11 07:22:41 +00:00
READ_AHEAD_BEFORE_PLAYBACK_ROUNDTRIPS, READ_AHEAD_BEFORE_PLAYBACK_SECONDS,
READ_AHEAD_DURING_PLAYBACK_ROUNDTRIPS, READ_AHEAD_DURING_PLAYBACK_SECONDS,
};
use crate::audio_backend::Sink;
use crate::metadata::{AudioItem, FileFormat};
use crate::mixer::AudioFilter;
const PRELOAD_NEXT_TRACK_BEFORE_END_DURATION_MS: u32 = 30000;
2015-09-01 11:20:37 +00:00
pub struct Player {
commands: Option<futures::sync::mpsc::UnboundedSender<PlayerCommand>>,
thread_handle: Option<thread::JoinHandle<()>>,
play_request_id_generator: SeqGenerator<u64>,
}
2015-06-23 14:38:29 +00:00
2015-09-01 11:20:37 +00:00
struct PlayerInternal {
session: Session,
config: PlayerConfig,
commands: futures::sync::mpsc::UnboundedReceiver<PlayerCommand>,
2017-01-29 14:11:20 +00:00
state: PlayerState,
preload: PlayerPreload,
sink: Box<dyn Sink>,
sink_running: bool,
audio_filter: Option<Box<dyn AudioFilter + Send>>,
event_senders: Vec<futures::sync::mpsc::UnboundedSender<PlayerEvent>>,
}
2015-07-09 20:08:14 +00:00
enum PlayerCommand {
Load {
track_id: SpotifyId,
play_request_id: u64,
play: bool,
position_ms: u32,
},
Preload {
track_id: SpotifyId,
},
2015-07-09 20:08:14 +00:00
Play,
Pause,
Stop,
2016-01-02 15:19:39 +00:00
Seek(u32),
AddEventSender(futures::sync::mpsc::UnboundedSender<PlayerEvent>),
EmitVolumeSetEvent(u16),
}
2018-02-20 20:57:42 +00:00
#[derive(Debug, Clone)]
pub enum PlayerEvent {
Loading {
play_request_id: u64,
track_id: SpotifyId,
position_ms: u32,
},
2018-02-20 20:57:42 +00:00
Started {
play_request_id: u64,
2018-02-20 20:57:42 +00:00
track_id: SpotifyId,
position_ms: u32,
},
Playing {
play_request_id: u64,
track_id: SpotifyId,
position_ms: u32,
duration_ms: u32,
2018-02-20 20:57:42 +00:00
},
Changed {
old_track_id: SpotifyId,
new_track_id: SpotifyId,
},
TimeToPreloadNextTrack {
play_request_id: u64,
track_id: SpotifyId,
},
EndOfTrack {
play_request_id: u64,
track_id: SpotifyId,
},
Paused {
play_request_id: u64,
track_id: SpotifyId,
position_ms: u32,
duration_ms: u32,
},
2018-02-20 20:57:42 +00:00
Stopped {
play_request_id: u64,
2018-02-20 20:57:42 +00:00
track_id: SpotifyId,
2018-02-26 01:50:41 +00:00
},
VolumeSet {
volume: u16,
},
}
impl PlayerEvent {
pub fn get_play_request_id(&self) -> Option<u64> {
use PlayerEvent::*;
match self {
Loading {
play_request_id, ..
}
| Started {
play_request_id, ..
}
| Playing {
play_request_id, ..
}
| TimeToPreloadNextTrack {
play_request_id, ..
}
| EndOfTrack {
play_request_id, ..
}
| Paused {
play_request_id, ..
}
| Stopped {
play_request_id, ..
} => Some(*play_request_id),
Changed { .. } | VolumeSet { .. } => None,
}
}
2018-02-20 20:57:42 +00:00
}
pub type PlayerEventChannel = futures::sync::mpsc::UnboundedReceiver<PlayerEvent>;
2018-02-23 19:08:20 +00:00
#[derive(Clone, Copy, Debug)]
struct NormalisationData {
track_gain_db: f32,
track_peak: f32,
album_gain_db: f32,
album_peak: f32,
}
impl NormalisationData {
fn parse_from_file<T: Read + Seek>(mut file: T) -> Result<NormalisationData> {
const SPOTIFY_NORMALIZATION_HEADER_START_OFFSET: u64 = 144;
2018-02-26 01:50:41 +00:00
file.seek(SeekFrom::Start(SPOTIFY_NORMALIZATION_HEADER_START_OFFSET))
.unwrap();
2018-02-23 19:08:20 +00:00
let track_gain_db = file.read_f32::<LittleEndian>().unwrap();
let track_peak = file.read_f32::<LittleEndian>().unwrap();
let album_gain_db = file.read_f32::<LittleEndian>().unwrap();
let album_peak = file.read_f32::<LittleEndian>().unwrap();
2018-02-23 19:08:20 +00:00
let r = NormalisationData {
2018-02-23 19:08:20 +00:00
track_gain_db: track_gain_db,
track_peak: track_peak,
album_gain_db: album_gain_db,
album_peak: album_peak,
};
Ok(r)
2018-02-23 19:08:20 +00:00
}
fn get_factor(config: &PlayerConfig, data: NormalisationData) -> f32 {
let mut normalisation_factor = f32::powf(
10.0,
(data.track_gain_db + config.normalisation_pregain) / 20.0,
);
2018-02-23 19:08:20 +00:00
if normalisation_factor * data.track_peak > 1.0 {
warn!("Reducing normalisation factor to prevent clipping. Please add negative pregain to avoid.");
normalisation_factor = 1.0 / data.track_peak;
2018-02-23 19:08:20 +00:00
}
debug!("Normalisation Data: {:?}", data);
debug!("Applied normalisation factor: {}", normalisation_factor);
2018-02-23 19:08:20 +00:00
normalisation_factor
}
}
2015-09-01 11:20:37 +00:00
impl Player {
2018-02-26 01:50:41 +00:00
pub fn new<F>(
config: PlayerConfig,
session: Session,
audio_filter: Option<Box<dyn AudioFilter + Send>>,
2018-02-26 01:50:41 +00:00
sink_builder: F,
) -> (Player, PlayerEventChannel)
where
F: FnOnce() -> Box<dyn Sink> + Send + 'static,
{
let (cmd_tx, cmd_rx) = futures::sync::mpsc::unbounded();
2018-02-23 19:16:03 +00:00
let (event_sender, event_receiver) = futures::sync::mpsc::unbounded();
let handle = thread::spawn(move || {
debug!("new Player[{}]", session.session_id());
2017-01-29 14:11:20 +00:00
let internal = PlayerInternal {
session: session,
config: config,
2017-01-29 14:11:20 +00:00
commands: cmd_rx,
2017-01-29 14:11:20 +00:00
state: PlayerState::Stopped,
preload: PlayerPreload::None,
2017-01-29 14:11:20 +00:00
sink: sink_builder(),
sink_running: false,
audio_filter: audio_filter,
event_senders: [event_sender].to_vec(),
2017-01-29 14:11:20 +00:00
};
let _ = internal.wait();
debug!("PlayerInternal thread finished.");
2017-01-29 14:11:20 +00:00
});
2015-09-01 11:20:37 +00:00
2018-02-26 01:50:41 +00:00
(
Player {
commands: Some(cmd_tx),
thread_handle: Some(handle),
play_request_id_generator: SeqGenerator::new(0),
2018-02-26 01:50:41 +00:00
},
event_receiver,
)
}
2015-07-09 20:08:14 +00:00
fn command(&self, cmd: PlayerCommand) {
self.commands.as_ref().unwrap().unbounded_send(cmd).unwrap();
}
2016-01-20 14:11:49 +00:00
pub fn load(&mut self, track_id: SpotifyId, start_playing: bool, position_ms: u32) -> u64 {
let play_request_id = self.play_request_id_generator.get();
self.command(PlayerCommand::Load {
track_id,
play_request_id,
play: start_playing,
position_ms,
});
play_request_id
}
2017-01-29 14:11:20 +00:00
2020-02-02 00:07:05 +00:00
pub fn preload(&self, track_id: SpotifyId) {
self.command(PlayerCommand::Preload { track_id });
2016-01-20 14:11:49 +00:00
}
pub fn play(&self) {
self.command(PlayerCommand::Play)
}
pub fn pause(&self) {
self.command(PlayerCommand::Pause)
}
pub fn stop(&self) {
self.command(PlayerCommand::Stop)
}
pub fn seek(&self, position_ms: u32) {
self.command(PlayerCommand::Seek(position_ms));
}
pub fn get_player_event_channel(&self) -> PlayerEventChannel {
let (event_sender, event_receiver) = futures::sync::mpsc::unbounded();
self.command(PlayerCommand::AddEventSender(event_sender));
event_receiver
}
2020-02-02 00:07:05 +00:00
pub fn get_end_of_track_future(&self) -> Box<dyn Future<Item = (), Error = ()>> {
let result = self
.get_player_event_channel()
.filter(|event| match event {
PlayerEvent::EndOfTrack { .. } | PlayerEvent::Stopped { .. } => true,
_ => false,
})
.into_future()
.map_err(|_| ())
.map(|_| ());
Box::new(result)
}
pub fn emit_volume_set_event(&self, volume: u16) {
self.command(PlayerCommand::EmitVolumeSetEvent(volume));
}
}
2015-06-23 14:38:29 +00:00
impl Drop for Player {
fn drop(&mut self) {
debug!("Shutting down player thread ...");
self.commands = None;
if let Some(handle) = self.thread_handle.take() {
match handle.join() {
Ok(_) => (),
2018-02-26 01:50:41 +00:00
Err(_) => error!("Player thread panicked!"),
}
}
}
}
struct PlayerLoadedTrackData {
decoder: Decoder,
normalisation_factor: f32,
stream_loader_controller: StreamLoaderController,
bytes_per_second: usize,
duration_ms: u32,
2020-02-02 00:07:05 +00:00
stream_position_pcm: u64,
}
enum PlayerPreload {
None,
Loading {
track_id: SpotifyId,
loader: Box<dyn Future<Item = PlayerLoadedTrackData, Error = ()>>,
},
Ready {
track_id: SpotifyId,
loaded_track: PlayerLoadedTrackData,
},
}
type Decoder = VorbisDecoder<Subfile<AudioDecrypt<AudioFile>>>;
2017-01-29 14:11:20 +00:00
enum PlayerState {
Stopped,
Loading {
track_id: SpotifyId,
play_request_id: u64,
start_playback: bool,
loader: Box<dyn Future<Item = PlayerLoadedTrackData, Error = ()>>,
},
2017-01-29 14:11:20 +00:00
Paused {
track_id: SpotifyId,
play_request_id: u64,
2017-01-29 14:11:20 +00:00
decoder: Decoder,
2018-02-23 19:08:20 +00:00
normalisation_factor: f32,
2019-11-01 19:46:28 +00:00
stream_loader_controller: StreamLoaderController,
bytes_per_second: usize,
duration_ms: u32,
2020-02-02 00:07:05 +00:00
stream_position_pcm: u64,
suggested_to_preload_next_track: bool,
2017-01-29 14:11:20 +00:00
},
Playing {
track_id: SpotifyId,
play_request_id: u64,
2017-01-29 14:11:20 +00:00
decoder: Decoder,
2018-02-23 19:08:20 +00:00
normalisation_factor: f32,
2019-11-01 19:46:28 +00:00
stream_loader_controller: StreamLoaderController,
bytes_per_second: usize,
duration_ms: u32,
2020-02-02 00:07:05 +00:00
stream_position_pcm: u64,
reported_nominal_start_time: Option<Instant>,
suggested_to_preload_next_track: bool,
2017-01-29 14:11:20 +00:00
},
2018-02-26 01:50:41 +00:00
EndOfTrack {
track_id: SpotifyId,
play_request_id: u64,
loaded_track: Option<PlayerLoadedTrackData>,
2018-02-26 01:50:41 +00:00
},
2017-01-29 14:11:20 +00:00
Invalid,
}
2017-01-29 14:11:20 +00:00
impl PlayerState {
fn is_playing(&self) -> bool {
use self::PlayerState::*;
match *self {
Stopped | EndOfTrack { .. } | Paused { .. } | Loading { .. } => false,
2017-01-29 14:11:20 +00:00
Playing { .. } => true,
Invalid => panic!("invalid state"),
}
}
fn is_stopped(&self) -> bool {
use self::PlayerState::*;
match *self {
Stopped => true,
_ => false,
}
}
2017-01-29 14:11:20 +00:00
fn decoder(&mut self) -> Option<&mut Decoder> {
use self::PlayerState::*;
match *self {
Stopped | EndOfTrack { .. } | Loading { .. } => None,
Paused {
ref mut decoder, ..
}
| Playing {
ref mut decoder, ..
} => Some(decoder),
2017-01-29 14:11:20 +00:00
Invalid => panic!("invalid state"),
}
2017-01-29 14:11:20 +00:00
}
2019-11-01 19:46:28 +00:00
fn stream_loader_controller(&mut self) -> Option<&mut StreamLoaderController> {
use self::PlayerState::*;
match *self {
Stopped | EndOfTrack { .. } | Loading { .. } => None,
2019-11-11 07:22:41 +00:00
Paused {
ref mut stream_loader_controller,
..
}
| Playing {
ref mut stream_loader_controller,
..
} => Some(stream_loader_controller),
2019-11-01 19:46:28 +00:00
Invalid => panic!("invalid state"),
}
}
fn playing_to_end_of_track(&mut self) {
use self::PlayerState::*;
2018-02-23 19:16:03 +00:00
match mem::replace(self, Invalid) {
2018-02-26 01:50:41 +00:00
Playing {
track_id,
play_request_id,
decoder,
duration_ms,
bytes_per_second,
normalisation_factor,
stream_loader_controller,
stream_position_pcm,
2018-02-26 01:50:41 +00:00
..
} => {
*self = EndOfTrack {
track_id,
play_request_id,
loaded_track: Some(PlayerLoadedTrackData {
decoder,
duration_ms,
bytes_per_second,
normalisation_factor,
stream_loader_controller,
stream_position_pcm,
}),
};
2018-02-26 01:50:41 +00:00
}
_ => panic!("Called playing_to_end_of_track in non-playing state."),
}
2017-01-29 14:11:20 +00:00
}
2017-01-29 14:11:20 +00:00
fn paused_to_playing(&mut self) {
use self::PlayerState::*;
match ::std::mem::replace(self, Invalid) {
2018-02-26 01:50:41 +00:00
Paused {
track_id,
play_request_id,
2018-02-26 01:50:41 +00:00
decoder,
normalisation_factor,
2019-11-01 19:46:28 +00:00
stream_loader_controller,
duration_ms,
2019-11-11 07:22:41 +00:00
bytes_per_second,
2020-02-02 00:07:05 +00:00
stream_position_pcm,
suggested_to_preload_next_track,
2018-02-26 01:50:41 +00:00
} => {
2017-01-29 14:11:20 +00:00
*self = Playing {
track_id,
play_request_id,
decoder,
normalisation_factor,
stream_loader_controller,
duration_ms,
bytes_per_second,
2020-02-02 00:07:05 +00:00
stream_position_pcm,
reported_nominal_start_time: None,
suggested_to_preload_next_track,
2017-01-29 14:11:20 +00:00
};
}
_ => panic!("invalid state"),
}
}
2017-01-29 14:11:20 +00:00
fn playing_to_paused(&mut self) {
use self::PlayerState::*;
match ::std::mem::replace(self, Invalid) {
2018-02-26 01:50:41 +00:00
Playing {
track_id,
play_request_id,
2018-02-26 01:50:41 +00:00
decoder,
normalisation_factor,
2019-11-01 19:46:28 +00:00
stream_loader_controller,
duration_ms,
bytes_per_second,
2020-02-02 00:07:05 +00:00
stream_position_pcm,
reported_nominal_start_time: _,
suggested_to_preload_next_track,
2018-02-26 01:50:41 +00:00
} => {
2017-01-29 14:11:20 +00:00
*self = Paused {
track_id,
play_request_id,
decoder,
normalisation_factor,
stream_loader_controller,
duration_ms,
bytes_per_second,
2020-02-02 00:07:05 +00:00
stream_position_pcm,
suggested_to_preload_next_track,
2017-01-29 14:11:20 +00:00
};
}
_ => panic!("invalid state"),
}
}
}
struct PlayerTrackLoader {
session: Session,
config: PlayerConfig,
}
impl PlayerTrackLoader {
fn find_available_alternative<'a>(&self, audio: &'a AudioItem) -> Option<Cow<'a, AudioItem>> {
if audio.available {
Some(Cow::Borrowed(audio))
} else {
if let Some(alternatives) = &audio.alternatives {
let alternatives = alternatives
.iter()
.map(|alt_id| AudioItem::get_audio_item(&self.session, *alt_id));
let alternatives = future::join_all(alternatives).wait().unwrap();
alternatives
.into_iter()
.find(|alt| alt.available)
.map(Cow::Owned)
} else {
None
}
}
}
fn stream_data_rate(&self, format: FileFormat) -> usize {
match format {
FileFormat::OGG_VORBIS_96 => 12 * 1024,
FileFormat::OGG_VORBIS_160 => 20 * 1024,
FileFormat::OGG_VORBIS_320 => 40 * 1024,
FileFormat::MP3_256 => 32 * 1024,
FileFormat::MP3_320 => 40 * 1024,
FileFormat::MP3_160 => 20 * 1024,
FileFormat::MP3_96 => 12 * 1024,
FileFormat::MP3_160_ENC => 20 * 1024,
FileFormat::MP4_128_DUAL => 16 * 1024,
FileFormat::OTHER3 => 40 * 1024, // better some high guess than nothing
FileFormat::AAC_160 => 20 * 1024,
FileFormat::AAC_320 => 40 * 1024,
FileFormat::MP4_128 => 16 * 1024,
FileFormat::OTHER5 => 40 * 1024, // better some high guess than nothing
}
}
2020-02-02 00:07:05 +00:00
fn load_track(&self, spotify_id: SpotifyId, position_ms: u32) -> Option<PlayerLoadedTrackData> {
let audio = match AudioItem::get_audio_item(&self.session, spotify_id).wait() {
Ok(audio) => audio,
Err(_) => {
error!("Unable to load audio item.");
return None;
}
};
info!("Loading <{}> with Spotify URI <{}>", audio.name, audio.uri);
let audio = match self.find_available_alternative(&audio) {
Some(audio) => audio,
None => {
warn!("<{}> is not available", audio.uri);
return None;
}
};
assert!(audio.duration >= 0);
let duration_ms = audio.duration as u32;
// (Most) podcasts seem to support only 96 bit Vorbis, so fall back to it
let formats = match self.config.bitrate {
Bitrate::Bitrate96 => [
FileFormat::OGG_VORBIS_96,
FileFormat::OGG_VORBIS_160,
FileFormat::OGG_VORBIS_320,
],
Bitrate::Bitrate160 => [
FileFormat::OGG_VORBIS_160,
FileFormat::OGG_VORBIS_96,
FileFormat::OGG_VORBIS_320,
],
Bitrate::Bitrate320 => [
FileFormat::OGG_VORBIS_320,
FileFormat::OGG_VORBIS_160,
FileFormat::OGG_VORBIS_96,
],
};
let format = formats
.iter()
.find(|format| audio.files.contains_key(format))
.unwrap();
let file_id = match audio.files.get(&format) {
Some(&file_id) => file_id,
None => {
warn!("<{}> in not available in format {:?}", audio.name, format);
return None;
}
};
let bytes_per_second = self.stream_data_rate(*format);
2020-02-02 00:07:05 +00:00
let play_from_beginning = position_ms == 0;
let key = self.session.audio_key().request(spotify_id, file_id);
let encrypted_file = AudioFile::open(
&self.session,
file_id,
bytes_per_second,
play_from_beginning,
);
let encrypted_file = match encrypted_file.wait() {
Ok(encrypted_file) => encrypted_file,
Err(_) => {
error!("Unable to load encrypted file.");
return None;
}
};
let mut stream_loader_controller = encrypted_file.get_stream_loader_controller();
if play_from_beginning {
// No need to seek -> we stream from the beginning
stream_loader_controller.set_stream_mode();
} else {
// we need to seek -> we set stream mode after the initial seek.
stream_loader_controller.set_random_access_mode();
}
let key = match key.wait() {
Ok(key) => key,
Err(_) => {
error!("Unable to load decryption key");
return None;
}
};
let mut decrypted_file = AudioDecrypt::new(key, encrypted_file);
let normalisation_factor = match NormalisationData::parse_from_file(&mut decrypted_file) {
Ok(normalisation_data) => {
NormalisationData::get_factor(&self.config, normalisation_data)
}
Err(_) => {
warn!("Unable to extract normalisation data, using default value.");
1.0 as f32
}
};
let audio_file = Subfile::new(decrypted_file, 0xa7);
let mut decoder = VorbisDecoder::new(audio_file).unwrap();
2020-02-02 00:07:05 +00:00
if position_ms != 0 {
match decoder.seek(position_ms as i64) {
Ok(_) => (),
Err(err) => error!("Vorbis error: {:?}", err),
}
stream_loader_controller.set_stream_mode();
}
2020-02-02 00:07:05 +00:00
let stream_position_pcm = PlayerInternal::position_ms_to_pcm(position_ms);
info!("<{}> ({} ms) loaded", audio.name, audio.duration);
Some(PlayerLoadedTrackData {
decoder,
normalisation_factor,
stream_loader_controller,
bytes_per_second,
duration_ms,
2020-02-02 00:07:05 +00:00
stream_position_pcm,
})
}
}
impl Future for PlayerInternal {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<(), ()> {
loop {
let mut all_futures_completed_or_not_ready = true;
// process commands that were sent to us
let cmd = match self.commands.poll() {
Ok(Async::Ready(None)) => return Ok(Async::Ready(())), // client has disconnected - shut down.
Ok(Async::Ready(Some(cmd))) => {
all_futures_completed_or_not_ready = false;
Some(cmd)
2017-01-31 08:21:30 +00:00
}
Ok(Async::NotReady) => None,
Err(_) => None,
};
2017-01-29 14:11:20 +00:00
if let Some(cmd) = cmd {
self.handle_command(cmd);
}
// Handle loading of a new track to play
if let PlayerState::Loading {
ref mut loader,
track_id,
start_playback,
play_request_id,
} = self.state
{
match loader.poll() {
Ok(Async::Ready(loaded_track)) => {
self.start_playback(
track_id,
play_request_id,
loaded_track,
start_playback,
false,
);
if let PlayerState::Loading { .. } = self.state {
panic!("The state wasn't changed by start_playback()");
}
}
Ok(Async::NotReady) => (),
Err(_) => {
self.handle_player_stop();
assert!(self.state.is_stopped());
}
}
}
// handle pending preload requests.
if let PlayerPreload::Loading {
ref mut loader,
track_id,
} = self.preload
{
match loader.poll() {
Ok(Async::Ready(loaded_track)) => {
self.preload = PlayerPreload::Ready {
track_id,
loaded_track,
};
}
Ok(Async::NotReady) => (),
Err(_) => {
self.preload = PlayerPreload::None;
}
}
}
2020-02-02 23:11:27 +00:00
if self.state.is_playing() {
self.ensure_sink_running();
}
if self.sink_running {
2018-02-23 19:08:20 +00:00
let mut current_normalisation_factor: f32 = 1.0;
2018-02-26 01:50:41 +00:00
let packet = if let PlayerState::Playing {
track_id,
play_request_id,
2018-02-26 01:50:41 +00:00
ref mut decoder,
normalisation_factor,
2020-02-02 00:07:05 +00:00
ref mut stream_position_pcm,
ref mut reported_nominal_start_time,
duration_ms,
2018-02-26 01:50:41 +00:00
..
} = self.state
{
2018-02-23 19:08:20 +00:00
current_normalisation_factor = normalisation_factor;
let packet = decoder.next_packet().expect("Vorbis error");
if let Some(ref packet) = packet {
2020-02-02 00:07:05 +00:00
*stream_position_pcm =
*stream_position_pcm + (packet.data().len() / 2) as u64;
let stream_position_millis = Self::position_pcm_to_ms(*stream_position_pcm);
let notify_about_position = match *reported_nominal_start_time {
None => true,
Some(reported_nominal_start_time) => {
// only notify if we're behind. If we're ahead it's probably due to a buffer of the backend and we;re actually in time.
let lag = (Instant::now() - reported_nominal_start_time).as_millis()
as i64
- stream_position_millis as i64;
if lag > 1000 {
true
} else {
false
}
}
};
if notify_about_position {
*reported_nominal_start_time = Some(
Instant::now()
- Duration::from_millis(stream_position_millis as u64),
);
self.send_event(PlayerEvent::Playing {
track_id,
play_request_id,
position_ms: stream_position_millis as u32,
duration_ms,
});
}
}
Some(packet)
} else {
None
};
if let Some(packet) = packet {
2018-02-23 19:08:20 +00:00
self.handle_packet(packet, current_normalisation_factor);
}
2017-01-29 14:11:20 +00:00
}
2018-04-21 15:46:29 +00:00
if let PlayerState::Playing {
track_id,
play_request_id,
duration_ms,
2020-02-02 00:07:05 +00:00
stream_position_pcm,
ref mut stream_loader_controller,
ref mut suggested_to_preload_next_track,
..
}
| PlayerState::Paused {
track_id,
play_request_id,
duration_ms,
2020-02-02 00:07:05 +00:00
stream_position_pcm,
ref mut stream_loader_controller,
ref mut suggested_to_preload_next_track,
..
} = self.state
{
2020-02-02 00:07:05 +00:00
let stream_position_millis = Self::position_pcm_to_ms(stream_position_pcm);
if (!*suggested_to_preload_next_track)
&& ((duration_ms as i64 - stream_position_millis as i64)
< PRELOAD_NEXT_TRACK_BEFORE_END_DURATION_MS as i64)
&& stream_loader_controller.range_to_end_available()
{
*suggested_to_preload_next_track = true;
self.send_event(PlayerEvent::TimeToPreloadNextTrack {
track_id,
play_request_id,
});
}
}
2018-04-21 15:46:29 +00:00
if self.session.is_invalid() {
return Ok(Async::Ready(()));
}
if (!self.sink_running) && all_futures_completed_or_not_ready {
return Ok(Async::NotReady);
2018-04-21 15:46:29 +00:00
}
2017-01-29 14:11:20 +00:00
}
}
}
2016-01-02 15:19:39 +00:00
impl PlayerInternal {
2020-02-02 00:07:05 +00:00
fn position_pcm_to_ms(position_pcm: u64) -> u32 {
(position_pcm * 10 / 441) as u32
}
fn position_ms_to_pcm(position_ms: u32) -> u64 {
position_ms as u64 * 441 / 10
}
2020-02-02 23:11:27 +00:00
fn ensure_sink_running(&mut self) {
if !self.sink_running {
trace!("== Starting sink ==");
match self.sink.start() {
Ok(()) => self.sink_running = true,
Err(err) => error!("Could not start audio: {}", err),
}
}
}
2020-02-02 23:11:27 +00:00
fn ensure_sink_stopped(&mut self) {
if self.sink_running {
2020-02-02 23:11:27 +00:00
trace!("== Stopping sink ==");
self.sink.stop().unwrap();
self.sink_running = false;
}
}
fn handle_player_stop(&mut self) {
match self.state {
PlayerState::Playing {
track_id,
play_request_id,
..
}
| PlayerState::Paused {
track_id,
play_request_id,
..
}
| PlayerState::EndOfTrack {
track_id,
play_request_id,
..
}
| PlayerState::Loading {
track_id,
play_request_id,
..
} => {
2020-02-02 23:11:27 +00:00
self.ensure_sink_stopped();
self.send_event(PlayerEvent::Stopped {
track_id,
play_request_id,
});
self.state = PlayerState::Stopped;
}
PlayerState::Stopped => (),
PlayerState::Invalid => panic!("invalid state"),
}
}
fn handle_play(&mut self) {
if let PlayerState::Paused {
track_id,
play_request_id,
stream_position_pcm,
..
} = self.state
{
self.state.paused_to_playing();
let position_ms = Self::position_pcm_to_ms(stream_position_pcm);
self.send_event(PlayerEvent::Started {
track_id,
play_request_id,
position_ms,
});
2020-02-02 23:11:27 +00:00
self.ensure_sink_running();
} else {
warn!("Player::play called from invalid state");
}
}
fn handle_pause(&mut self) {
if let PlayerState::Playing {
track_id,
play_request_id,
stream_position_pcm,
duration_ms,
..
} = self.state
{
self.state.playing_to_paused();
2020-02-02 23:11:27 +00:00
self.ensure_sink_stopped();
let position_ms = Self::position_pcm_to_ms(stream_position_pcm);
self.send_event(PlayerEvent::Paused {
track_id,
play_request_id,
position_ms,
duration_ms,
});
} else {
warn!("Player::pause called from invalid state");
}
}
2018-02-23 19:08:20 +00:00
fn handle_packet(&mut self, packet: Option<VorbisPacket>, normalisation_factor: f32) {
2017-01-29 14:11:20 +00:00
match packet {
Some(mut packet) => {
if packet.data().len() > 0 {
if let Some(ref editor) = self.audio_filter {
editor.modify_stream(&mut packet.data_mut())
};
if self.config.normalisation && normalisation_factor != 1.0 {
for x in packet.data_mut().iter_mut() {
*x = (*x as f32 * normalisation_factor) as i16;
}
2018-02-23 19:08:20 +00:00
}
if let Err(err) = self.sink.write(&packet.data()) {
error!("Could not write audio: {}", err);
2020-02-02 23:11:27 +00:00
self.ensure_sink_stopped();
}
}
2017-01-29 14:11:20 +00:00
}
None => {
self.state.playing_to_end_of_track();
if let PlayerState::EndOfTrack {
track_id,
play_request_id,
..
} = self.state
{
self.send_event(PlayerEvent::EndOfTrack {
track_id,
play_request_id,
})
} else {
unreachable!();
}
}
}
}
fn start_playback(
&mut self,
track_id: SpotifyId,
play_request_id: u64,
loaded_track: PlayerLoadedTrackData,
start_playback: bool,
state_is_invalid_because_the_same_track_is_getting_repeated: bool,
) {
2020-02-02 00:07:05 +00:00
let position_ms = Self::position_pcm_to_ms(loaded_track.stream_position_pcm);
match self.state {
PlayerState::Playing {
track_id: old_track_id,
..
}
| PlayerState::Paused {
track_id: old_track_id,
..
2017-01-29 14:11:20 +00:00
}
| PlayerState::EndOfTrack {
track_id: old_track_id,
..
} => self.send_event(PlayerEvent::Changed {
old_track_id: old_track_id,
new_track_id: track_id,
}),
PlayerState::Stopped => self.send_event(PlayerEvent::Started {
track_id,
play_request_id,
position_ms,
}),
PlayerState::Loading { .. } => (),
PlayerState::Invalid { .. } => {
if state_is_invalid_because_the_same_track_is_getting_repeated {
self.send_event(PlayerEvent::Changed {
old_track_id: track_id,
new_track_id: track_id,
})
} else {
panic!("Player is in an invalid state.")
}
}
}
if start_playback {
2020-02-02 23:11:27 +00:00
self.ensure_sink_running();
self.send_event(PlayerEvent::Playing {
track_id,
play_request_id,
position_ms,
duration_ms: loaded_track.duration_ms,
});
self.state = PlayerState::Playing {
track_id: track_id,
play_request_id: play_request_id,
decoder: loaded_track.decoder,
normalisation_factor: loaded_track.normalisation_factor,
stream_loader_controller: loaded_track.stream_loader_controller,
duration_ms: loaded_track.duration_ms,
bytes_per_second: loaded_track.bytes_per_second,
2020-02-02 00:07:05 +00:00
stream_position_pcm: loaded_track.stream_position_pcm,
reported_nominal_start_time: Some(
Instant::now() - Duration::from_millis(position_ms as u64),
),
suggested_to_preload_next_track: false,
};
} else {
self.state = PlayerState::Paused {
track_id: track_id,
play_request_id: play_request_id,
decoder: loaded_track.decoder,
normalisation_factor: loaded_track.normalisation_factor,
stream_loader_controller: loaded_track.stream_loader_controller,
duration_ms: loaded_track.duration_ms,
bytes_per_second: loaded_track.bytes_per_second,
2020-02-02 00:07:05 +00:00
stream_position_pcm: loaded_track.stream_position_pcm,
suggested_to_preload_next_track: false,
};
self.send_event(PlayerEvent::Paused {
track_id,
play_request_id,
position_ms,
duration_ms: loaded_track.duration_ms,
});
2017-01-29 14:11:20 +00:00
}
}
2017-01-29 14:11:20 +00:00
fn handle_command(&mut self, cmd: PlayerCommand) {
debug!("command={:?}", cmd);
2017-01-29 14:11:20 +00:00
match cmd {
PlayerCommand::Load {
track_id,
play_request_id,
play,
position_ms,
} => {
// emit the correct player event
match self.state {
PlayerState::Playing {
track_id: old_track_id,
..
}
| PlayerState::Paused {
track_id: old_track_id,
..
}
| PlayerState::EndOfTrack {
track_id: old_track_id,
..
}
| PlayerState::Loading {
track_id: old_track_id,
..
} => self.send_event(PlayerEvent::Changed {
old_track_id: old_track_id,
new_track_id: track_id,
}),
PlayerState::Stopped => self.send_event(PlayerEvent::Started {
track_id,
play_request_id,
position_ms,
}),
PlayerState::Invalid { .. } => panic!("Player is in an invalid state."),
}
let mut load_command_processed = false;
// Check if there's a matching loaded track in the EndOfTrack player state.
// This is the case if we're repeating the same track again.
if let PlayerState::EndOfTrack {
track_id: previous_track_id,
ref mut loaded_track,
..
} = self.state
{
if previous_track_id == track_id {
let loaded_track = mem::replace(&mut *loaded_track, None);
if let Some(mut loaded_track) = loaded_track {
if Self::position_ms_to_pcm(position_ms)
!= loaded_track.stream_position_pcm
{
loaded_track
.stream_loader_controller
.set_random_access_mode();
let _ = loaded_track.decoder.seek(position_ms as i64); // This may be blocking.
// But most likely the track is fully
// loaded already because we played
// to the end of it.
loaded_track.stream_loader_controller.set_stream_mode();
loaded_track.stream_position_pcm =
Self::position_ms_to_pcm(position_ms);
}
self.start_playback(
track_id,
play_request_id,
loaded_track,
play,
false,
);
load_command_processed = true;
}
}
}
2017-01-29 14:11:20 +00:00
// Check if we are already playing the track. If so, just do a seek and update our info.
if let PlayerState::Playing {
track_id: current_track_id,
ref mut stream_position_pcm,
ref mut decoder,
ref mut stream_loader_controller,
..
}
| PlayerState::Paused {
track_id: current_track_id,
ref mut stream_position_pcm,
ref mut decoder,
ref mut stream_loader_controller,
..
} = self.state
{
if current_track_id == track_id {
if Self::position_ms_to_pcm(position_ms) != *stream_position_pcm {
stream_loader_controller.set_random_access_mode();
let _ = decoder.seek(position_ms as i64); // This may be blocking.
stream_loader_controller.set_stream_mode();
*stream_position_pcm = Self::position_ms_to_pcm(position_ms);
}
let old_state = mem::replace(&mut self.state, PlayerState::Invalid);
if let PlayerState::Playing {
stream_position_pcm,
decoder,
stream_loader_controller,
bytes_per_second,
duration_ms,
normalisation_factor,
..
}
| PlayerState::Paused {
stream_position_pcm,
decoder,
stream_loader_controller,
bytes_per_second,
duration_ms,
normalisation_factor,
..
} = old_state
{
let loaded_track = PlayerLoadedTrackData {
decoder,
normalisation_factor,
stream_loader_controller,
bytes_per_second,
duration_ms,
stream_position_pcm,
};
self.start_playback(
track_id,
play_request_id,
loaded_track,
play,
true,
);
load_command_processed = true;
} else {
unreachable!();
}
}
}
// Check if the requested track has been preloaded already. If so use the preloaded data.
if !load_command_processed {
if let PlayerPreload::Ready {
track_id: loaded_track_id,
..
} = self.preload
{
if track_id == loaded_track_id {
let preload = std::mem::replace(&mut self.preload, PlayerPreload::None);
if let PlayerPreload::Ready {
track_id,
mut loaded_track,
} = preload
{
if Self::position_ms_to_pcm(position_ms)
!= loaded_track.stream_position_pcm
{
loaded_track
.stream_loader_controller
.set_random_access_mode();
let _ = loaded_track.decoder.seek(position_ms as i64); // This may be blocking
loaded_track.stream_loader_controller.set_stream_mode();
}
self.start_playback(
track_id,
play_request_id,
loaded_track,
play,
false,
);
load_command_processed = true;
}
}
}
}
// We need to load the track - either from scratch or by completing a preload.
// In any case we go into a Loading state to load the track.
if !load_command_processed {
2020-02-02 23:11:27 +00:00
self.ensure_sink_stopped();
self.send_event(PlayerEvent::Loading {
track_id,
play_request_id,
position_ms,
});
let loader = if let PlayerPreload::Loading {
track_id: loaded_track_id,
..
} = self.preload
{
if (track_id == loaded_track_id) && (position_ms == 0) {
let mut preload = PlayerPreload::None;
std::mem::swap(&mut preload, &mut self.preload);
if let PlayerPreload::Loading { loader, .. } = preload {
Some(loader)
} else {
None
}
} else {
None
2017-01-29 14:11:20 +00:00
}
} else {
None
};
self.preload = PlayerPreload::None;
let loader = loader
.or_else(|| Some(self.load_track(track_id, position_ms)))
.unwrap();
2016-01-02 15:48:44 +00:00
self.state = PlayerState::Loading {
track_id,
play_request_id,
start_playback: play,
loader,
};
}
}
PlayerCommand::Preload { track_id } => {
2020-02-02 22:15:15 +00:00
debug!("Preloading track");
let mut preload_track = true;
if let PlayerPreload::Loading {
track_id: currently_loading,
..
}
| PlayerPreload::Ready {
track_id: currently_loading,
..
} = self.preload
{
if currently_loading == track_id {
// we're already loading the requested track.
preload_track = false;
} else {
// we're loading something else - cancel it.
self.preload = PlayerPreload::None;
2017-01-29 14:11:20 +00:00
}
2016-01-02 15:19:39 +00:00
}
if let PlayerState::Playing {
track_id: current_track_id,
..
}
| PlayerState::Paused {
track_id: current_track_id,
..
}
| PlayerState::EndOfTrack {
track_id: current_track_id,
..
} = self.state
{
if current_track_id == track_id {
// we already have the requested track loaded.
preload_track = false;
}
}
if preload_track {
2020-02-02 00:07:05 +00:00
let loader = self.load_track(track_id, 0);
self.preload = PlayerPreload::Loading { track_id, loader }
}
2017-01-29 14:11:20 +00:00
}
2020-02-02 00:07:05 +00:00
PlayerCommand::Seek(position_ms) => {
2019-11-01 19:46:28 +00:00
if let Some(stream_loader_controller) = self.state.stream_loader_controller() {
stream_loader_controller.set_random_access_mode();
}
2017-01-29 14:11:20 +00:00
if let Some(decoder) = self.state.decoder() {
2020-02-02 00:07:05 +00:00
match decoder.seek(position_ms as i64) {
Ok(_) => {
if let PlayerState::Playing {
2020-02-02 00:07:05 +00:00
ref mut stream_position_pcm,
..
}
| PlayerState::Paused {
2020-02-02 00:07:05 +00:00
ref mut stream_position_pcm,
..
} = self.state
{
2020-02-02 00:07:05 +00:00
*stream_position_pcm = Self::position_ms_to_pcm(position_ms);
}
}
Err(err) => error!("Vorbis error: {:?}", err),
}
2017-01-29 14:11:20 +00:00
} else {
warn!("Player::seek called from invalid state");
2016-01-02 15:19:39 +00:00
}
2019-11-01 19:46:28 +00:00
// If we're playing, ensure, that we have enough data leaded to avoid a buffer underrun.
if let Some(stream_loader_controller) = self.state.stream_loader_controller() {
stream_loader_controller.set_stream_mode();
2019-11-01 22:22:07 +00:00
}
self.preload_data_before_playback();
2020-01-17 17:11:07 +00:00
if let PlayerState::Playing {
track_id,
play_request_id,
ref mut reported_nominal_start_time,
duration_ms,
..
2020-01-17 17:11:07 +00:00
} = self.state
{
*reported_nominal_start_time =
2020-02-02 00:07:05 +00:00
Some(Instant::now() - Duration::from_millis(position_ms as u64));
self.send_event(PlayerEvent::Playing {
track_id,
play_request_id,
2020-02-02 00:07:05 +00:00
position_ms: position_ms,
duration_ms,
});
}
if let PlayerState::Paused {
track_id,
play_request_id,
duration_ms,
..
} = self.state
{
self.send_event(PlayerEvent::Paused {
track_id,
play_request_id,
2020-02-02 00:07:05 +00:00
position_ms: position_ms,
duration_ms,
});
2019-11-01 19:46:28 +00:00
}
}
2017-01-29 14:11:20 +00:00
PlayerCommand::Play => {
self.handle_play();
2017-01-29 14:11:20 +00:00
}
2017-01-29 14:11:20 +00:00
PlayerCommand::Pause => {
self.handle_pause();
2017-01-29 14:11:20 +00:00
}
PlayerCommand::Stop => self.handle_player_stop(),
2015-06-23 14:38:29 +00:00
PlayerCommand::AddEventSender(sender) => self.event_senders.push(sender),
2016-01-20 13:55:36 +00:00
PlayerCommand::EmitVolumeSetEvent(volume) => {
self.send_event(PlayerEvent::VolumeSet { volume })
}
2017-01-29 14:11:20 +00:00
}
}
2015-07-09 20:08:14 +00:00
fn send_event(&mut self, event: PlayerEvent) {
let mut index = 0;
while index < self.event_senders.len() {
match self.event_senders[index].unbounded_send(event.clone()) {
Ok(_) => index += 1,
Err(_) => {
self.event_senders.remove(index);
}
}
2019-11-01 19:46:28 +00:00
}
}
2020-02-02 00:07:05 +00:00
fn load_track(
2019-11-11 07:22:41 +00:00
&self,
spotify_id: SpotifyId,
2020-02-02 00:07:05 +00:00
position_ms: u32,
) -> Box<dyn Future<Item = PlayerLoadedTrackData, Error = ()>> {
// This method creates a future that returns the loaded stream and associated info.
// Ideally all work should be done using asynchronous code. However, seek() on the
// audio stream is implemented in a blocking fashion. Thus, we can't turn it into future
// easily. Instead we spawn a thread to do the work and return a one-shot channel as the
// future to work with.
let loader = PlayerTrackLoader {
session: self.session.clone(),
config: self.config.clone(),
};
let (result_tx, result_rx) = futures::sync::oneshot::channel();
2019-11-01 19:48:18 +00:00
std::thread::spawn(move || {
loader
2020-02-02 00:07:05 +00:00
.load_track(spotify_id, position_ms)
.and_then(move |data| {
let _ = result_tx.send(data);
Some(())
});
});
2017-01-29 14:11:20 +00:00
Box::new(result_rx.map_err(|_| ()))
}
fn preload_data_before_playback(&mut self) {
if let PlayerState::Playing {
2020-01-17 17:11:07 +00:00
bytes_per_second,
ref mut stream_loader_controller,
..
} = self.state
{
// Request our read ahead range
let request_data_length = max(
(READ_AHEAD_DURING_PLAYBACK_ROUNDTRIPS
* (0.001 * stream_loader_controller.ping_time_ms() as f64)
* bytes_per_second as f64) as usize,
(READ_AHEAD_DURING_PLAYBACK_SECONDS * bytes_per_second as f64) as usize,
);
stream_loader_controller.fetch_next(request_data_length);
// Request the part we want to wait for blocking. This effecively means we wait for the previous request to partially complete.
let wait_for_data_length = max(
(READ_AHEAD_BEFORE_PLAYBACK_ROUNDTRIPS
* (0.001 * stream_loader_controller.ping_time_ms() as f64)
* bytes_per_second as f64) as usize,
(READ_AHEAD_BEFORE_PLAYBACK_SECONDS * bytes_per_second as f64) as usize,
);
stream_loader_controller.fetch_next_blocking(wait_for_data_length);
2019-11-01 19:46:28 +00:00
}
}
2015-07-09 20:08:14 +00:00
}
2017-01-29 14:11:20 +00:00
impl Drop for PlayerInternal {
fn drop(&mut self) {
debug!("drop PlayerInternal[{}]", self.session.session_id());
}
}
impl ::std::fmt::Debug for PlayerCommand {
fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
match *self {
PlayerCommand::Load {
track_id,
play,
position_ms,
..
} => f
2018-07-03 11:08:42 +00:00
.debug_tuple("Load")
.field(&track_id)
2018-02-26 01:50:41 +00:00
.field(&play)
.field(&position_ms)
2018-02-26 01:50:41 +00:00
.finish(),
PlayerCommand::Preload { track_id } => {
f.debug_tuple("Preload").field(&track_id).finish()
}
2018-02-26 01:50:41 +00:00
PlayerCommand::Play => f.debug_tuple("Play").finish(),
PlayerCommand::Pause => f.debug_tuple("Pause").finish(),
PlayerCommand::Stop => f.debug_tuple("Stop").finish(),
PlayerCommand::Seek(position) => f.debug_tuple("Seek").field(&position).finish(),
PlayerCommand::AddEventSender(_) => f.debug_tuple("AddEventSender").finish(),
PlayerCommand::EmitVolumeSetEvent(volume) => {
f.debug_tuple("VolumeSet").field(&volume).finish()
}
}
}
}
2018-02-10 15:26:08 +00:00
struct Subfile<T: Read + Seek> {
stream: T,
offset: u64,
}
impl<T: Read + Seek> Subfile<T> {
pub fn new(mut stream: T, offset: u64) -> Subfile<T> {
stream.seek(SeekFrom::Start(offset)).unwrap();
Subfile {
stream: stream,
offset: offset,
}
}
}
impl<T: Read + Seek> Read for Subfile<T> {
fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
self.stream.read(buf)
}
}
impl<T: Read + Seek> Seek for Subfile<T> {
fn seek(&mut self, mut pos: SeekFrom) -> Result<u64> {
pos = match pos {
SeekFrom::Start(offset) => SeekFrom::Start(offset + self.offset),
x => x,
};
let newpos = self.stream.seek(pos)?;
2018-02-10 15:26:08 +00:00
if newpos > self.offset {
Ok(newpos - self.offset)
} else {
Ok(0)
}
}
}