spirc.rs 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940
  1. use std;
  2. use std::time::{SystemTime, UNIX_EPOCH};
  3. use futures::future;
  4. use futures::sync::{mpsc, oneshot};
  5. use futures::{Async, Future, Poll, Sink, Stream};
  6. use protobuf::{self, Message};
  7. use rand;
  8. use rand::seq::SliceRandom;
  9. use serde_json;
  10. use context::StationContext;
  11. use librespot_core::config::ConnectConfig;
  12. use librespot_core::mercury::MercuryError;
  13. use librespot_core::session::Session;
  14. use librespot_core::spotify_id::{SpotifyAudioType, SpotifyId, SpotifyIdError};
  15. use librespot_core::util::SeqGenerator;
  16. use librespot_core::version;
  17. use librespot_core::volume::Volume;
  18. use playback::mixer::Mixer;
  19. use playback::player::Player;
  20. use protocol;
  21. use protocol::spirc::{DeviceState, Frame, MessageType, PlayStatus, State, TrackRef};
  22. pub struct SpircTask {
  23. player: Player,
  24. mixer: Box<Mixer>,
  25. linear_volume: bool,
  26. sequence: SeqGenerator<u32>,
  27. ident: String,
  28. device: DeviceState,
  29. state: State,
  30. subscription: Box<Stream<Item = Frame, Error = MercuryError>>,
  31. sender: Box<Sink<SinkItem = Frame, SinkError = MercuryError>>,
  32. commands: mpsc::UnboundedReceiver<SpircCommand>,
  33. end_of_track: Box<Future<Item = (), Error = oneshot::Canceled>>,
  34. shutdown: bool,
  35. session: Session,
  36. context_fut: Box<Future<Item = serde_json::Value, Error = MercuryError>>,
  37. context: Option<StationContext>,
  38. }
  39. pub enum SpircCommand {
  40. Play,
  41. PlayPause,
  42. Pause,
  43. Prev,
  44. Next,
  45. VolumeUp,
  46. VolumeDown,
  47. Shutdown,
  48. }
  49. const CONTEXT_TRACKS_HISTORY: usize = 10;
  50. const CONTEXT_FETCH_THRESHOLD: u32 = 5;
  51. pub struct Spirc {
  52. commands: mpsc::UnboundedSender<SpircCommand>,
  53. }
  54. fn initial_state() -> State {
  55. let mut frame = protocol::spirc::State::new();
  56. frame.set_repeat(false);
  57. frame.set_shuffle(false);
  58. frame.set_status(PlayStatus::kPlayStatusStop);
  59. frame.set_position_ms(0);
  60. frame.set_position_measured_at(0);
  61. frame
  62. }
  63. fn initial_device_state(config: ConnectConfig) -> DeviceState {
  64. {
  65. let mut msg = DeviceState::new();
  66. msg.set_sw_version(version::version_string());
  67. msg.set_is_active(false);
  68. msg.set_can_play(true);
  69. msg.set_volume(0);
  70. msg.set_name(config.name);
  71. {
  72. let repeated = msg.mut_capabilities();
  73. {
  74. let msg = repeated.push_default();
  75. msg.set_typ(protocol::spirc::CapabilityType::kCanBePlayer);
  76. {
  77. let repeated = msg.mut_intValue();
  78. repeated.push(1)
  79. };
  80. msg
  81. };
  82. {
  83. let msg = repeated.push_default();
  84. msg.set_typ(protocol::spirc::CapabilityType::kDeviceType);
  85. {
  86. let repeated = msg.mut_intValue();
  87. repeated.push(config.device_type as i64)
  88. };
  89. msg
  90. };
  91. {
  92. let msg = repeated.push_default();
  93. msg.set_typ(protocol::spirc::CapabilityType::kGaiaEqConnectId);
  94. {
  95. let repeated = msg.mut_intValue();
  96. repeated.push(1)
  97. };
  98. msg
  99. };
  100. {
  101. let msg = repeated.push_default();
  102. msg.set_typ(protocol::spirc::CapabilityType::kSupportsLogout);
  103. {
  104. let repeated = msg.mut_intValue();
  105. repeated.push(0)
  106. };
  107. msg
  108. };
  109. {
  110. let msg = repeated.push_default();
  111. msg.set_typ(protocol::spirc::CapabilityType::kIsObservable);
  112. {
  113. let repeated = msg.mut_intValue();
  114. repeated.push(1)
  115. };
  116. msg
  117. };
  118. {
  119. let msg = repeated.push_default();
  120. msg.set_typ(protocol::spirc::CapabilityType::kVolumeSteps);
  121. {
  122. let repeated = msg.mut_intValue();
  123. repeated.push(64)
  124. };
  125. msg
  126. };
  127. {
  128. let msg = repeated.push_default();
  129. msg.set_typ(protocol::spirc::CapabilityType::kSupportsPlaylistV2);
  130. {
  131. let repeated = msg.mut_intValue();
  132. repeated.push(64)
  133. };
  134. msg
  135. };
  136. {
  137. let msg = repeated.push_default();
  138. msg.set_typ(protocol::spirc::CapabilityType::kSupportedContexts);
  139. {
  140. let repeated = msg.mut_stringValue();
  141. repeated.push(::std::convert::Into::into("album"));
  142. repeated.push(::std::convert::Into::into("playlist"));
  143. repeated.push(::std::convert::Into::into("search"));
  144. repeated.push(::std::convert::Into::into("inbox"));
  145. repeated.push(::std::convert::Into::into("toplist"));
  146. repeated.push(::std::convert::Into::into("starred"));
  147. repeated.push(::std::convert::Into::into("publishedstarred"));
  148. repeated.push(::std::convert::Into::into("track"))
  149. };
  150. msg
  151. };
  152. {
  153. let msg = repeated.push_default();
  154. msg.set_typ(protocol::spirc::CapabilityType::kSupportedTypes);
  155. {
  156. let repeated = msg.mut_stringValue();
  157. repeated.push(::std::convert::Into::into("audio/local"));
  158. repeated.push(::std::convert::Into::into("audio/track"));
  159. repeated.push(::std::convert::Into::into("audio/episode"));
  160. repeated.push(::std::convert::Into::into("local"));
  161. repeated.push(::std::convert::Into::into("track"))
  162. };
  163. msg
  164. };
  165. };
  166. msg
  167. }
  168. }
  169. fn calc_logarithmic_volume(volume: u16) -> u16 {
  170. // Volume conversion taken from https://www.dr-lex.be/info-stuff/volumecontrols.html#ideal2
  171. // Convert the given volume [0..0xffff] to a dB gain
  172. // We assume a dB range of 60dB.
  173. // Use the equation: a * exp(b * x)
  174. // in which a = IDEAL_FACTOR, b = 1/1000
  175. const IDEAL_FACTOR: f64 = 6.908;
  176. let normalized_volume = volume as f64 / std::u16::MAX as f64; // To get a value between 0 and 1
  177. let mut val = std::u16::MAX;
  178. // Prevent val > std::u16::MAX due to rounding errors
  179. if normalized_volume < 0.999 {
  180. let new_volume = (normalized_volume * IDEAL_FACTOR).exp() / 1000.0;
  181. val = (new_volume * std::u16::MAX as f64) as u16;
  182. }
  183. debug!("input volume:{} to mixer: {}", volume, val);
  184. // return the scale factor (0..0xffff) (equivalent to a voltage multiplier).
  185. val
  186. }
  187. fn volume_to_mixer(volume: u16, linear_volume: bool) -> u16 {
  188. if linear_volume {
  189. debug!("linear volume: {}", volume);
  190. volume
  191. } else {
  192. calc_logarithmic_volume(volume)
  193. }
  194. }
  195. impl Spirc {
  196. pub fn new(
  197. config: ConnectConfig,
  198. session: Session,
  199. player: Player,
  200. mixer: Box<Mixer>,
  201. ) -> (Spirc, SpircTask) {
  202. debug!("new Spirc[{}]", session.session_id());
  203. let ident = session.device_id().to_owned();
  204. // Uri updated in response to issue #288
  205. let uri = format!("hm://remote/user/{}/", session.username());
  206. let subscription = session.mercury().subscribe(&uri as &str);
  207. let subscription = subscription
  208. .map(|stream| stream.map_err(|_| MercuryError))
  209. .flatten_stream();
  210. let subscription = Box::new(subscription.map(|response| -> Frame {
  211. let data = response.payload.first().unwrap();
  212. protobuf::parse_from_bytes(data).unwrap()
  213. }));
  214. let sender = Box::new(
  215. session
  216. .mercury()
  217. .sender(uri)
  218. .with(|frame: Frame| Ok(frame.write_to_bytes().unwrap())),
  219. );
  220. let (cmd_tx, cmd_rx) = mpsc::unbounded();
  221. let volume = config.volume;
  222. let linear_volume = config.linear_volume;
  223. let device = initial_device_state(config);
  224. let mut task = SpircTask {
  225. player: player,
  226. mixer: mixer,
  227. linear_volume: linear_volume,
  228. sequence: SeqGenerator::new(1),
  229. ident: ident,
  230. device: device,
  231. state: initial_state(),
  232. subscription: subscription,
  233. sender: sender,
  234. commands: cmd_rx,
  235. end_of_track: Box::new(future::empty()),
  236. shutdown: false,
  237. session: session.clone(),
  238. context_fut: Box::new(future::empty()),
  239. context: None,
  240. };
  241. task.set_volume(volume);
  242. let spirc = Spirc { commands: cmd_tx };
  243. task.hello();
  244. (spirc, task)
  245. }
  246. pub fn play(&self) {
  247. let _ = self.commands.unbounded_send(SpircCommand::Play);
  248. }
  249. pub fn play_pause(&self) {
  250. let _ = self.commands.unbounded_send(SpircCommand::PlayPause);
  251. }
  252. pub fn pause(&self) {
  253. let _ = self.commands.unbounded_send(SpircCommand::Pause);
  254. }
  255. pub fn prev(&self) {
  256. let _ = self.commands.unbounded_send(SpircCommand::Prev);
  257. }
  258. pub fn next(&self) {
  259. let _ = self.commands.unbounded_send(SpircCommand::Next);
  260. }
  261. pub fn volume_up(&self) {
  262. let _ = self.commands.unbounded_send(SpircCommand::VolumeUp);
  263. }
  264. pub fn volume_down(&self) {
  265. let _ = self.commands.unbounded_send(SpircCommand::VolumeDown);
  266. }
  267. pub fn shutdown(&self) {
  268. let _ = self.commands.unbounded_send(SpircCommand::Shutdown);
  269. }
  270. }
  271. impl Future for SpircTask {
  272. type Item = ();
  273. type Error = ();
  274. fn poll(&mut self) -> Poll<(), ()> {
  275. loop {
  276. let mut progress = false;
  277. if self.session.is_invalid() {
  278. return Ok(Async::Ready(()));
  279. }
  280. if !self.shutdown {
  281. match self.subscription.poll().unwrap() {
  282. Async::Ready(Some(frame)) => {
  283. progress = true;
  284. self.handle_frame(frame);
  285. }
  286. Async::Ready(None) => panic!("subscription terminated"),
  287. Async::NotReady => (),
  288. }
  289. match self.commands.poll().unwrap() {
  290. Async::Ready(Some(command)) => {
  291. progress = true;
  292. self.handle_command(command);
  293. }
  294. Async::Ready(None) => (),
  295. Async::NotReady => (),
  296. }
  297. match self.end_of_track.poll() {
  298. Ok(Async::Ready(())) => {
  299. progress = true;
  300. self.handle_end_of_track();
  301. }
  302. Ok(Async::NotReady) => (),
  303. Err(oneshot::Canceled) => self.end_of_track = Box::new(future::empty()),
  304. }
  305. match self.context_fut.poll() {
  306. Ok(Async::Ready(value)) => {
  307. let r_context = serde_json::from_value::<StationContext>(value.clone());
  308. self.context = match r_context {
  309. Ok(context) => {
  310. info!(
  311. "Resolved {:?} tracks from <{:?}>",
  312. context.tracks.len(),
  313. self.state.get_context_uri(),
  314. );
  315. Some(context)
  316. }
  317. Err(e) => {
  318. error!("Unable to parse JSONContext {:?}\n{:?}", e, value);
  319. None
  320. }
  321. };
  322. // It needn't be so verbose - can be as simple as
  323. // if let Some(ref context) = r_context {
  324. // info!("Got {:?} tracks from <{}>", context.tracks.len(), context.uri);
  325. // }
  326. // self.context = r_context;
  327. progress = true;
  328. self.context_fut = Box::new(future::empty());
  329. }
  330. Ok(Async::NotReady) => (),
  331. Err(err) => {
  332. self.context_fut = Box::new(future::empty());
  333. error!("ContextError: {:?}", err)
  334. }
  335. }
  336. }
  337. let poll_sender = self.sender.poll_complete().unwrap();
  338. // Only shutdown once we've flushed out all our messages
  339. if self.shutdown && poll_sender.is_ready() {
  340. return Ok(Async::Ready(()));
  341. }
  342. if !progress {
  343. return Ok(Async::NotReady);
  344. }
  345. }
  346. }
  347. }
  348. impl SpircTask {
  349. fn now_ms(&mut self) -> i64 {
  350. let dur = match SystemTime::now().duration_since(UNIX_EPOCH) {
  351. Ok(dur) => dur,
  352. Err(err) => err.duration(),
  353. };
  354. ((dur.as_secs() as i64 + self.session.time_delta()) * 1000
  355. + (dur.subsec_nanos() / 1000_000) as i64)
  356. }
  357. fn handle_command(&mut self, cmd: SpircCommand) {
  358. let active = self.device.get_is_active();
  359. match cmd {
  360. SpircCommand::Play => {
  361. if active {
  362. self.handle_play();
  363. self.notify(None);
  364. } else {
  365. CommandSender::new(self, MessageType::kMessageTypePlay).send();
  366. }
  367. }
  368. SpircCommand::PlayPause => {
  369. if active {
  370. self.handle_play_pause();
  371. self.notify(None);
  372. } else {
  373. CommandSender::new(self, MessageType::kMessageTypePlayPause).send();
  374. }
  375. }
  376. SpircCommand::Pause => {
  377. if active {
  378. self.handle_pause();
  379. self.notify(None);
  380. } else {
  381. CommandSender::new(self, MessageType::kMessageTypePause).send();
  382. }
  383. }
  384. SpircCommand::Prev => {
  385. if active {
  386. self.handle_prev();
  387. self.notify(None);
  388. } else {
  389. CommandSender::new(self, MessageType::kMessageTypePrev).send();
  390. }
  391. }
  392. SpircCommand::Next => {
  393. if active {
  394. self.handle_next();
  395. self.notify(None);
  396. } else {
  397. CommandSender::new(self, MessageType::kMessageTypeNext).send();
  398. }
  399. }
  400. SpircCommand::VolumeUp => {
  401. if active {
  402. self.handle_volume_up();
  403. self.notify(None);
  404. } else {
  405. CommandSender::new(self, MessageType::kMessageTypeVolumeUp).send();
  406. }
  407. }
  408. SpircCommand::VolumeDown => {
  409. if active {
  410. self.handle_volume_down();
  411. self.notify(None);
  412. } else {
  413. CommandSender::new(self, MessageType::kMessageTypeVolumeDown).send();
  414. }
  415. }
  416. SpircCommand::Shutdown => {
  417. CommandSender::new(self, MessageType::kMessageTypeGoodbye).send();
  418. self.shutdown = true;
  419. self.commands.close();
  420. }
  421. }
  422. }
  423. fn handle_frame(&mut self, frame: Frame) {
  424. debug!(
  425. "{:?} {:?} {} {} {}",
  426. frame.get_typ(),
  427. frame.get_device_state().get_name(),
  428. frame.get_ident(),
  429. frame.get_seq_nr(),
  430. frame.get_state_update_id()
  431. );
  432. if frame.get_ident() == self.ident
  433. || (frame.get_recipient().len() > 0 && !frame.get_recipient().contains(&self.ident))
  434. {
  435. return;
  436. }
  437. match frame.get_typ() {
  438. MessageType::kMessageTypeHello => {
  439. self.notify(Some(frame.get_ident()));
  440. }
  441. MessageType::kMessageTypeLoad => {
  442. if !self.device.get_is_active() {
  443. let now = self.now_ms();
  444. self.device.set_is_active(true);
  445. self.device.set_became_active_at(now);
  446. }
  447. self.update_tracks(&frame);
  448. if self.state.get_track().len() > 0 {
  449. let now = self.now_ms();
  450. self.state.set_position_ms(frame.get_state().get_position_ms());
  451. self.state.set_position_measured_at(now as u64);
  452. let play = frame.get_state().get_status() == PlayStatus::kPlayStatusPlay;
  453. self.load_track(play);
  454. } else {
  455. info!("No more tracks left in queue");
  456. self.state.set_status(PlayStatus::kPlayStatusStop);
  457. }
  458. self.notify(None);
  459. }
  460. MessageType::kMessageTypePlay => {
  461. self.handle_play();
  462. self.notify(None);
  463. }
  464. MessageType::kMessageTypePlayPause => {
  465. self.handle_play_pause();
  466. self.notify(None);
  467. }
  468. MessageType::kMessageTypePause => {
  469. self.handle_pause();
  470. self.notify(None);
  471. }
  472. MessageType::kMessageTypeNext => {
  473. self.handle_next();
  474. self.notify(None);
  475. }
  476. MessageType::kMessageTypePrev => {
  477. self.handle_prev();
  478. self.notify(None);
  479. }
  480. MessageType::kMessageTypeVolumeUp => {
  481. self.handle_volume_up();
  482. self.notify(None);
  483. }
  484. MessageType::kMessageTypeVolumeDown => {
  485. self.handle_volume_down();
  486. self.notify(None);
  487. }
  488. MessageType::kMessageTypeRepeat => {
  489. self.state.set_repeat(frame.get_state().get_repeat());
  490. self.notify(None);
  491. }
  492. MessageType::kMessageTypeShuffle => {
  493. self.state.set_shuffle(frame.get_state().get_shuffle());
  494. if self.state.get_shuffle() {
  495. let current_index = self.state.get_playing_track_index();
  496. {
  497. let tracks = self.state.mut_track();
  498. tracks.swap(0, current_index as usize);
  499. if let Some((_, rest)) = tracks.split_first_mut() {
  500. let mut rng = rand::thread_rng();
  501. rest.shuffle(&mut rng);
  502. }
  503. }
  504. self.state.set_playing_track_index(0);
  505. } else {
  506. let context = self.state.get_context_uri();
  507. debug!("{:?}", context);
  508. }
  509. self.notify(None);
  510. }
  511. MessageType::kMessageTypeSeek => {
  512. let position = frame.get_position();
  513. let now = self.now_ms();
  514. self.state.set_position_ms(position);
  515. self.state.set_position_measured_at(now as u64);
  516. self.player.seek(position);
  517. self.notify(None);
  518. }
  519. MessageType::kMessageTypeReplace => {
  520. self.update_tracks(&frame);
  521. self.notify(None);
  522. }
  523. MessageType::kMessageTypeVolume => {
  524. self.set_volume(frame.get_volume() as u16);
  525. self.notify(None);
  526. }
  527. MessageType::kMessageTypeNotify => {
  528. if self.device.get_is_active() && frame.get_device_state().get_is_active() {
  529. self.device.set_is_active(false);
  530. self.state.set_status(PlayStatus::kPlayStatusStop);
  531. self.player.stop();
  532. self.mixer.stop();
  533. }
  534. }
  535. _ => (),
  536. }
  537. }
  538. fn handle_play(&mut self) {
  539. if self.state.get_status() == PlayStatus::kPlayStatusPause {
  540. self.mixer.start();
  541. self.player.play();
  542. self.state.set_status(PlayStatus::kPlayStatusPlay);
  543. let now = self.now_ms();
  544. self.state.set_position_measured_at(now as u64);
  545. }
  546. }
  547. fn handle_play_pause(&mut self) {
  548. match self.state.get_status() {
  549. PlayStatus::kPlayStatusPlay => self.handle_pause(),
  550. PlayStatus::kPlayStatusPause => self.handle_play(),
  551. _ => (),
  552. }
  553. }
  554. fn handle_pause(&mut self) {
  555. if self.state.get_status() == PlayStatus::kPlayStatusPlay {
  556. self.player.pause();
  557. self.mixer.stop();
  558. self.state.set_status(PlayStatus::kPlayStatusPause);
  559. let now = self.now_ms() as u64;
  560. let position = self.state.get_position_ms();
  561. let diff = now - self.state.get_position_measured_at();
  562. self.state.set_position_ms(position + diff as u32);
  563. self.state.set_position_measured_at(now);
  564. }
  565. }
  566. fn consume_queued_track(&mut self) -> usize {
  567. // Removes current track if it is queued
  568. // Returns the index of the next track
  569. let current_index = self.state.get_playing_track_index() as usize;
  570. if self.state.get_track()[current_index].get_queued() {
  571. self.state.mut_track().remove(current_index);
  572. return current_index;
  573. }
  574. current_index + 1
  575. }
  576. fn handle_next(&mut self) {
  577. let mut new_index = self.consume_queued_track() as u32;
  578. let mut continue_playing = true;
  579. debug!(
  580. "At track {:?} of {:?} <{:?}> update [{}]",
  581. new_index,
  582. self.state.get_track().len(),
  583. self.state.get_context_uri(),
  584. self.state.get_track().len() as u32 - new_index < CONTEXT_FETCH_THRESHOLD
  585. );
  586. let context_uri = self.state.get_context_uri().to_owned();
  587. if (context_uri.starts_with("spotify:station:") || context_uri.starts_with("spotify:dailymix:"))
  588. && ((self.state.get_track().len() as u32) - new_index) < CONTEXT_FETCH_THRESHOLD
  589. {
  590. self.context_fut = self.resolve_station(&context_uri);
  591. self.update_tracks_from_context();
  592. }
  593. if new_index >= self.state.get_track().len() as u32 {
  594. new_index = 0; // Loop around back to start
  595. continue_playing = self.state.get_repeat();
  596. }
  597. self.state.set_playing_track_index(new_index);
  598. self.state.set_position_ms(0);
  599. let now = self.now_ms();
  600. self.state.set_position_measured_at(now as u64);
  601. self.load_track(continue_playing);
  602. }
  603. fn handle_prev(&mut self) {
  604. // Previous behaves differently based on the position
  605. // Under 3s it goes to the previous song (starts playing)
  606. // Over 3s it seeks to zero (retains previous play status)
  607. if self.position() < 3000 {
  608. // Queued tracks always follow the currently playing track.
  609. // They should not be considered when calculating the previous
  610. // track so extract them beforehand and reinsert them after it.
  611. let mut queue_tracks = Vec::new();
  612. {
  613. let queue_index = self.consume_queued_track();
  614. let tracks = self.state.mut_track();
  615. while queue_index < tracks.len() && tracks[queue_index].get_queued() {
  616. queue_tracks.push(tracks.remove(queue_index));
  617. }
  618. }
  619. let current_index = self.state.get_playing_track_index();
  620. let new_index = if current_index > 0 {
  621. current_index - 1
  622. } else if self.state.get_repeat() {
  623. self.state.get_track().len() as u32 - 1
  624. } else {
  625. 0
  626. };
  627. // Reinsert queued tracks after the new playing track.
  628. let mut pos = (new_index + 1) as usize;
  629. for track in queue_tracks.into_iter() {
  630. self.state.mut_track().insert(pos, track);
  631. pos += 1;
  632. }
  633. let now = self.now_ms();
  634. self.state.set_playing_track_index(new_index);
  635. self.state.set_position_ms(0);
  636. self.state.set_position_measured_at(now as u64);
  637. self.load_track(true);
  638. } else {
  639. let now = self.now_ms();
  640. self.state.set_position_ms(0);
  641. self.state.set_position_measured_at(now as u64);
  642. self.player.seek(0);
  643. }
  644. }
  645. fn handle_volume_up(&mut self) {
  646. let mut volume: u32 = self.device.get_volume() as u32 + 4096;
  647. if volume > 0xFFFF {
  648. volume = 0xFFFF;
  649. }
  650. self.set_volume(volume as u16);
  651. }
  652. fn handle_volume_down(&mut self) {
  653. let mut volume: i32 = self.device.get_volume() as i32 - 4096;
  654. if volume < 0 {
  655. volume = 0;
  656. }
  657. self.set_volume(volume as u16);
  658. }
  659. fn handle_end_of_track(&mut self) {
  660. self.handle_next();
  661. self.notify(None);
  662. }
  663. fn position(&mut self) -> u32 {
  664. let diff = self.now_ms() as u64 - self.state.get_position_measured_at();
  665. self.state.get_position_ms() + diff as u32
  666. }
  667. fn resolve_station(&self, uri: &str) -> Box<Future<Item = serde_json::Value, Error = MercuryError>> {
  668. let radio_uri = format!("hm://radio-apollo/v3/stations/{}", uri);
  669. self.resolve_uri(&radio_uri)
  670. }
  671. fn resolve_uri(&self, uri: &str) -> Box<Future<Item = serde_json::Value, Error = MercuryError>> {
  672. let request = self.session.mercury().get(uri);
  673. Box::new(request.and_then(move |response| {
  674. let data = response.payload.first().expect("Empty payload on context uri");
  675. let response: serde_json::Value = serde_json::from_slice(&data).unwrap();
  676. Ok(response)
  677. }))
  678. }
  679. fn update_tracks_from_context(&mut self) {
  680. if let Some(ref context) = self.context {
  681. self.context_fut = self.resolve_uri(&context.next_page_url);
  682. let new_tracks = &context.tracks;
  683. debug!("Adding {:?} tracks from context to frame", new_tracks.len());
  684. let mut track_vec = self.state.take_track().into_vec();
  685. if let Some(head) = track_vec.len().checked_sub(CONTEXT_TRACKS_HISTORY) {
  686. track_vec.drain(0..head);
  687. }
  688. track_vec.extend_from_slice(&new_tracks);
  689. self.state.set_track(protobuf::RepeatedField::from_vec(track_vec));
  690. // Update playing index
  691. if let Some(new_index) = self
  692. .state
  693. .get_playing_track_index()
  694. .checked_sub(CONTEXT_TRACKS_HISTORY as u32)
  695. {
  696. self.state.set_playing_track_index(new_index);
  697. }
  698. }
  699. }
  700. fn update_tracks(&mut self, frame: &protocol::spirc::Frame) {
  701. debug!("State: {:?}", frame.get_state());
  702. let index = frame.get_state().get_playing_track_index();
  703. let context_uri = frame.get_state().get_context_uri().to_owned();
  704. let tracks = frame.get_state().get_track();
  705. debug!("Frame has {:?} tracks", tracks.len());
  706. if context_uri.starts_with("spotify:station:") || context_uri.starts_with("spotify:dailymix:") {
  707. self.context_fut = self.resolve_station(&context_uri);
  708. }
  709. self.state.set_playing_track_index(index);
  710. self.state.set_track(tracks.into_iter().cloned().collect());
  711. self.state.set_context_uri(context_uri);
  712. self.state.set_repeat(frame.get_state().get_repeat());
  713. self.state.set_shuffle(frame.get_state().get_shuffle());
  714. }
  715. // should this be a method of SpotifyId directly?
  716. fn get_spotify_id_for_track(&self, track_ref: &TrackRef) -> Result<SpotifyId, SpotifyIdError> {
  717. SpotifyId::from_raw(track_ref.get_gid()).or_else(|_| {
  718. let uri = track_ref.get_uri();
  719. debug!("Malformed or no gid, attempting to parse URI <{}>", uri);
  720. SpotifyId::from_uri(uri)
  721. })
  722. }
  723. fn load_track(&mut self, play: bool) {
  724. let context_uri = self.state.get_context_uri().to_owned();
  725. let mut index = self.state.get_playing_track_index();
  726. let start_index = index;
  727. let tracks_len = self.state.get_track().len() as u32;
  728. debug!(
  729. "Loading context: <{}> index: [{}] of {}",
  730. context_uri, index, tracks_len
  731. );
  732. // Cycle through all tracks, break if we don't find any playable tracks
  733. // TODO: This will panic if no playable tracks are found!
  734. // tracks in each frame either have a gid or uri (that may or may not be a valid track)
  735. // E.g - context based frames sometimes contain tracks with <spotify:meta:page:>
  736. let track = {
  737. let mut track_ref = self.state.get_track()[index as usize].clone();
  738. let mut track_id = self.get_spotify_id_for_track(&track_ref);
  739. while track_id.is_err() || track_id.unwrap().audio_type == SpotifyAudioType::NonPlayable {
  740. warn!(
  741. "Skipping track <{:?}> at position [{}] of {}",
  742. track_ref.get_uri(),
  743. index,
  744. tracks_len
  745. );
  746. index = if index + 1 < tracks_len { index + 1 } else { 0 };
  747. self.state.set_playing_track_index(index);
  748. if index == start_index {
  749. warn!("No playable track found in state: {:?}", self.state);
  750. break;
  751. }
  752. track_ref = self.state.get_track()[index as usize].clone();
  753. track_id = self.get_spotify_id_for_track(&track_ref);
  754. }
  755. track_id
  756. }
  757. .expect("Invalid SpotifyId");
  758. let position = self.state.get_position_ms();
  759. let end_of_track = self.player.load(track, play, position);
  760. if play {
  761. self.state.set_status(PlayStatus::kPlayStatusPlay);
  762. } else {
  763. self.state.set_status(PlayStatus::kPlayStatusPause);
  764. }
  765. self.end_of_track = Box::new(end_of_track);
  766. }
  767. fn hello(&mut self) {
  768. CommandSender::new(self, MessageType::kMessageTypeHello).send();
  769. }
  770. fn notify(&mut self, recipient: Option<&str>) {
  771. let mut cs = CommandSender::new(self, MessageType::kMessageTypeNotify);
  772. if let Some(s) = recipient {
  773. cs = cs.recipient(&s);
  774. }
  775. cs.send();
  776. }
  777. fn set_volume(&mut self, volume: u16) {
  778. self.device.set_volume(volume as u32);
  779. self.mixer.set_volume(volume_to_mixer(volume, self.linear_volume));
  780. if let Some(cache) = self.session.cache() {
  781. cache.save_volume(Volume { volume })
  782. }
  783. }
  784. }
  785. impl Drop for SpircTask {
  786. fn drop(&mut self) {
  787. debug!("drop Spirc[{}]", self.session.session_id());
  788. }
  789. }
  790. struct CommandSender<'a> {
  791. spirc: &'a mut SpircTask,
  792. frame: protocol::spirc::Frame,
  793. }
  794. impl<'a> CommandSender<'a> {
  795. fn new(spirc: &'a mut SpircTask, cmd: MessageType) -> CommandSender {
  796. let mut frame = protocol::spirc::Frame::new();
  797. frame.set_version(1);
  798. frame.set_protocol_version(::std::convert::Into::into("2.0.0"));
  799. frame.set_ident(spirc.ident.clone());
  800. frame.set_seq_nr(spirc.sequence.get());
  801. frame.set_typ(cmd);
  802. frame.set_device_state(spirc.device.clone());
  803. frame.set_state_update_id(spirc.now_ms());
  804. CommandSender {
  805. spirc: spirc,
  806. frame: frame,
  807. }
  808. }
  809. fn recipient(mut self, recipient: &'a str) -> CommandSender {
  810. self.frame.mut_recipient().push(recipient.to_owned());
  811. self
  812. }
  813. #[allow(dead_code)]
  814. fn state(mut self, state: protocol::spirc::State) -> CommandSender<'a> {
  815. self.frame.set_state(state);
  816. self
  817. }
  818. fn send(mut self) {
  819. if !self.frame.has_state() && self.spirc.device.get_is_active() {
  820. self.frame.set_state(self.spirc.state.clone());
  821. }
  822. let send = self.spirc.sender.start_send(self.frame).unwrap();
  823. assert!(send.is_ready());
  824. }
  825. }