use crypto::digest::Digest; use crypto::sha1::Sha1; use std::io; use std::result::Result; use std::sync::{Mutex, RwLock, Arc, Weak}; use std::str::FromStr; use futures::Future as Future_; use futures::{Stream, BoxFuture, IntoFuture}; use tokio_core::reactor::{Handle, Remote}; use apresolve::apresolve_or_fallback; use authentication::Credentials; use cache::Cache; use component::Lazy; use connection::{self, adaptor}; use audio_key::AudioKeyManager; use channel::ChannelManager; use mercury::MercuryManager; use metadata::MetadataManager; use audio_file::AudioFileManager; #[derive(Clone, Copy, Debug, PartialOrd, Ord, PartialEq, Eq)] pub enum Bitrate { Bitrate96, Bitrate160, Bitrate320, } impl FromStr for Bitrate { type Err = String; fn from_str(s: &str) -> Result { match s { "96" => Ok(Bitrate::Bitrate96), "160" => Ok(Bitrate::Bitrate160), "320" => Ok(Bitrate::Bitrate320), _ => Err(s.into()), } } } pub struct Config { pub user_agent: String, pub name: String, pub device_id: String, pub bitrate: Bitrate, pub onstart: Option, pub onstop: Option, } pub struct SessionData { country: String, canonical_username: String, } pub struct SessionInternal { config: Config, data: RwLock, cache: Box, rx_connection: Mutex), io::Error>>, tx_connection: Mutex)>>, audio_key: Lazy, audio_file: Lazy, channel: Lazy, mercury: Lazy, metadata: Lazy, handle: Remote, } #[derive(Clone)] pub struct Session(pub Arc); #[derive(Clone)] pub struct SessionWeak(pub Weak); pub fn device_id(name: &str) -> String { let mut h = Sha1::new(); h.input_str(&name); h.result_str() } impl Session { pub fn connect(config: Config, credentials: Credentials, cache: Box, handle: Handle) -> Box), Error=io::Error>> { let access_point = apresolve_or_fallback::(&handle); let handle_ = handle.clone(); let connection = access_point.and_then(move |addr| { info!("Connecting to AP \"{}\"", addr); connection::connect::<&str>(&addr, &handle_) }); let device_id = config.device_id.clone(); let authentication = connection.and_then(move |connection| { connection::authenticate(connection, credentials, device_id) }); let result = authentication.map(move |(transport, reusable_credentials)| { info!("Authenticated !"); cache.put_credentials(&reusable_credentials); let (session, task) = Session::create( &handle, transport, config, cache, reusable_credentials.username.clone() ); (session, task) }); Box::new(result) } fn create(handle: &Handle, transport: connection::Transport, config: Config, cache: Box, username: String) -> (Session, BoxFuture<(), io::Error>) { let transport = transport.map(|(cmd, data)| (cmd, data.as_ref().to_owned())); let (tx, rx, task) = adaptor::adapt(transport); let session = Session(Arc::new(SessionInternal { config: config, data: RwLock::new(SessionData { country: String::new(), canonical_username: username, }), rx_connection: Mutex::new(rx), tx_connection: Mutex::new(tx), cache: cache, audio_key: Lazy::new(), audio_file: Lazy::new(), channel: Lazy::new(), mercury: Lazy::new(), metadata: Lazy::new(), handle: handle.remote().clone(), })); (session, task) } pub fn audio_key(&self) -> &AudioKeyManager { self.0.audio_key.get(|| AudioKeyManager::new(self.weak())) } pub fn audio_file(&self) -> &AudioFileManager { self.0.audio_file.get(|| AudioFileManager::new(self.weak())) } pub fn channel(&self) -> &ChannelManager { self.0.channel.get(|| ChannelManager::new(self.weak())) } pub fn mercury(&self) -> &MercuryManager { self.0.mercury.get(|| MercuryManager::new(self.weak())) } pub fn metadata(&self) -> &MetadataManager { self.0.metadata.get(|| MetadataManager::new(self.weak())) } pub fn spawn(&self, f: F) where F: FnOnce(&Handle) -> R + Send + 'static, R: IntoFuture, R::Future: 'static { self.0.handle.spawn(f) } pub fn poll(&self) { let (cmd, data) = self.recv(); match cmd { 0x4 => self.send_packet(0x49, data), 0x4a => (), 0x1b => { self.0.data.write().unwrap().country = String::from_utf8(data).unwrap(); } 0x9 | 0xa => self.channel().dispatch(cmd, data), 0xd | 0xe => self.audio_key().dispatch(cmd, data), 0xb2...0xb6 => self.mercury().dispatch(cmd, data), _ => (), } } pub fn recv(&self) -> (u8, Vec) { self.0.rx_connection.lock().unwrap().recv().unwrap() } pub fn send_packet(&self, cmd: u8, data: Vec) { self.0.tx_connection.lock().unwrap().send((cmd, data)) } pub fn cache(&self) -> &Cache { self.0.cache.as_ref() } pub fn config(&self) -> &Config { &self.0.config } pub fn username(&self) -> String { self.0.data.read().unwrap().canonical_username.clone() } pub fn country(&self) -> String { self.0.data.read().unwrap().country.clone() } pub fn device_id(&self) -> &str { &self.config().device_id } pub fn weak(&self) -> SessionWeak { SessionWeak(Arc::downgrade(&self.0)) } } impl SessionWeak { pub fn upgrade(&self) -> Session { Session(self.0.upgrade().expect("Session died")) } } pub trait PacketHandler { fn handle(&mut self, cmd: u8, data: Vec, session: &Session); }