From 05118b40f8c8f9362432f9c7c47e6ce2927ce61c Mon Sep 17 00:00:00 2001
From: Paul Lietar <paul@lietar.net>
Date: Thu, 19 Jan 2017 12:56:49 +0000
Subject: [PATCH] Move MetadataManager to use tokio

---
 src/audio_key.rs   | 25 +++++++++++---
 src/lib.rs         |  1 -
 src/link.rs        | 83 ----------------------------------------------
 src/mercury/mod.rs |  7 ++--
 src/metadata.rs    | 29 ++++++----------
 src/player.rs      | 17 +++++-----
 src/session.rs     | 16 ++++-----
 7 files changed, 49 insertions(+), 129 deletions(-)
 delete mode 100644 src/link.rs

diff --git a/src/audio_key.rs b/src/audio_key.rs
index 98f52243..eab364f3 100644
--- a/src/audio_key.rs
+++ b/src/audio_key.rs
@@ -1,5 +1,6 @@
 use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
 use futures::sync::oneshot;
+use futures::{Async, Future, Poll};
 use std::collections::HashMap;
 use std::io::Write;
 
@@ -12,12 +13,10 @@ pub struct AudioKey(pub [u8; 16]);
 #[derive(Debug,Hash,PartialEq,Eq,Copy,Clone)]
 pub struct AudioKeyError;
 
-pub type Result = ::std::result::Result<AudioKey, AudioKeyError>;
-
 component! {
     AudioKeyManager : AudioKeyManagerInner {
         sequence: SeqGenerator<u32> = SeqGenerator::new(0),
-        pending: HashMap<u32, oneshot::Sender<Result>> = HashMap::new(),
+        pending: HashMap<u32, oneshot::Sender<Result<AudioKey, AudioKeyError>>> = HashMap::new(),
     }
 }
 
@@ -43,7 +42,7 @@ impl AudioKeyManager {
         }
     }
 
-    pub fn request<'a>(&self, track: SpotifyId, file: FileId) -> oneshot::Receiver<Result> {
+    pub fn request<'a>(&self, track: SpotifyId, file: FileId) -> AudioKeyFuture<AudioKey> {
         let (tx, rx) = oneshot::channel();
 
         let seq = self.lock(move |inner| {
@@ -53,7 +52,7 @@ impl AudioKeyManager {
         });
 
         self.send_key_request(seq, track, file);
-        rx
+        AudioKeyFuture(rx)
     }
 
     fn send_key_request<'a>(&self, seq: u32, track: SpotifyId, file: FileId) {
@@ -66,3 +65,19 @@ impl AudioKeyManager {
         self.session().send_packet(0xc, data)
     }
 }
+
+pub struct AudioKeyFuture<T>(oneshot::Receiver<Result<T, AudioKeyError>>);
+impl <T> Future for AudioKeyFuture<T> {
+    type Item = T;
+    type Error = AudioKeyError;
+
+    fn poll(&mut self) -> Poll<T, AudioKeyError> {
+        match self.0.poll() {
+            Ok(Async::Ready(Ok(value))) => Ok(Async::Ready(value)),
+            Ok(Async::Ready(Err(err))) => Err(err),
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Err(oneshot::Canceled) => Err(AudioKeyError),
+        }
+    }
+}
+
diff --git a/src/lib.rs b/src/lib.rs
index 73f9123f..21e445f1 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -58,7 +58,6 @@ pub mod audio_file;
 pub mod audio_key;
 pub mod cache;
 pub mod diffie_hellman;
-pub mod link;
 pub mod mercury;
 pub mod metadata;
 pub mod player;
diff --git a/src/link.rs b/src/link.rs
deleted file mode 100644
index 2bd8e68c..00000000
--- a/src/link.rs
+++ /dev/null
@@ -1,83 +0,0 @@
-use util::SpotifyId;
-use session::Session;
-use metadata::{MetadataRef, Album, Artist, Track};
-
-#[derive(Debug,Clone)]
-pub enum Link {
-    Track {
-        id: SpotifyId,
-        offset: u32,
-    },
-
-    Album {
-        id: SpotifyId,
-    },
-
-    Artist {
-        id: SpotifyId,
-    },
-
-    /*
-    Search,
-    Playlist,
-    Profile,
-    Starred,
-    LocalTrack,
-    Image,
-    */
-}
-
-impl Link {
-    pub fn from_str(uri: &str) -> Result<Link, ()> {
-        let mut parts = uri.split(':');
-
-        if parts.next() != Some("spotify") {
-            return Err(())
-        }
-
-        match parts.next() {
-            Some("track") => parts.next()
-                                  .map(SpotifyId::from_base62)
-                                  .map(|id| Link::Track {
-                                      id: id,
-                                      offset: 0,
-                                  })
-                                  .ok_or(()),
-            Some("album") => parts.next()
-                                  .map(SpotifyId::from_base62)
-                                  .map(|id| Link::Album {
-                                      id: id,
-                                  })
-                                  .ok_or(()),
-            Some("artist") => parts.next()
-                                   .map(SpotifyId::from_base62)
-                                   .map(|id| Link::Artist {
-                                       id: id,
-                                   })
-                                   .ok_or(()),
-            _ => Err(())
-        }
-    }
-
-    pub fn as_track(&self, session: &Session) -> Option<MetadataRef<Track>> {
-        match *self {
-            Link::Track { id, .. } => Some(session.metadata::<Track>(id)),
-            _ => None,
-        }
-    }
-
-    pub fn as_album(&self, session: &Session) -> Option<MetadataRef<Album>> {
-        match *self {
-            Link::Album { id, .. } => Some(session.metadata::<Album>(id)),
-            _ => None,
-        }
-    }
-
-    pub fn as_artist(&self, session: &Session) -> Option<MetadataRef<Artist>> {
-        match *self {
-            Link::Artist { id, .. } => Some(session.metadata::<Artist>(id)),
-            _ => None,
-        }
-    }
-}
-
diff --git a/src/mercury/mod.rs b/src/mercury/mod.rs
index d2ce23c0..077bcb69 100644
--- a/src/mercury/mod.rs
+++ b/src/mercury/mod.rs
@@ -1,12 +1,11 @@
 use byteorder::{BigEndian, ByteOrder, ReadBytesExt};
 use futures::sync::{oneshot, mpsc};
-use futures::{BoxFuture, Future};
+use futures::{Async, Poll, BoxFuture, Future};
+use protobuf;
+use protocol;
 use std::collections::HashMap;
 use std::io::Read;
 use std::mem;
-use protocol;
-use protobuf;
-use futures::{Async, Poll};
 
 use util::SeqGenerator;
 
diff --git a/src/metadata.rs b/src/metadata.rs
index c59a4063..43094e5c 100644
--- a/src/metadata.rs
+++ b/src/metadata.rs
@@ -1,11 +1,11 @@
-use eventual::Future;
+use futures::{Future, BoxFuture};
 use linear_map::LinearMap;
 use protobuf;
-use futures::Future as Future_;
 
+use mercury::MercuryError;
 use protocol;
-use util::{SpotifyId, FileId, StrChunksExt};
 use session::Session;
+use util::{SpotifyId, FileId, StrChunksExt};
 
 pub use protocol::metadata::AudioFile_Format as FileFormat;
 
@@ -59,11 +59,6 @@ pub struct Artist {
     pub top_tracks: Vec<SpotifyId>,
 }
 
-pub type MetadataRef<T> = Future<T, ()>;
-pub type TrackRef = MetadataRef<Track>;
-pub type AlbumRef = MetadataRef<Album>;
-pub type ArtistRef = MetadataRef<Artist>;
-
 impl MetadataTrait for Track {
     type Message = protocol::metadata::Track;
 
@@ -180,26 +175,22 @@ impl MetadataTrait for Artist {
     }
 }
 
-pub struct MetadataManager;
+component! {
+    MetadataManager : MetadataManagerInner { }
+}
 
 impl MetadataManager {
-    pub fn new() -> MetadataManager {
-        MetadataManager
-    }
-
-    pub fn get<T: MetadataTrait>(&mut self, session: &Session, id: SpotifyId) -> MetadataRef<T> {
-        let session = session.clone();
+    pub fn get<T: MetadataTrait>(&self, id: SpotifyId) -> BoxFuture<T, MercuryError> {
+        let session = self.session();
 
         let uri = format!("{}/{}", T::base_url(), id.to_base16());
         let request = session.mercury().get(uri);
 
-        let result = request.and_then(move |response| {
+        request.and_then(move |response| {
             let data = response.payload.first().expect("Empty payload");
             let msg: T::Message = protobuf::parse_from_bytes(data).unwrap();
 
             Ok(T::parse(&msg, &session))
-        }).wait();
-
-        Future::of(result.unwrap())
+        }).boxed()
     }
 }
diff --git a/src/player.rs b/src/player.rs
index b3f948a4..0220b5a3 100644
--- a/src/player.rs
+++ b/src/player.rs
@@ -1,14 +1,13 @@
-use eventual::{self, Async};
 use std::borrow::Cow;
 use std::sync::{mpsc, Mutex, Arc, MutexGuard};
 use std::thread;
 use std::io::{Read, Seek};
 use vorbis;
-use futures::Future;
+use futures::{future, Future};
 
 use audio_decrypt::AudioDecrypt;
 use audio_backend::Sink;
-use metadata::{FileFormat, Track, TrackRef};
+use metadata::{FileFormat, Track};
 use session::{Bitrate, Session};
 use util::{self, ReadSeek, SpotifyId, Subfile};
 pub use spirc::PlayStatus;
@@ -170,16 +169,16 @@ fn find_available_alternative<'a>(session: &Session, track: &'a Track) -> Option
         let alternatives = track.alternatives
             .iter()
             .map(|alt_id| {
-                session.metadata::<Track>(*alt_id)
-            })
-        .collect::<Vec<TrackRef>>();
+                session.metadata().get::<Track>(*alt_id)
+            });
+        let alternatives = future::join_all(alternatives).wait().unwrap();
 
-        eventual::sequence(alternatives.into_iter()).iter().find(|alt| alt.available).map(Cow::Owned)
+        alternatives.into_iter().find(|alt| alt.available).map(Cow::Owned)
     }
 }
 
 fn load_track(session: &Session, track_id: SpotifyId) -> Option<vorbis::Decoder<Subfile<AudioDecrypt<Box<ReadSeek>>>>> {
-    let track = session.metadata::<Track>(track_id).await().unwrap();
+    let track = session.metadata().get::<Track>(track_id).wait().unwrap();
 
     info!("Loading track \"{}\"", track.name);
 
@@ -207,7 +206,7 @@ fn load_track(session: &Session, track_id: SpotifyId) -> Option<vorbis::Decoder<
         }
     };
 
-    let key = session.audio_key().request(track.id, file_id).wait().unwrap().unwrap();
+    let key = session.audio_key().request(track.id, file_id).wait().unwrap();
 
     let audio_file = Subfile::new(AudioDecrypt::new(key, session.audio_file(file_id)), 0xa7);
     let decoder = vorbis::Decoder::new(audio_file).unwrap();
diff --git a/src/session.rs b/src/session.rs
index 7255fd9f..195e3034 100644
--- a/src/session.rs
+++ b/src/session.rs
@@ -17,12 +17,12 @@ use audio_file::AudioFile;
 use authentication::Credentials;
 use cache::Cache;
 use connection::{self, adaptor};
-use metadata::{MetadataManager, MetadataRef, MetadataTrait};
 use stream::StreamManager;
-use util::{SpotifyId, FileId, ReadSeek, Lazy};
+use util::{FileId, ReadSeek, Lazy};
 
 use audio_key::AudioKeyManager;
 use mercury::MercuryManager;
+use metadata::MetadataManager;
 
 use stream;
 
@@ -63,13 +63,13 @@ pub struct SessionInternal {
     data: RwLock<SessionData>,
 
     cache: Box<Cache + Send + Sync>,
-    metadata: Mutex<MetadataManager>,
     stream: Mutex<StreamManager>,
     rx_connection: Mutex<adaptor::StreamAdaptor<(u8, Vec<u8>), io::Error>>,
     tx_connection: Mutex<adaptor::SinkAdaptor<(u8, Vec<u8>)>>,
 
     audio_key: Lazy<AudioKeyManager>,
     mercury: Lazy<MercuryManager>,
+    metadata: Lazy<MetadataManager>,
 }
 
 #[derive(Clone)]
@@ -130,11 +130,11 @@ impl Session {
             tx_connection: Mutex::new(tx),
 
             cache: cache,
-            metadata: Mutex::new(MetadataManager::new()),
             stream: Mutex::new(StreamManager::new()),
 
             audio_key: Lazy::new(),
             mercury: Lazy::new(),
+            metadata: Lazy::new(),
         }));
 
         (session, task)
@@ -148,6 +148,10 @@ impl Session {
         self.0.mercury.get(|| MercuryManager::new(self.weak()))
     }
 
+    pub fn metadata(&self) -> &MetadataManager {
+        self.0.metadata.get(|| MetadataManager::new(self.weak()))
+    }
+
     pub fn poll(&self) {
         let (cmd, data) = self.recv();
 
@@ -227,10 +231,6 @@ impl Session {
         self.0.stream.lock().unwrap().create(handler, self)
     }
 
-    pub fn metadata<T: MetadataTrait>(&self, id: SpotifyId) -> MetadataRef<T> {
-        self.0.metadata.lock().unwrap().get(self, id)
-    }
-
     pub fn cache(&self) -> &Cache {
         self.0.cache.as_ref()
     }