player.rs 56 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711
  1. use byteorder::{LittleEndian, ReadBytesExt};
  2. use futures;
  3. use futures::{future, Async, Future, Poll, Stream};
  4. use std;
  5. use std::borrow::Cow;
  6. use std::cmp::max;
  7. use std::io::{Read, Result, Seek, SeekFrom};
  8. use std::mem;
  9. use std::thread;
  10. use std::time::{Duration, Instant};
  11. use crate::config::{Bitrate, PlayerConfig};
  12. use librespot_core::session::Session;
  13. use librespot_core::spotify_id::SpotifyId;
  14. use librespot_core::util::SeqGenerator;
  15. use crate::audio::{AudioDecrypt, AudioFile, StreamLoaderController};
  16. use crate::audio::{VorbisDecoder, VorbisPacket};
  17. use crate::audio::{
  18. READ_AHEAD_BEFORE_PLAYBACK_ROUNDTRIPS, READ_AHEAD_BEFORE_PLAYBACK_SECONDS,
  19. READ_AHEAD_DURING_PLAYBACK_ROUNDTRIPS, READ_AHEAD_DURING_PLAYBACK_SECONDS,
  20. };
  21. use crate::audio_backend::Sink;
  22. use crate::metadata::{AudioItem, FileFormat};
  23. use crate::mixer::AudioFilter;
  24. const PRELOAD_NEXT_TRACK_BEFORE_END_DURATION_MS: u32 = 30000;
  25. pub struct Player {
  26. commands: Option<futures::sync::mpsc::UnboundedSender<PlayerCommand>>,
  27. thread_handle: Option<thread::JoinHandle<()>>,
  28. play_request_id_generator: SeqGenerator<u64>,
  29. }
  30. #[derive(PartialEq, Debug, Clone, Copy)]
  31. pub enum SinkStatus {
  32. Running,
  33. Closed,
  34. TemporarilyClosed,
  35. }
  36. pub type SinkEventCallback = Box<dyn Fn(SinkStatus) + Send>;
  37. struct PlayerInternal {
  38. session: Session,
  39. config: PlayerConfig,
  40. commands: futures::sync::mpsc::UnboundedReceiver<PlayerCommand>,
  41. state: PlayerState,
  42. preload: PlayerPreload,
  43. sink: Box<dyn Sink>,
  44. sink_status: SinkStatus,
  45. sink_event_callback: Option<SinkEventCallback>,
  46. audio_filter: Option<Box<dyn AudioFilter + Send>>,
  47. event_senders: Vec<futures::sync::mpsc::UnboundedSender<PlayerEvent>>,
  48. }
  49. enum PlayerCommand {
  50. Load {
  51. track_id: SpotifyId,
  52. play_request_id: u64,
  53. play: bool,
  54. position_ms: u32,
  55. },
  56. Preload {
  57. track_id: SpotifyId,
  58. },
  59. Play,
  60. Pause,
  61. Stop,
  62. Seek(u32),
  63. AddEventSender(futures::sync::mpsc::UnboundedSender<PlayerEvent>),
  64. SetSinkEventCallback(Option<SinkEventCallback>),
  65. EmitVolumeSetEvent(u16),
  66. }
  67. #[derive(Debug, Clone)]
  68. pub enum PlayerEvent {
  69. // Fired when the player is stopped (e.g. by issuing a "stop" command to the player).
  70. Stopped {
  71. play_request_id: u64,
  72. track_id: SpotifyId,
  73. },
  74. // The player started working on playback of a track while it was in a stopped state.
  75. // This is always immediately followed up by a "Loading" or "Playing" event.
  76. Started {
  77. play_request_id: u64,
  78. track_id: SpotifyId,
  79. position_ms: u32,
  80. },
  81. // Same as started but in the case that the player already had a track loaded.
  82. // The player was either playing the loaded track or it was paused.
  83. Changed {
  84. old_track_id: SpotifyId,
  85. new_track_id: SpotifyId,
  86. },
  87. // The player is delayed by loading a track.
  88. Loading {
  89. play_request_id: u64,
  90. track_id: SpotifyId,
  91. position_ms: u32,
  92. },
  93. // The player is preloading a track.
  94. Preloading {
  95. track_id: SpotifyId,
  96. },
  97. // The player is playing a track.
  98. // This event is issued at the start of playback of whenever the position must be communicated
  99. // because it is out of sync. This includes:
  100. // start of a track
  101. // un-pausing
  102. // after a seek
  103. // after a buffer-underrun
  104. Playing {
  105. play_request_id: u64,
  106. track_id: SpotifyId,
  107. position_ms: u32,
  108. duration_ms: u32,
  109. },
  110. // The player entered a paused state.
  111. Paused {
  112. play_request_id: u64,
  113. track_id: SpotifyId,
  114. position_ms: u32,
  115. duration_ms: u32,
  116. },
  117. // The player thinks it's a good idea to issue a preload command for the next track now.
  118. // This event is intended for use within spirc.
  119. TimeToPreloadNextTrack {
  120. play_request_id: u64,
  121. track_id: SpotifyId,
  122. },
  123. // The player reached the end of a track.
  124. // This event is intended for use within spirc. Spirc will respond by issuing another command
  125. // which will trigger another event (e.g. Changed or Stopped)
  126. EndOfTrack {
  127. play_request_id: u64,
  128. track_id: SpotifyId,
  129. },
  130. // The player was unable to load the requested track.
  131. Unavailable {
  132. play_request_id: u64,
  133. track_id: SpotifyId,
  134. },
  135. // The mixer volume was set to a new level.
  136. VolumeSet {
  137. volume: u16,
  138. },
  139. }
  140. impl PlayerEvent {
  141. pub fn get_play_request_id(&self) -> Option<u64> {
  142. use PlayerEvent::*;
  143. match self {
  144. Loading {
  145. play_request_id, ..
  146. }
  147. | Unavailable {
  148. play_request_id, ..
  149. }
  150. | Started {
  151. play_request_id, ..
  152. }
  153. | Playing {
  154. play_request_id, ..
  155. }
  156. | TimeToPreloadNextTrack {
  157. play_request_id, ..
  158. }
  159. | EndOfTrack {
  160. play_request_id, ..
  161. }
  162. | Paused {
  163. play_request_id, ..
  164. }
  165. | Stopped {
  166. play_request_id, ..
  167. } => Some(*play_request_id),
  168. Changed { .. } | Preloading { .. } | VolumeSet { .. } => None,
  169. }
  170. }
  171. }
  172. pub type PlayerEventChannel = futures::sync::mpsc::UnboundedReceiver<PlayerEvent>;
  173. #[derive(Clone, Copy, Debug)]
  174. struct NormalisationData {
  175. track_gain_db: f32,
  176. track_peak: f32,
  177. album_gain_db: f32,
  178. album_peak: f32,
  179. }
  180. impl NormalisationData {
  181. fn parse_from_file<T: Read + Seek>(mut file: T) -> Result<NormalisationData> {
  182. const SPOTIFY_NORMALIZATION_HEADER_START_OFFSET: u64 = 144;
  183. file.seek(SeekFrom::Start(SPOTIFY_NORMALIZATION_HEADER_START_OFFSET))
  184. .unwrap();
  185. let track_gain_db = file.read_f32::<LittleEndian>().unwrap();
  186. let track_peak = file.read_f32::<LittleEndian>().unwrap();
  187. let album_gain_db = file.read_f32::<LittleEndian>().unwrap();
  188. let album_peak = file.read_f32::<LittleEndian>().unwrap();
  189. let r = NormalisationData {
  190. track_gain_db: track_gain_db,
  191. track_peak: track_peak,
  192. album_gain_db: album_gain_db,
  193. album_peak: album_peak,
  194. };
  195. Ok(r)
  196. }
  197. fn get_factor(config: &PlayerConfig, data: NormalisationData) -> f32 {
  198. let mut normalisation_factor = f32::powf(
  199. 10.0,
  200. (data.track_gain_db + config.normalisation_pregain) / 20.0,
  201. );
  202. if normalisation_factor * data.track_peak > 1.0 {
  203. warn!("Reducing normalisation factor to prevent clipping. Please add negative pregain to avoid.");
  204. normalisation_factor = 1.0 / data.track_peak;
  205. }
  206. debug!("Normalisation Data: {:?}", data);
  207. debug!("Applied normalisation factor: {}", normalisation_factor);
  208. normalisation_factor
  209. }
  210. }
  211. impl Player {
  212. pub fn new<F>(
  213. config: PlayerConfig,
  214. session: Session,
  215. audio_filter: Option<Box<dyn AudioFilter + Send>>,
  216. sink_builder: F,
  217. ) -> (Player, PlayerEventChannel)
  218. where
  219. F: FnOnce() -> Box<dyn Sink> + Send + 'static,
  220. {
  221. let (cmd_tx, cmd_rx) = futures::sync::mpsc::unbounded();
  222. let (event_sender, event_receiver) = futures::sync::mpsc::unbounded();
  223. let handle = thread::spawn(move || {
  224. debug!("new Player[{}]", session.session_id());
  225. let internal = PlayerInternal {
  226. session: session,
  227. config: config,
  228. commands: cmd_rx,
  229. state: PlayerState::Stopped,
  230. preload: PlayerPreload::None,
  231. sink: sink_builder(),
  232. sink_status: SinkStatus::Closed,
  233. sink_event_callback: None,
  234. audio_filter: audio_filter,
  235. event_senders: [event_sender].to_vec(),
  236. };
  237. // While PlayerInternal is written as a future, it still contains blocking code.
  238. // It must be run by using wait() in a dedicated thread.
  239. let _ = internal.wait();
  240. debug!("PlayerInternal thread finished.");
  241. });
  242. (
  243. Player {
  244. commands: Some(cmd_tx),
  245. thread_handle: Some(handle),
  246. play_request_id_generator: SeqGenerator::new(0),
  247. },
  248. event_receiver,
  249. )
  250. }
  251. fn command(&self, cmd: PlayerCommand) {
  252. self.commands.as_ref().unwrap().unbounded_send(cmd).unwrap();
  253. }
  254. pub fn load(&mut self, track_id: SpotifyId, start_playing: bool, position_ms: u32) -> u64 {
  255. let play_request_id = self.play_request_id_generator.get();
  256. self.command(PlayerCommand::Load {
  257. track_id,
  258. play_request_id,
  259. play: start_playing,
  260. position_ms,
  261. });
  262. play_request_id
  263. }
  264. pub fn preload(&self, track_id: SpotifyId) {
  265. self.command(PlayerCommand::Preload { track_id });
  266. }
  267. pub fn play(&self) {
  268. self.command(PlayerCommand::Play)
  269. }
  270. pub fn pause(&self) {
  271. self.command(PlayerCommand::Pause)
  272. }
  273. pub fn stop(&self) {
  274. self.command(PlayerCommand::Stop)
  275. }
  276. pub fn seek(&self, position_ms: u32) {
  277. self.command(PlayerCommand::Seek(position_ms));
  278. }
  279. pub fn get_player_event_channel(&self) -> PlayerEventChannel {
  280. let (event_sender, event_receiver) = futures::sync::mpsc::unbounded();
  281. self.command(PlayerCommand::AddEventSender(event_sender));
  282. event_receiver
  283. }
  284. pub fn get_end_of_track_future(&self) -> Box<dyn Future<Item = (), Error = ()>> {
  285. let result = self
  286. .get_player_event_channel()
  287. .filter(|event| match event {
  288. PlayerEvent::EndOfTrack { .. } | PlayerEvent::Stopped { .. } => true,
  289. _ => false,
  290. })
  291. .into_future()
  292. .map_err(|_| ())
  293. .map(|_| ());
  294. Box::new(result)
  295. }
  296. pub fn set_sink_event_callback(&self, callback: Option<SinkEventCallback>) {
  297. self.command(PlayerCommand::SetSinkEventCallback(callback));
  298. }
  299. pub fn emit_volume_set_event(&self, volume: u16) {
  300. self.command(PlayerCommand::EmitVolumeSetEvent(volume));
  301. }
  302. }
  303. impl Drop for Player {
  304. fn drop(&mut self) {
  305. debug!("Shutting down player thread ...");
  306. self.commands = None;
  307. if let Some(handle) = self.thread_handle.take() {
  308. match handle.join() {
  309. Ok(_) => (),
  310. Err(_) => error!("Player thread panicked!"),
  311. }
  312. }
  313. }
  314. }
  315. struct PlayerLoadedTrackData {
  316. decoder: Decoder,
  317. normalisation_factor: f32,
  318. stream_loader_controller: StreamLoaderController,
  319. bytes_per_second: usize,
  320. duration_ms: u32,
  321. stream_position_pcm: u64,
  322. }
  323. enum PlayerPreload {
  324. None,
  325. Loading {
  326. track_id: SpotifyId,
  327. loader: Box<dyn Future<Item = PlayerLoadedTrackData, Error = ()>>,
  328. },
  329. Ready {
  330. track_id: SpotifyId,
  331. loaded_track: PlayerLoadedTrackData,
  332. },
  333. }
  334. type Decoder = VorbisDecoder<Subfile<AudioDecrypt<AudioFile>>>;
  335. enum PlayerState {
  336. Stopped,
  337. Loading {
  338. track_id: SpotifyId,
  339. play_request_id: u64,
  340. start_playback: bool,
  341. loader: Box<dyn Future<Item = PlayerLoadedTrackData, Error = ()>>,
  342. },
  343. Paused {
  344. track_id: SpotifyId,
  345. play_request_id: u64,
  346. decoder: Decoder,
  347. normalisation_factor: f32,
  348. stream_loader_controller: StreamLoaderController,
  349. bytes_per_second: usize,
  350. duration_ms: u32,
  351. stream_position_pcm: u64,
  352. suggested_to_preload_next_track: bool,
  353. },
  354. Playing {
  355. track_id: SpotifyId,
  356. play_request_id: u64,
  357. decoder: Decoder,
  358. normalisation_factor: f32,
  359. stream_loader_controller: StreamLoaderController,
  360. bytes_per_second: usize,
  361. duration_ms: u32,
  362. stream_position_pcm: u64,
  363. reported_nominal_start_time: Option<Instant>,
  364. suggested_to_preload_next_track: bool,
  365. },
  366. EndOfTrack {
  367. track_id: SpotifyId,
  368. play_request_id: u64,
  369. loaded_track: PlayerLoadedTrackData,
  370. },
  371. Invalid,
  372. }
  373. impl PlayerState {
  374. fn is_playing(&self) -> bool {
  375. use self::PlayerState::*;
  376. match *self {
  377. Stopped | EndOfTrack { .. } | Paused { .. } | Loading { .. } => false,
  378. Playing { .. } => true,
  379. Invalid => panic!("invalid state"),
  380. }
  381. }
  382. #[allow(dead_code)]
  383. fn is_stopped(&self) -> bool {
  384. use self::PlayerState::*;
  385. match *self {
  386. Stopped => true,
  387. _ => false,
  388. }
  389. }
  390. fn is_loading(&self) -> bool {
  391. use self::PlayerState::*;
  392. match *self {
  393. Loading { .. } => true,
  394. _ => false,
  395. }
  396. }
  397. fn decoder(&mut self) -> Option<&mut Decoder> {
  398. use self::PlayerState::*;
  399. match *self {
  400. Stopped | EndOfTrack { .. } | Loading { .. } => None,
  401. Paused {
  402. ref mut decoder, ..
  403. }
  404. | Playing {
  405. ref mut decoder, ..
  406. } => Some(decoder),
  407. Invalid => panic!("invalid state"),
  408. }
  409. }
  410. fn stream_loader_controller(&mut self) -> Option<&mut StreamLoaderController> {
  411. use self::PlayerState::*;
  412. match *self {
  413. Stopped | EndOfTrack { .. } | Loading { .. } => None,
  414. Paused {
  415. ref mut stream_loader_controller,
  416. ..
  417. }
  418. | Playing {
  419. ref mut stream_loader_controller,
  420. ..
  421. } => Some(stream_loader_controller),
  422. Invalid => panic!("invalid state"),
  423. }
  424. }
  425. fn playing_to_end_of_track(&mut self) {
  426. use self::PlayerState::*;
  427. match mem::replace(self, Invalid) {
  428. Playing {
  429. track_id,
  430. play_request_id,
  431. decoder,
  432. duration_ms,
  433. bytes_per_second,
  434. normalisation_factor,
  435. stream_loader_controller,
  436. stream_position_pcm,
  437. ..
  438. } => {
  439. *self = EndOfTrack {
  440. track_id,
  441. play_request_id,
  442. loaded_track: PlayerLoadedTrackData {
  443. decoder,
  444. duration_ms,
  445. bytes_per_second,
  446. normalisation_factor,
  447. stream_loader_controller,
  448. stream_position_pcm,
  449. },
  450. };
  451. }
  452. _ => panic!("Called playing_to_end_of_track in non-playing state."),
  453. }
  454. }
  455. fn paused_to_playing(&mut self) {
  456. use self::PlayerState::*;
  457. match ::std::mem::replace(self, Invalid) {
  458. Paused {
  459. track_id,
  460. play_request_id,
  461. decoder,
  462. normalisation_factor,
  463. stream_loader_controller,
  464. duration_ms,
  465. bytes_per_second,
  466. stream_position_pcm,
  467. suggested_to_preload_next_track,
  468. } => {
  469. *self = Playing {
  470. track_id,
  471. play_request_id,
  472. decoder,
  473. normalisation_factor,
  474. stream_loader_controller,
  475. duration_ms,
  476. bytes_per_second,
  477. stream_position_pcm,
  478. reported_nominal_start_time: None,
  479. suggested_to_preload_next_track,
  480. };
  481. }
  482. _ => panic!("invalid state"),
  483. }
  484. }
  485. fn playing_to_paused(&mut self) {
  486. use self::PlayerState::*;
  487. match ::std::mem::replace(self, Invalid) {
  488. Playing {
  489. track_id,
  490. play_request_id,
  491. decoder,
  492. normalisation_factor,
  493. stream_loader_controller,
  494. duration_ms,
  495. bytes_per_second,
  496. stream_position_pcm,
  497. reported_nominal_start_time: _,
  498. suggested_to_preload_next_track,
  499. } => {
  500. *self = Paused {
  501. track_id,
  502. play_request_id,
  503. decoder,
  504. normalisation_factor,
  505. stream_loader_controller,
  506. duration_ms,
  507. bytes_per_second,
  508. stream_position_pcm,
  509. suggested_to_preload_next_track,
  510. };
  511. }
  512. _ => panic!("invalid state"),
  513. }
  514. }
  515. }
  516. struct PlayerTrackLoader {
  517. session: Session,
  518. config: PlayerConfig,
  519. }
  520. impl PlayerTrackLoader {
  521. fn find_available_alternative<'a>(&self, audio: &'a AudioItem) -> Option<Cow<'a, AudioItem>> {
  522. if audio.available {
  523. Some(Cow::Borrowed(audio))
  524. } else {
  525. if let Some(alternatives) = &audio.alternatives {
  526. let alternatives = alternatives
  527. .iter()
  528. .map(|alt_id| AudioItem::get_audio_item(&self.session, *alt_id));
  529. let alternatives = future::join_all(alternatives).wait().unwrap();
  530. alternatives
  531. .into_iter()
  532. .find(|alt| alt.available)
  533. .map(Cow::Owned)
  534. } else {
  535. None
  536. }
  537. }
  538. }
  539. fn stream_data_rate(&self, format: FileFormat) -> usize {
  540. match format {
  541. FileFormat::OGG_VORBIS_96 => 12 * 1024,
  542. FileFormat::OGG_VORBIS_160 => 20 * 1024,
  543. FileFormat::OGG_VORBIS_320 => 40 * 1024,
  544. FileFormat::MP3_256 => 32 * 1024,
  545. FileFormat::MP3_320 => 40 * 1024,
  546. FileFormat::MP3_160 => 20 * 1024,
  547. FileFormat::MP3_96 => 12 * 1024,
  548. FileFormat::MP3_160_ENC => 20 * 1024,
  549. FileFormat::MP4_128_DUAL => 16 * 1024,
  550. FileFormat::OTHER3 => 40 * 1024, // better some high guess than nothing
  551. FileFormat::AAC_160 => 20 * 1024,
  552. FileFormat::AAC_320 => 40 * 1024,
  553. FileFormat::MP4_128 => 16 * 1024,
  554. FileFormat::OTHER5 => 40 * 1024, // better some high guess than nothing
  555. }
  556. }
  557. fn load_track(&self, spotify_id: SpotifyId, position_ms: u32) -> Option<PlayerLoadedTrackData> {
  558. let audio = match AudioItem::get_audio_item(&self.session, spotify_id).wait() {
  559. Ok(audio) => audio,
  560. Err(_) => {
  561. error!("Unable to load audio item.");
  562. return None;
  563. }
  564. };
  565. info!("Loading <{}> with Spotify URI <{}>", audio.name, audio.uri);
  566. let audio = match self.find_available_alternative(&audio) {
  567. Some(audio) => audio,
  568. None => {
  569. warn!("<{}> is not available", audio.uri);
  570. return None;
  571. }
  572. };
  573. assert!(audio.duration >= 0);
  574. let duration_ms = audio.duration as u32;
  575. // (Most) podcasts seem to support only 96 bit Vorbis, so fall back to it
  576. let formats = match self.config.bitrate {
  577. Bitrate::Bitrate96 => [
  578. FileFormat::OGG_VORBIS_96,
  579. FileFormat::OGG_VORBIS_160,
  580. FileFormat::OGG_VORBIS_320,
  581. ],
  582. Bitrate::Bitrate160 => [
  583. FileFormat::OGG_VORBIS_160,
  584. FileFormat::OGG_VORBIS_96,
  585. FileFormat::OGG_VORBIS_320,
  586. ],
  587. Bitrate::Bitrate320 => [
  588. FileFormat::OGG_VORBIS_320,
  589. FileFormat::OGG_VORBIS_160,
  590. FileFormat::OGG_VORBIS_96,
  591. ],
  592. };
  593. let format = formats
  594. .iter()
  595. .find(|format| audio.files.contains_key(format))
  596. .unwrap();
  597. let file_id = match audio.files.get(&format) {
  598. Some(&file_id) => file_id,
  599. None => {
  600. warn!("<{}> in not available in format {:?}", audio.name, format);
  601. return None;
  602. }
  603. };
  604. let bytes_per_second = self.stream_data_rate(*format);
  605. let play_from_beginning = position_ms == 0;
  606. let key = self.session.audio_key().request(spotify_id, file_id);
  607. let encrypted_file = AudioFile::open(
  608. &self.session,
  609. file_id,
  610. bytes_per_second,
  611. play_from_beginning,
  612. );
  613. let encrypted_file = match encrypted_file.wait() {
  614. Ok(encrypted_file) => encrypted_file,
  615. Err(_) => {
  616. error!("Unable to load encrypted file.");
  617. return None;
  618. }
  619. };
  620. let mut stream_loader_controller = encrypted_file.get_stream_loader_controller();
  621. if play_from_beginning {
  622. // No need to seek -> we stream from the beginning
  623. stream_loader_controller.set_stream_mode();
  624. } else {
  625. // we need to seek -> we set stream mode after the initial seek.
  626. stream_loader_controller.set_random_access_mode();
  627. }
  628. let key = match key.wait() {
  629. Ok(key) => key,
  630. Err(_) => {
  631. error!("Unable to load decryption key");
  632. return None;
  633. }
  634. };
  635. let mut decrypted_file = AudioDecrypt::new(key, encrypted_file);
  636. let normalisation_factor = match NormalisationData::parse_from_file(&mut decrypted_file) {
  637. Ok(normalisation_data) => {
  638. NormalisationData::get_factor(&self.config, normalisation_data)
  639. }
  640. Err(_) => {
  641. warn!("Unable to extract normalisation data, using default value.");
  642. 1.0 as f32
  643. }
  644. };
  645. let audio_file = Subfile::new(decrypted_file, 0xa7);
  646. let mut decoder = VorbisDecoder::new(audio_file).unwrap();
  647. if position_ms != 0 {
  648. match decoder.seek(position_ms as i64) {
  649. Ok(_) => (),
  650. Err(err) => error!("Vorbis error: {:?}", err),
  651. }
  652. stream_loader_controller.set_stream_mode();
  653. }
  654. let stream_position_pcm = PlayerInternal::position_ms_to_pcm(position_ms);
  655. info!("<{}> ({} ms) loaded", audio.name, audio.duration);
  656. Some(PlayerLoadedTrackData {
  657. decoder,
  658. normalisation_factor,
  659. stream_loader_controller,
  660. bytes_per_second,
  661. duration_ms,
  662. stream_position_pcm,
  663. })
  664. }
  665. }
  666. impl Future for PlayerInternal {
  667. type Item = ();
  668. type Error = ();
  669. fn poll(&mut self) -> Poll<(), ()> {
  670. // While this is written as a future, it still contains blocking code.
  671. // It must be run on its own thread.
  672. loop {
  673. let mut all_futures_completed_or_not_ready = true;
  674. // process commands that were sent to us
  675. let cmd = match self.commands.poll() {
  676. Ok(Async::Ready(None)) => return Ok(Async::Ready(())), // client has disconnected - shut down.
  677. Ok(Async::Ready(Some(cmd))) => {
  678. all_futures_completed_or_not_ready = false;
  679. Some(cmd)
  680. }
  681. Ok(Async::NotReady) => None,
  682. Err(_) => None,
  683. };
  684. if let Some(cmd) = cmd {
  685. self.handle_command(cmd);
  686. }
  687. // Handle loading of a new track to play
  688. if let PlayerState::Loading {
  689. ref mut loader,
  690. track_id,
  691. start_playback,
  692. play_request_id,
  693. } = self.state
  694. {
  695. match loader.poll() {
  696. Ok(Async::Ready(loaded_track)) => {
  697. self.start_playback(
  698. track_id,
  699. play_request_id,
  700. loaded_track,
  701. start_playback,
  702. );
  703. if let PlayerState::Loading { .. } = self.state {
  704. panic!("The state wasn't changed by start_playback()");
  705. }
  706. }
  707. Ok(Async::NotReady) => (),
  708. Err(_) => {
  709. warn!("Unable to load <{:?}>\nSkipping to next track", track_id);
  710. assert!(self.state.is_loading());
  711. self.send_event(PlayerEvent::EndOfTrack {
  712. track_id,
  713. play_request_id,
  714. })
  715. }
  716. }
  717. }
  718. // handle pending preload requests.
  719. if let PlayerPreload::Loading {
  720. ref mut loader,
  721. track_id,
  722. } = self.preload
  723. {
  724. match loader.poll() {
  725. Ok(Async::Ready(loaded_track)) => {
  726. self.send_event(PlayerEvent::Preloading { track_id });
  727. self.preload = PlayerPreload::Ready {
  728. track_id,
  729. loaded_track,
  730. };
  731. }
  732. Ok(Async::NotReady) => (),
  733. Err(_) => {
  734. debug!("Unable to preload {:?}", track_id);
  735. self.preload = PlayerPreload::None;
  736. // Let Spirc know that the track was unavailable.
  737. if let PlayerState::Playing {
  738. play_request_id, ..
  739. }
  740. | PlayerState::Paused {
  741. play_request_id, ..
  742. } = self.state
  743. {
  744. self.send_event(PlayerEvent::Unavailable {
  745. track_id,
  746. play_request_id,
  747. });
  748. }
  749. }
  750. }
  751. }
  752. if self.state.is_playing() {
  753. self.ensure_sink_running();
  754. if let PlayerState::Playing {
  755. track_id,
  756. play_request_id,
  757. ref mut decoder,
  758. normalisation_factor,
  759. ref mut stream_position_pcm,
  760. ref mut reported_nominal_start_time,
  761. duration_ms,
  762. ..
  763. } = self.state
  764. {
  765. let packet = decoder.next_packet().expect("Vorbis error");
  766. if let Some(ref packet) = packet {
  767. *stream_position_pcm =
  768. *stream_position_pcm + (packet.data().len() / 2) as u64;
  769. let stream_position_millis = Self::position_pcm_to_ms(*stream_position_pcm);
  770. let notify_about_position = match *reported_nominal_start_time {
  771. None => true,
  772. Some(reported_nominal_start_time) => {
  773. // 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.
  774. let lag = (Instant::now() - reported_nominal_start_time).as_millis()
  775. as i64
  776. - stream_position_millis as i64;
  777. if lag > 1000 {
  778. true
  779. } else {
  780. false
  781. }
  782. }
  783. };
  784. if notify_about_position {
  785. *reported_nominal_start_time = Some(
  786. Instant::now()
  787. - Duration::from_millis(stream_position_millis as u64),
  788. );
  789. self.send_event(PlayerEvent::Playing {
  790. track_id,
  791. play_request_id,
  792. position_ms: stream_position_millis as u32,
  793. duration_ms,
  794. });
  795. }
  796. }
  797. self.handle_packet(packet, normalisation_factor);
  798. } else {
  799. unreachable!();
  800. };
  801. }
  802. if let PlayerState::Playing {
  803. track_id,
  804. play_request_id,
  805. duration_ms,
  806. stream_position_pcm,
  807. ref mut stream_loader_controller,
  808. ref mut suggested_to_preload_next_track,
  809. ..
  810. }
  811. | PlayerState::Paused {
  812. track_id,
  813. play_request_id,
  814. duration_ms,
  815. stream_position_pcm,
  816. ref mut stream_loader_controller,
  817. ref mut suggested_to_preload_next_track,
  818. ..
  819. } = self.state
  820. {
  821. if (!*suggested_to_preload_next_track)
  822. && ((duration_ms as i64 - Self::position_pcm_to_ms(stream_position_pcm) as i64)
  823. < PRELOAD_NEXT_TRACK_BEFORE_END_DURATION_MS as i64)
  824. && stream_loader_controller.range_to_end_available()
  825. {
  826. *suggested_to_preload_next_track = true;
  827. self.send_event(PlayerEvent::TimeToPreloadNextTrack {
  828. track_id,
  829. play_request_id,
  830. });
  831. }
  832. }
  833. if self.session.is_invalid() {
  834. return Ok(Async::Ready(()));
  835. }
  836. if (!self.state.is_playing()) && all_futures_completed_or_not_ready {
  837. return Ok(Async::NotReady);
  838. }
  839. }
  840. }
  841. }
  842. impl PlayerInternal {
  843. fn position_pcm_to_ms(position_pcm: u64) -> u32 {
  844. (position_pcm * 10 / 441) as u32
  845. }
  846. fn position_ms_to_pcm(position_ms: u32) -> u64 {
  847. position_ms as u64 * 441 / 10
  848. }
  849. fn ensure_sink_running(&mut self) {
  850. if self.sink_status != SinkStatus::Running {
  851. trace!("== Starting sink ==");
  852. if let Some(callback) = &mut self.sink_event_callback {
  853. callback(SinkStatus::Running);
  854. }
  855. match self.sink.start() {
  856. Ok(()) => self.sink_status = SinkStatus::Running,
  857. Err(err) => error!("Could not start audio: {}", err),
  858. }
  859. }
  860. }
  861. fn ensure_sink_stopped(&mut self, temporarily: bool) {
  862. match self.sink_status {
  863. SinkStatus::Running => {
  864. trace!("== Stopping sink ==");
  865. self.sink.stop().unwrap();
  866. self.sink_status = if temporarily {
  867. SinkStatus::TemporarilyClosed
  868. } else {
  869. SinkStatus::Closed
  870. };
  871. if let Some(callback) = &mut self.sink_event_callback {
  872. callback(self.sink_status);
  873. }
  874. }
  875. SinkStatus::TemporarilyClosed => {
  876. if !temporarily {
  877. self.sink_status = SinkStatus::Closed;
  878. if let Some(callback) = &mut self.sink_event_callback {
  879. callback(SinkStatus::Closed);
  880. }
  881. }
  882. }
  883. SinkStatus::Closed => (),
  884. }
  885. }
  886. fn handle_player_stop(&mut self) {
  887. match self.state {
  888. PlayerState::Playing {
  889. track_id,
  890. play_request_id,
  891. ..
  892. }
  893. | PlayerState::Paused {
  894. track_id,
  895. play_request_id,
  896. ..
  897. }
  898. | PlayerState::EndOfTrack {
  899. track_id,
  900. play_request_id,
  901. ..
  902. }
  903. | PlayerState::Loading {
  904. track_id,
  905. play_request_id,
  906. ..
  907. } => {
  908. self.ensure_sink_stopped(false);
  909. self.send_event(PlayerEvent::Stopped {
  910. track_id,
  911. play_request_id,
  912. });
  913. self.state = PlayerState::Stopped;
  914. }
  915. PlayerState::Stopped => (),
  916. PlayerState::Invalid => panic!("invalid state"),
  917. }
  918. }
  919. fn handle_play(&mut self) {
  920. if let PlayerState::Paused {
  921. track_id,
  922. play_request_id,
  923. stream_position_pcm,
  924. duration_ms,
  925. ..
  926. } = self.state
  927. {
  928. self.state.paused_to_playing();
  929. let position_ms = Self::position_pcm_to_ms(stream_position_pcm);
  930. self.send_event(PlayerEvent::Playing {
  931. track_id,
  932. play_request_id,
  933. position_ms,
  934. duration_ms,
  935. });
  936. self.ensure_sink_running();
  937. } else {
  938. warn!("Player::play called from invalid state");
  939. }
  940. }
  941. fn handle_pause(&mut self) {
  942. if let PlayerState::Playing {
  943. track_id,
  944. play_request_id,
  945. stream_position_pcm,
  946. duration_ms,
  947. ..
  948. } = self.state
  949. {
  950. self.state.playing_to_paused();
  951. self.ensure_sink_stopped(false);
  952. let position_ms = Self::position_pcm_to_ms(stream_position_pcm);
  953. self.send_event(PlayerEvent::Paused {
  954. track_id,
  955. play_request_id,
  956. position_ms,
  957. duration_ms,
  958. });
  959. } else {
  960. warn!("Player::pause called from invalid state");
  961. }
  962. }
  963. fn handle_packet(&mut self, packet: Option<VorbisPacket>, normalisation_factor: f32) {
  964. match packet {
  965. Some(mut packet) => {
  966. if packet.data().len() > 0 {
  967. if let Some(ref editor) = self.audio_filter {
  968. editor.modify_stream(&mut packet.data_mut())
  969. };
  970. if self.config.normalisation && normalisation_factor != 1.0 {
  971. for x in packet.data_mut().iter_mut() {
  972. *x = (*x as f32 * normalisation_factor) as i16;
  973. }
  974. }
  975. if let Err(err) = self.sink.write(&packet.data()) {
  976. error!("Could not write audio: {}", err);
  977. self.ensure_sink_stopped(false);
  978. }
  979. }
  980. }
  981. None => {
  982. self.state.playing_to_end_of_track();
  983. if let PlayerState::EndOfTrack {
  984. track_id,
  985. play_request_id,
  986. ..
  987. } = self.state
  988. {
  989. self.send_event(PlayerEvent::EndOfTrack {
  990. track_id,
  991. play_request_id,
  992. })
  993. } else {
  994. unreachable!();
  995. }
  996. }
  997. }
  998. }
  999. fn start_playback(
  1000. &mut self,
  1001. track_id: SpotifyId,
  1002. play_request_id: u64,
  1003. loaded_track: PlayerLoadedTrackData,
  1004. start_playback: bool,
  1005. ) {
  1006. let position_ms = Self::position_pcm_to_ms(loaded_track.stream_position_pcm);
  1007. if start_playback {
  1008. self.ensure_sink_running();
  1009. self.send_event(PlayerEvent::Playing {
  1010. track_id,
  1011. play_request_id,
  1012. position_ms,
  1013. duration_ms: loaded_track.duration_ms,
  1014. });
  1015. self.state = PlayerState::Playing {
  1016. track_id: track_id,
  1017. play_request_id: play_request_id,
  1018. decoder: loaded_track.decoder,
  1019. normalisation_factor: loaded_track.normalisation_factor,
  1020. stream_loader_controller: loaded_track.stream_loader_controller,
  1021. duration_ms: loaded_track.duration_ms,
  1022. bytes_per_second: loaded_track.bytes_per_second,
  1023. stream_position_pcm: loaded_track.stream_position_pcm,
  1024. reported_nominal_start_time: Some(
  1025. Instant::now() - Duration::from_millis(position_ms as u64),
  1026. ),
  1027. suggested_to_preload_next_track: false,
  1028. };
  1029. } else {
  1030. self.ensure_sink_stopped(false);
  1031. self.state = PlayerState::Paused {
  1032. track_id: track_id,
  1033. play_request_id: play_request_id,
  1034. decoder: loaded_track.decoder,
  1035. normalisation_factor: loaded_track.normalisation_factor,
  1036. stream_loader_controller: loaded_track.stream_loader_controller,
  1037. duration_ms: loaded_track.duration_ms,
  1038. bytes_per_second: loaded_track.bytes_per_second,
  1039. stream_position_pcm: loaded_track.stream_position_pcm,
  1040. suggested_to_preload_next_track: false,
  1041. };
  1042. self.send_event(PlayerEvent::Paused {
  1043. track_id,
  1044. play_request_id,
  1045. position_ms,
  1046. duration_ms: loaded_track.duration_ms,
  1047. });
  1048. }
  1049. }
  1050. fn handle_command_load(
  1051. &mut self,
  1052. track_id: SpotifyId,
  1053. play_request_id: u64,
  1054. play: bool,
  1055. position_ms: u32,
  1056. ) {
  1057. if !self.config.gapless {
  1058. self.ensure_sink_stopped(play);
  1059. }
  1060. // emit the correct player event
  1061. match self.state {
  1062. PlayerState::Playing {
  1063. track_id: old_track_id,
  1064. ..
  1065. }
  1066. | PlayerState::Paused {
  1067. track_id: old_track_id,
  1068. ..
  1069. }
  1070. | PlayerState::EndOfTrack {
  1071. track_id: old_track_id,
  1072. ..
  1073. }
  1074. | PlayerState::Loading {
  1075. track_id: old_track_id,
  1076. ..
  1077. } => self.send_event(PlayerEvent::Changed {
  1078. old_track_id: old_track_id,
  1079. new_track_id: track_id,
  1080. }),
  1081. PlayerState::Stopped => self.send_event(PlayerEvent::Started {
  1082. track_id,
  1083. play_request_id,
  1084. position_ms,
  1085. }),
  1086. PlayerState::Invalid { .. } => panic!("Player is in an invalid state."),
  1087. }
  1088. // Now we check at different positions whether we already have a pre-loaded version
  1089. // of this track somewhere. If so, use it and return.
  1090. // Check if there's a matching loaded track in the EndOfTrack player state.
  1091. // This is the case if we're repeating the same track again.
  1092. if let PlayerState::EndOfTrack {
  1093. track_id: previous_track_id,
  1094. ..
  1095. } = self.state
  1096. {
  1097. if previous_track_id == track_id {
  1098. let mut loaded_track = match mem::replace(&mut self.state, PlayerState::Invalid) {
  1099. PlayerState::EndOfTrack { loaded_track, .. } => loaded_track,
  1100. _ => unreachable!(),
  1101. };
  1102. if Self::position_ms_to_pcm(position_ms) != loaded_track.stream_position_pcm {
  1103. loaded_track
  1104. .stream_loader_controller
  1105. .set_random_access_mode();
  1106. let _ = loaded_track.decoder.seek(position_ms as i64); // This may be blocking.
  1107. // But most likely the track is fully
  1108. // loaded already because we played
  1109. // to the end of it.
  1110. loaded_track.stream_loader_controller.set_stream_mode();
  1111. loaded_track.stream_position_pcm = Self::position_ms_to_pcm(position_ms);
  1112. }
  1113. self.preload = PlayerPreload::None;
  1114. self.start_playback(track_id, play_request_id, loaded_track, play);
  1115. if let PlayerState::Invalid = self.state {
  1116. panic!("start_playback() hasn't set a valid player state.");
  1117. }
  1118. return;
  1119. }
  1120. }
  1121. // Check if we are already playing the track. If so, just do a seek and update our info.
  1122. if let PlayerState::Playing {
  1123. track_id: current_track_id,
  1124. ref mut stream_position_pcm,
  1125. ref mut decoder,
  1126. ref mut stream_loader_controller,
  1127. ..
  1128. }
  1129. | PlayerState::Paused {
  1130. track_id: current_track_id,
  1131. ref mut stream_position_pcm,
  1132. ref mut decoder,
  1133. ref mut stream_loader_controller,
  1134. ..
  1135. } = self.state
  1136. {
  1137. if current_track_id == track_id {
  1138. // we can use the current decoder. Ensure it's at the correct position.
  1139. if Self::position_ms_to_pcm(position_ms) != *stream_position_pcm {
  1140. stream_loader_controller.set_random_access_mode();
  1141. let _ = decoder.seek(position_ms as i64); // This may be blocking.
  1142. stream_loader_controller.set_stream_mode();
  1143. *stream_position_pcm = Self::position_ms_to_pcm(position_ms);
  1144. }
  1145. // Move the info from the current state into a PlayerLoadedTrackData so we can use
  1146. // the usual code path to start playback.
  1147. let old_state = mem::replace(&mut self.state, PlayerState::Invalid);
  1148. if let PlayerState::Playing {
  1149. stream_position_pcm,
  1150. decoder,
  1151. stream_loader_controller,
  1152. bytes_per_second,
  1153. duration_ms,
  1154. normalisation_factor,
  1155. ..
  1156. }
  1157. | PlayerState::Paused {
  1158. stream_position_pcm,
  1159. decoder,
  1160. stream_loader_controller,
  1161. bytes_per_second,
  1162. duration_ms,
  1163. normalisation_factor,
  1164. ..
  1165. } = old_state
  1166. {
  1167. let loaded_track = PlayerLoadedTrackData {
  1168. decoder,
  1169. normalisation_factor,
  1170. stream_loader_controller,
  1171. bytes_per_second,
  1172. duration_ms,
  1173. stream_position_pcm,
  1174. };
  1175. self.preload = PlayerPreload::None;
  1176. self.start_playback(track_id, play_request_id, loaded_track, play);
  1177. if let PlayerState::Invalid = self.state {
  1178. panic!("start_playback() hasn't set a valid player state.");
  1179. }
  1180. return;
  1181. } else {
  1182. unreachable!();
  1183. }
  1184. }
  1185. }
  1186. // Check if the requested track has been preloaded already. If so use the preloaded data.
  1187. if let PlayerPreload::Ready {
  1188. track_id: loaded_track_id,
  1189. ..
  1190. } = self.preload
  1191. {
  1192. if track_id == loaded_track_id {
  1193. let preload = std::mem::replace(&mut self.preload, PlayerPreload::None);
  1194. if let PlayerPreload::Ready {
  1195. track_id,
  1196. mut loaded_track,
  1197. } = preload
  1198. {
  1199. if Self::position_ms_to_pcm(position_ms) != loaded_track.stream_position_pcm {
  1200. loaded_track
  1201. .stream_loader_controller
  1202. .set_random_access_mode();
  1203. let _ = loaded_track.decoder.seek(position_ms as i64); // This may be blocking
  1204. loaded_track.stream_loader_controller.set_stream_mode();
  1205. }
  1206. self.start_playback(track_id, play_request_id, loaded_track, play);
  1207. return;
  1208. } else {
  1209. unreachable!();
  1210. }
  1211. }
  1212. }
  1213. // We need to load the track - either from scratch or by completing a preload.
  1214. // In any case we go into a Loading state to load the track.
  1215. self.ensure_sink_stopped(play);
  1216. self.send_event(PlayerEvent::Loading {
  1217. track_id,
  1218. play_request_id,
  1219. position_ms,
  1220. });
  1221. // Try to extract a pending loader from the preloading mechanism
  1222. let loader = if let PlayerPreload::Loading {
  1223. track_id: loaded_track_id,
  1224. ..
  1225. } = self.preload
  1226. {
  1227. if (track_id == loaded_track_id) && (position_ms == 0) {
  1228. let mut preload = PlayerPreload::None;
  1229. std::mem::swap(&mut preload, &mut self.preload);
  1230. if let PlayerPreload::Loading { loader, .. } = preload {
  1231. Some(loader)
  1232. } else {
  1233. None
  1234. }
  1235. } else {
  1236. None
  1237. }
  1238. } else {
  1239. None
  1240. };
  1241. self.preload = PlayerPreload::None;
  1242. // If we don't have a loader yet, create one from scratch.
  1243. let loader = loader
  1244. .or_else(|| Some(self.load_track(track_id, position_ms)))
  1245. .unwrap();
  1246. // Set ourselves to a loading state.
  1247. self.state = PlayerState::Loading {
  1248. track_id,
  1249. play_request_id,
  1250. start_playback: play,
  1251. loader,
  1252. };
  1253. }
  1254. fn handle_command_preload(&mut self, track_id: SpotifyId) {
  1255. debug!("Preloading track");
  1256. let mut preload_track = true;
  1257. // check whether the track is already loaded somewhere or being loaded.
  1258. if let PlayerPreload::Loading {
  1259. track_id: currently_loading,
  1260. ..
  1261. }
  1262. | PlayerPreload::Ready {
  1263. track_id: currently_loading,
  1264. ..
  1265. } = self.preload
  1266. {
  1267. if currently_loading == track_id {
  1268. // we're already preloading the requested track.
  1269. preload_track = false;
  1270. } else {
  1271. // we're preloading something else - cancel it.
  1272. self.preload = PlayerPreload::None;
  1273. }
  1274. }
  1275. if let PlayerState::Playing {
  1276. track_id: current_track_id,
  1277. ..
  1278. }
  1279. | PlayerState::Paused {
  1280. track_id: current_track_id,
  1281. ..
  1282. }
  1283. | PlayerState::EndOfTrack {
  1284. track_id: current_track_id,
  1285. ..
  1286. } = self.state
  1287. {
  1288. if current_track_id == track_id {
  1289. // we already have the requested track loaded.
  1290. preload_track = false;
  1291. }
  1292. }
  1293. // schedule the preload of the current track if desired.
  1294. if preload_track {
  1295. let loader = self.load_track(track_id, 0);
  1296. self.preload = PlayerPreload::Loading { track_id, loader }
  1297. }
  1298. }
  1299. fn handle_command_seek(&mut self, position_ms: u32) {
  1300. if let Some(stream_loader_controller) = self.state.stream_loader_controller() {
  1301. stream_loader_controller.set_random_access_mode();
  1302. }
  1303. if let Some(decoder) = self.state.decoder() {
  1304. match decoder.seek(position_ms as i64) {
  1305. Ok(_) => {
  1306. if let PlayerState::Playing {
  1307. ref mut stream_position_pcm,
  1308. ..
  1309. }
  1310. | PlayerState::Paused {
  1311. ref mut stream_position_pcm,
  1312. ..
  1313. } = self.state
  1314. {
  1315. *stream_position_pcm = Self::position_ms_to_pcm(position_ms);
  1316. }
  1317. }
  1318. Err(err) => error!("Vorbis error: {:?}", err),
  1319. }
  1320. } else {
  1321. warn!("Player::seek called from invalid state");
  1322. }
  1323. // If we're playing, ensure, that we have enough data leaded to avoid a buffer underrun.
  1324. if let Some(stream_loader_controller) = self.state.stream_loader_controller() {
  1325. stream_loader_controller.set_stream_mode();
  1326. }
  1327. // ensure we have a bit of a buffer of downloaded data
  1328. self.preload_data_before_playback();
  1329. if let PlayerState::Playing {
  1330. track_id,
  1331. play_request_id,
  1332. ref mut reported_nominal_start_time,
  1333. duration_ms,
  1334. ..
  1335. } = self.state
  1336. {
  1337. *reported_nominal_start_time =
  1338. Some(Instant::now() - Duration::from_millis(position_ms as u64));
  1339. self.send_event(PlayerEvent::Playing {
  1340. track_id,
  1341. play_request_id,
  1342. position_ms,
  1343. duration_ms,
  1344. });
  1345. }
  1346. if let PlayerState::Paused {
  1347. track_id,
  1348. play_request_id,
  1349. duration_ms,
  1350. ..
  1351. } = self.state
  1352. {
  1353. self.send_event(PlayerEvent::Paused {
  1354. track_id,
  1355. play_request_id,
  1356. position_ms,
  1357. duration_ms,
  1358. });
  1359. }
  1360. }
  1361. fn handle_command(&mut self, cmd: PlayerCommand) {
  1362. debug!("command={:?}", cmd);
  1363. match cmd {
  1364. PlayerCommand::Load {
  1365. track_id,
  1366. play_request_id,
  1367. play,
  1368. position_ms,
  1369. } => self.handle_command_load(track_id, play_request_id, play, position_ms),
  1370. PlayerCommand::Preload { track_id } => self.handle_command_preload(track_id),
  1371. PlayerCommand::Seek(position_ms) => self.handle_command_seek(position_ms),
  1372. PlayerCommand::Play => self.handle_play(),
  1373. PlayerCommand::Pause => self.handle_pause(),
  1374. PlayerCommand::Stop => self.handle_player_stop(),
  1375. PlayerCommand::AddEventSender(sender) => self.event_senders.push(sender),
  1376. PlayerCommand::SetSinkEventCallback(callback) => self.sink_event_callback = callback,
  1377. PlayerCommand::EmitVolumeSetEvent(volume) => {
  1378. self.send_event(PlayerEvent::VolumeSet { volume })
  1379. }
  1380. }
  1381. }
  1382. fn send_event(&mut self, event: PlayerEvent) {
  1383. let mut index = 0;
  1384. while index < self.event_senders.len() {
  1385. match self.event_senders[index].unbounded_send(event.clone()) {
  1386. Ok(_) => index += 1,
  1387. Err(_) => {
  1388. self.event_senders.remove(index);
  1389. }
  1390. }
  1391. }
  1392. }
  1393. fn load_track(
  1394. &self,
  1395. spotify_id: SpotifyId,
  1396. position_ms: u32,
  1397. ) -> Box<dyn Future<Item = PlayerLoadedTrackData, Error = ()>> {
  1398. // This method creates a future that returns the loaded stream and associated info.
  1399. // Ideally all work should be done using asynchronous code. However, seek() on the
  1400. // audio stream is implemented in a blocking fashion. Thus, we can't turn it into future
  1401. // easily. Instead we spawn a thread to do the work and return a one-shot channel as the
  1402. // future to work with.
  1403. let loader = PlayerTrackLoader {
  1404. session: self.session.clone(),
  1405. config: self.config.clone(),
  1406. };
  1407. let (result_tx, result_rx) = futures::sync::oneshot::channel();
  1408. std::thread::spawn(move || {
  1409. loader
  1410. .load_track(spotify_id, position_ms)
  1411. .and_then(move |data| {
  1412. let _ = result_tx.send(data);
  1413. Some(())
  1414. });
  1415. });
  1416. Box::new(result_rx.map_err(|_| ()))
  1417. }
  1418. fn preload_data_before_playback(&mut self) {
  1419. if let PlayerState::Playing {
  1420. bytes_per_second,
  1421. ref mut stream_loader_controller,
  1422. ..
  1423. } = self.state
  1424. {
  1425. // Request our read ahead range
  1426. let request_data_length = max(
  1427. (READ_AHEAD_DURING_PLAYBACK_ROUNDTRIPS
  1428. * (0.001 * stream_loader_controller.ping_time_ms() as f64)
  1429. * bytes_per_second as f64) as usize,
  1430. (READ_AHEAD_DURING_PLAYBACK_SECONDS * bytes_per_second as f64) as usize,
  1431. );
  1432. stream_loader_controller.fetch_next(request_data_length);
  1433. // Request the part we want to wait for blocking. This effecively means we wait for the previous request to partially complete.
  1434. let wait_for_data_length = max(
  1435. (READ_AHEAD_BEFORE_PLAYBACK_ROUNDTRIPS
  1436. * (0.001 * stream_loader_controller.ping_time_ms() as f64)
  1437. * bytes_per_second as f64) as usize,
  1438. (READ_AHEAD_BEFORE_PLAYBACK_SECONDS * bytes_per_second as f64) as usize,
  1439. );
  1440. stream_loader_controller.fetch_next_blocking(wait_for_data_length);
  1441. }
  1442. }
  1443. }
  1444. impl Drop for PlayerInternal {
  1445. fn drop(&mut self) {
  1446. debug!("drop PlayerInternal[{}]", self.session.session_id());
  1447. }
  1448. }
  1449. impl ::std::fmt::Debug for PlayerCommand {
  1450. fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
  1451. match *self {
  1452. PlayerCommand::Load {
  1453. track_id,
  1454. play,
  1455. position_ms,
  1456. ..
  1457. } => f
  1458. .debug_tuple("Load")
  1459. .field(&track_id)
  1460. .field(&play)
  1461. .field(&position_ms)
  1462. .finish(),
  1463. PlayerCommand::Preload { track_id } => {
  1464. f.debug_tuple("Preload").field(&track_id).finish()
  1465. }
  1466. PlayerCommand::Play => f.debug_tuple("Play").finish(),
  1467. PlayerCommand::Pause => f.debug_tuple("Pause").finish(),
  1468. PlayerCommand::Stop => f.debug_tuple("Stop").finish(),
  1469. PlayerCommand::Seek(position) => f.debug_tuple("Seek").field(&position).finish(),
  1470. PlayerCommand::AddEventSender(_) => f.debug_tuple("AddEventSender").finish(),
  1471. PlayerCommand::SetSinkEventCallback(_) => {
  1472. f.debug_tuple("SetSinkEventCallback").finish()
  1473. }
  1474. PlayerCommand::EmitVolumeSetEvent(volume) => {
  1475. f.debug_tuple("VolumeSet").field(&volume).finish()
  1476. }
  1477. }
  1478. }
  1479. }
  1480. impl ::std::fmt::Debug for PlayerState {
  1481. fn fmt(&self, f: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
  1482. use PlayerState::*;
  1483. match *self {
  1484. Stopped => f.debug_struct("Stopped").finish(),
  1485. Loading {
  1486. track_id,
  1487. play_request_id,
  1488. ..
  1489. } => f
  1490. .debug_struct("Loading")
  1491. .field("track_id", &track_id)
  1492. .field("play_request_id", &play_request_id)
  1493. .finish(),
  1494. Paused {
  1495. track_id,
  1496. play_request_id,
  1497. ..
  1498. } => f
  1499. .debug_struct("Paused")
  1500. .field("track_id", &track_id)
  1501. .field("play_request_id", &play_request_id)
  1502. .finish(),
  1503. Playing {
  1504. track_id,
  1505. play_request_id,
  1506. ..
  1507. } => f
  1508. .debug_struct("Playing")
  1509. .field("track_id", &track_id)
  1510. .field("play_request_id", &play_request_id)
  1511. .finish(),
  1512. EndOfTrack {
  1513. track_id,
  1514. play_request_id,
  1515. ..
  1516. } => f
  1517. .debug_struct("EndOfTrack")
  1518. .field("track_id", &track_id)
  1519. .field("play_request_id", &play_request_id)
  1520. .finish(),
  1521. Invalid => f.debug_struct("Invalid").finish(),
  1522. }
  1523. }
  1524. }
  1525. struct Subfile<T: Read + Seek> {
  1526. stream: T,
  1527. offset: u64,
  1528. }
  1529. impl<T: Read + Seek> Subfile<T> {
  1530. pub fn new(mut stream: T, offset: u64) -> Subfile<T> {
  1531. stream.seek(SeekFrom::Start(offset)).unwrap();
  1532. Subfile {
  1533. stream: stream,
  1534. offset: offset,
  1535. }
  1536. }
  1537. }
  1538. impl<T: Read + Seek> Read for Subfile<T> {
  1539. fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
  1540. self.stream.read(buf)
  1541. }
  1542. }
  1543. impl<T: Read + Seek> Seek for Subfile<T> {
  1544. fn seek(&mut self, mut pos: SeekFrom) -> Result<u64> {
  1545. pos = match pos {
  1546. SeekFrom::Start(offset) => SeekFrom::Start(offset + self.offset),
  1547. x => x,
  1548. };
  1549. let newpos = self.stream.seek(pos)?;
  1550. if newpos > self.offset {
  1551. Ok(newpos - self.offset)
  1552. } else {
  1553. Ok(0)
  1554. }
  1555. }
  1556. }