Merge pull request #191 from Fulkerson/master

Add support for HTTP proxy

.travis.yml

@@ -1,6 +1,6 @@
 language: rust
 rust:
-  - 1.20.0
+  - 1.21.0
   - stable
   - beta
   - nightly

Cargo.toml

@@ -50,7 +50,7 @@ serde_json = "0.9.5"
 tokio-core = "0.1.2"
 tokio-io = "0.1"
 tokio-signal = "0.1.2"
-url = "1.3"
+url = "1.7.0"
 
 [build-dependencies]
 rand            = "0.3.13"

README.md

@@ -26,7 +26,7 @@ If you wish to learn more about how librespot works overall, the best way is to
 If you run into a bug when using librespot, please search the existing issues before opening a new one. Chances are, we've encountered it before, and have provided a resolution. If not, please open a new one, and where possible, include the backtrace librespot generates on crashing, along with anything we can use to reproduce the issue, eg. the Spotify URI of the song that caused the crash.
 
 # Building
-Rust 1.20.0 or later is required to build librespot.
+Rust 1.21.0 or later is required to build librespot.
 
 **If you are building librespot on macOS, the homebrew provided rust may fail due to the way in which homebrew installs rust. In this case, uninstall the homebrew version of rust and use [rustup](https://www.rustup.rs/), and librespot should then build. This should have been fixed in more recent versions of Homebrew, but we're leaving this notice here as a warning.**
 

audio/src/fetch.rs

@@ -1,8 +1,8 @@
 use bit_set::BitSet;
 use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
-use futures::{Async, Future, Poll};
 use futures::Stream;
 use futures::sync::{mpsc, oneshot};
+use futures::{Async, Future, Poll};
 use std::cmp::min;
 use std::fs;
 use std::io::{self, Read, Seek, SeekFrom, Write};
@@ -288,20 +288,12 @@ impl Future for AudioFileFetch {
                 Ok(Async::Ready(Some(data))) => {
                     progress = true;
 
-                    self.output
-                        .as_mut()
-                        .unwrap()
-                        .write_all(data.as_ref())
-                        .unwrap();
+                    self.output.as_mut().unwrap().write_all(data.as_ref()).unwrap();
                 }
                 Ok(Async::Ready(None)) => {
                     progress = true;
 
-                    trace!(
-                        "chunk {} / {} complete",
-                        self.index,
-                        self.shared.chunk_count
-                    );
+                    trace!("chunk {} / {} complete", self.index, self.shared.chunk_count);
 
                     let full = {
                         let mut bitmap = self.shared.bitmap.lock().unwrap();

connect/src/discovery.rs

@@ -2,10 +2,10 @@ use base64;
 use crypto;
 use crypto::digest::Digest;
 use crypto::mac::Mac;
-use futures::{Future, Poll, Stream};
 use futures::sync::mpsc;
-use hyper::{self, Get, Post, StatusCode};
+use futures::{Future, Poll, Stream};
 use hyper::server::{Http, Request, Response, Service};
+use hyper::{self, Get, Post, StatusCode};
 
 #[cfg(feature = "with-dns-sd")]
 use dns_sd::DNSService;

connect/src/spirc.rs

@@ -1,6 +1,6 @@
-use futures::{Async, Future, Poll, Sink, Stream};
 use futures::future;
 use futures::sync::{mpsc, oneshot};
+use futures::{Async, Future, Poll, Sink, Stream};
 use protobuf::{self, Message};
 
 use core::config::ConnectConfig;
@@ -442,8 +442,7 @@ impl SpircTask {
                 self.update_tracks(&frame);
 
                 if self.state.get_track().len() > 0 {
-                    self.state
-                        .set_position_ms(frame.get_state().get_position_ms());
+                    self.state.set_position_ms(frame.get_state().get_position_ms());
                     self.state.set_position_measured_at(now_ms() as u64);
 
                     let play = frame.get_state().get_status() == PlayStatus::kPlayStatusPlay;
@@ -530,10 +529,8 @@ impl SpircTask {
 
             MessageType::kMessageTypeVolume => {
                 self.device.set_volume(frame.get_volume());
-                self.mixer.set_volume(volume_to_mixer(
-                    frame.get_volume() as u16,
-                    self.linear_volume,
-                ));
+                self.mixer
+                    .set_volume(volume_to_mixer(frame.get_volume() as u16, self.linear_volume));
                 self.notify(None);
             }
 

core/Cargo.toml

@@ -11,10 +11,12 @@ path = "../protocol"
 base64 = "0.5.0"
 byteorder = "1.0"
 bytes = "0.4"
-error-chain = { version = "0.9.0", default_features = false }
-extprim = "1.5.1" 
+error-chain = { version = "0.11.0", default_features = false }
+extprim = "1.5.1"
 futures = "0.1.8"
+httparse = "1.2.4"
 hyper = "0.11.2"
+hyper-proxy = { version = "0.4.1", default_features = false }
 lazy_static = "0.2.0"
 log = "0.3.5"
 num-bigint = "0.1.35"
@@ -30,6 +32,7 @@ serde_json = "0.9.5"
 shannon = "0.2.0"
 tokio-core = "0.1.2"
 tokio-io = "0.1"
+url = "1.7.0"
 uuid = { version = "0.4", features = ["v4"] }
 
 [build-dependencies]

core/src/apresolve.rs

@@ -1,11 +1,14 @@
-const AP_FALLBACK: &'static str = "ap.spotify.com:80";
+const AP_FALLBACK: &'static str = "ap.spotify.com:443";
 const APRESOLVE_ENDPOINT: &'static str = "http://apresolve.spotify.com/";
 
 use futures::{Future, Stream};
-use hyper::{self, Client, Uri};
+use hyper::client::HttpConnector;
+use hyper::{self, Client, Method, Request, Uri};
+use hyper_proxy::{Intercept, Proxy, ProxyConnector};
 use serde_json;
 use std::str::FromStr;
 use tokio_core::reactor::Handle;
+use url::Url;
 
 error_chain!{}
 
@@ -14,11 +17,29 @@ pub struct APResolveData {
     ap_list: Vec<String>,
 }
 
-fn apresolve(handle: &Handle) -> Box<Future<Item = String, Error = Error>> {
+fn apresolve(handle: &Handle, proxy: &Option<Url>) -> Box<Future<Item = String, Error = Error>> {
     let url = Uri::from_str(APRESOLVE_ENDPOINT).expect("invalid AP resolve URL");
+    let use_proxy = proxy.is_some();
 
-    let client = Client::new(handle);
-    let response = client.get(url);
+    let mut req = Request::new(Method::Get, url.clone());
+    let response = match *proxy {
+        Some(ref val) => {
+            let proxy_url = Uri::from_str(val.as_str()).expect("invalid http proxy");
+            let proxy = Proxy::new(Intercept::All, proxy_url);
+            let connector = HttpConnector::new(4, handle);
+            let proxy_connector = ProxyConnector::from_proxy_unsecured(connector, proxy);
+            if let Some(headers) = proxy_connector.http_headers(&url) {
+                req.headers_mut().extend(headers.iter());
+                req.set_proxy(true);
+            }
+            let client = Client::configure().connector(proxy_connector).build(handle);
+            client.request(req)
+        }
+        _ => {
+            let client = Client::new(handle);
+            client.request(req)
+        }
+    };
 
     let body = response.and_then(|response| {
         response.body().fold(Vec::new(), |mut acc, chunk| {
@@ -32,19 +53,33 @@ fn apresolve(handle: &Handle) -> Box<Future<Item = String, Error = Error>> {
     let data =
         body.and_then(|body| serde_json::from_str::<APResolveData>(&body).chain_err(|| "invalid JSON"));
 
-    let ap = data.and_then(|data| {
-        let ap = data.ap_list.first().ok_or("empty AP List")?;
+    let ap = data.and_then(move |data| {
+        let mut aps = data.ap_list.iter().filter(|ap| {
+            if use_proxy {
+                // It is unlikely that the proxy will accept CONNECT on anything other than 443.
+                Uri::from_str(ap)
+                    .ok()
+                    .map_or(false, |uri| uri.port().map_or(false, |port| port == 443))
+            } else {
+                true
+            }
+        });
+
+        let ap = aps.next().ok_or("empty AP List")?;
         Ok(ap.clone())
     });
 
     Box::new(ap)
 }
 
-pub(crate) fn apresolve_or_fallback<E>(handle: &Handle) -> Box<Future<Item = String, Error = E>>
+pub(crate) fn apresolve_or_fallback<E>(
+    handle: &Handle,
+    proxy: &Option<Url>,
+) -> Box<Future<Item = String, Error = E>>
 where
     E: 'static,
 {
-    let ap = apresolve(handle).or_else(|e| {
+    let ap = apresolve(handle, proxy).or_else(|e| {
         warn!("Failed to resolve Access Point: {}", e.description());
         warn!("Using fallback \"{}\"", AP_FALLBACK);
         Ok(AP_FALLBACK.into())

core/src/audio_key.rs

@@ -1,7 +1,7 @@
 use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
 use bytes::Bytes;
-use futures::{Async, Future, Poll};
 use futures::sync::oneshot;
+use futures::{Async, Future, Poll};
 use std::collections::HashMap;
 use std::io::Write;
 
@@ -35,11 +35,7 @@ impl AudioKeyManager {
                     let _ = sender.send(Ok(AudioKey(key)));
                 }
                 0xe => {
-                    warn!(
-                        "error audio key {:x} {:x}",
-                        data.as_ref()[0],
-                        data.as_ref()[1]
-                    );
+                    warn!("error audio key {:x} {:x}", data.as_ref()[0], data.as_ref()[1]);
                     let _ = sender.send(Err(AudioKeyError));
                 }
                 _ => (),

core/src/authentication.rs

@@ -88,11 +88,8 @@ impl Credentials {
         let blob = {
             // Anyone know what this block mode is ?
             let mut data = vec![0u8; encrypted_blob.len()];
-            let mut cipher = aes::ecb_decryptor(
-                aes::KeySize::KeySize192,
-                &key,
-                crypto::blockmodes::NoPadding,
-            );
+            let mut cipher =
+                aes::ecb_decryptor(aes::KeySize::KeySize192, &key, crypto::blockmodes::NoPadding);
             cipher
                 .decrypt(
                     &mut crypto::buffer::RefReadBuffer::new(&encrypted_blob),
@@ -193,10 +190,9 @@ pub fn get_credentials<F: FnOnce(&String) -> String>(
             Some(credentials.clone())
         }
 
-        (Some(username), None, _) => Some(Credentials::with_password(
-            username.clone(),
-            prompt(&username),
-        )),
+        (Some(username), None, _) => {
+            Some(Credentials::with_password(username.clone(), prompt(&username)))
+        }
 
         (None, _, Some(credentials)) => Some(credentials),
 

core/src/channel.rs

@@ -1,7 +1,7 @@
 use byteorder::{BigEndian, ByteOrder};
 use bytes::Bytes;
-use futures::{Async, Poll, Stream};
 use futures::sync::{mpsc, BiLock};
+use futures::{Async, Poll, Stream};
 use std::collections::HashMap;
 
 use util::SeqGenerator;

core/src/config.rs

@@ -1,5 +1,6 @@
 use std::fmt;
 use std::str::FromStr;
+use url::Url;
 use uuid::Uuid;
 
 use version;
@@ -8,6 +9,7 @@ use version;
 pub struct SessionConfig {
     pub user_agent: String,
     pub device_id: String,
+    pub proxy: Option<Url>,
 }
 
 impl Default for SessionConfig {
@@ -16,6 +18,7 @@ impl Default for SessionConfig {
         SessionConfig {
             user_agent: version::version_string(),
             device_id: device_id,
+            proxy: None,
         }
     }
 }

core/src/connection/codec.rs

@@ -88,8 +88,7 @@ impl Decoder for APCodec {
 
                 let mut payload = buf.split_to(size + MAC_SIZE);
 
-                self.decode_cipher
-                    .decrypt(&mut payload.get_mut(..size).unwrap());
+                self.decode_cipher.decrypt(&mut payload.get_mut(..size).unwrap());
                 let mac = payload.split_off(size);
                 self.decode_cipher.check_mac(mac.as_ref())?;
 

core/src/connection/handshake.rs

@@ -7,9 +7,9 @@ use protobuf::{self, Message, MessageStatic};
 use rand::thread_rng;
 use std::io::{self, Read};
 use std::marker::PhantomData;
-use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_io::codec::Framed;
 use tokio_io::io::{read_exact, write_all, ReadExact, Window, WriteAll};
+use tokio_io::{AsyncRead, AsyncWrite};
 
 use super::codec::APCodec;
 use diffie_hellman::DHLocalKeys;
@@ -93,10 +93,7 @@ fn client_hello<T: AsyncWrite>(connection: T, gc: Vec<u8>) -> WriteAll<T, Vec<u8
     packet
         .mut_cryptosuites_supported()
         .push(protocol::keyexchange::Cryptosuite::CRYPTO_SUITE_SHANNON);
-    packet
-        .mut_login_crypto_hello()
-        .mut_diffie_hellman()
-        .set_gc(gc);
+    packet.mut_login_crypto_hello().mut_diffie_hellman().set_gc(gc);
     packet
         .mut_login_crypto_hello()
         .mut_diffie_hellman()

core/src/connection/mod.rs

@@ -11,21 +11,37 @@ use std::net::ToSocketAddrs;
 use tokio_core::net::TcpStream;
 use tokio_core::reactor::Handle;
 use tokio_io::codec::Framed;
+use url::Url;
 
 use authentication::Credentials;
 use version;
 
+use proxytunnel;
+
 pub type Transport = Framed<TcpStream, APCodec>;
 
-pub fn connect<A: ToSocketAddrs>(
-    addr: A,
+pub fn connect(
+    addr: String,
     handle: &Handle,
+    proxy: &Option<Url>,
 ) -> Box<Future<Item = Transport, Error = io::Error>> {
-    let addr = addr.to_socket_addrs().unwrap().next().unwrap();
-    let socket = TcpStream::connect(&addr, handle);
-    let connection = socket.and_then(|socket| handshake(socket));
+    let (addr, connect_url) = match *proxy {
+        Some(ref url) => {
+            info!("Using proxy \"{}\"", url);
+            (url.to_socket_addrs().unwrap().next().unwrap(), Some(addr))
+        }
+        None => (addr.to_socket_addrs().unwrap().next().unwrap(), None),
+    };
 
-    Box::new(connection)
+    let socket = TcpStream::connect(&addr, handle);
+    if let Some(connect_url) = connect_url {
+        let connection = socket
+            .and_then(move |socket| proxytunnel::connect(socket, &connect_url).and_then(handshake));
+        Box::new(connection)
+    } else {
+        let connection = socket.and_then(handshake);
+        Box::new(connection)
+    }
 }
 
 pub fn authenticate(
@@ -37,26 +53,18 @@ pub fn authenticate(
     use protocol::keyexchange::APLoginFailed;
 
     let mut packet = ClientResponseEncrypted::new();
-    packet
-        .mut_login_credentials()
-        .set_username(credentials.username);
-    packet
-        .mut_login_credentials()
-        .set_typ(credentials.auth_type);
+    packet.mut_login_credentials().set_username(credentials.username);
+    packet.mut_login_credentials().set_typ(credentials.auth_type);
     packet
         .mut_login_credentials()
         .set_auth_data(credentials.auth_data);
-    packet
-        .mut_system_info()
-        .set_cpu_family(CpuFamily::CPU_UNKNOWN);
+    packet.mut_system_info().set_cpu_family(CpuFamily::CPU_UNKNOWN);
     packet.mut_system_info().set_os(Os::OS_UNKNOWN);
-    packet
-        .mut_system_info()
-        .set_system_information_string(format!(
-            "librespot_{}_{}",
-            version::short_sha(),
-            version::build_id()
-        ));
+    packet.mut_system_info().set_system_information_string(format!(
+        "librespot_{}_{}",
+        version::short_sha(),
+        version::build_id()
+    ));
     packet.mut_system_info().set_device_id(device_id);
     packet.set_version_string(version::version_string());
 

core/src/diffie_hellman.rs

@@ -7,17 +7,13 @@ use util;
 lazy_static! {
     pub static ref DH_GENERATOR: BigUint = BigUint::from_u64(0x2).unwrap();
     pub static ref DH_PRIME: BigUint = BigUint::from_bytes_be(&[
-        0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xc9,
-        0x0f, 0xda, 0xa2, 0x21, 0x68, 0xc2, 0x34, 0xc4, 0xc6,
-        0x62, 0x8b, 0x80, 0xdc, 0x1c, 0xd1, 0x29, 0x02, 0x4e,
-        0x08, 0x8a, 0x67, 0xcc, 0x74, 0x02, 0x0b, 0xbe, 0xa6,
-        0x3b, 0x13, 0x9b, 0x22, 0x51, 0x4a, 0x08, 0x79, 0x8e,
-        0x34, 0x04, 0xdd, 0xef, 0x95, 0x19, 0xb3, 0xcd, 0x3a,
-        0x43, 0x1b, 0x30, 0x2b, 0x0a, 0x6d, 0xf2, 0x5f, 0x14,
-        0x37, 0x4f, 0xe1, 0x35, 0x6d, 0x6d, 0x51, 0xc2, 0x45,
-        0xe4, 0x85, 0xb5, 0x76, 0x62, 0x5e, 0x7e, 0xc6, 0xf4,
-        0x4c, 0x42, 0xe9, 0xa6, 0x3a, 0x36, 0x20, 0xff, 0xff,
-        0xff, 0xff, 0xff, 0xff, 0xff, 0xff ]);
+        0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xc9, 0x0f, 0xda, 0xa2, 0x21, 0x68, 0xc2, 0x34,
+        0xc4, 0xc6, 0x62, 0x8b, 0x80, 0xdc, 0x1c, 0xd1, 0x29, 0x02, 0x4e, 0x08, 0x8a, 0x67, 0xcc, 0x74,
+        0x02, 0x0b, 0xbe, 0xa6, 0x3b, 0x13, 0x9b, 0x22, 0x51, 0x4a, 0x08, 0x79, 0x8e, 0x34, 0x04, 0xdd,
+        0xef, 0x95, 0x19, 0xb3, 0xcd, 0x3a, 0x43, 0x1b, 0x30, 0x2b, 0x0a, 0x6d, 0xf2, 0x5f, 0x14, 0x37,
+        0x4f, 0xe1, 0x35, 0x6d, 0x6d, 0x51, 0xc2, 0x45, 0xe4, 0x85, 0xb5, 0x76, 0x62, 0x5e, 0x7e, 0xc6,
+        0xf4, 0x4c, 0x42, 0xe9, 0xa6, 0x3a, 0x36, 0x20, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+    ]);
 }
 
 pub struct DHLocalKeys {
@@ -43,11 +39,7 @@ impl DHLocalKeys {
     }
 
     pub fn shared_secret(&self, remote_key: &[u8]) -> Vec<u8> {
-        let shared_key = util::powm(
-            &BigUint::from_bytes_be(remote_key),
-            &self.private_key,
-            &DH_PRIME,
-        );
+        let shared_key = util::powm(&BigUint::from_bytes_be(remote_key), &self.private_key, &DH_PRIME);
         shared_key.to_bytes_be()
     }
 }

core/src/lib.rs

@@ -16,7 +16,9 @@ extern crate byteorder;
 extern crate bytes;
 extern crate crypto;
 extern crate extprim;
+extern crate httparse;
 extern crate hyper;
+extern crate hyper_proxy;
 extern crate num_bigint;
 extern crate num_integer;
 extern crate num_traits;
@@ -28,6 +30,7 @@ extern crate serde_json;
 extern crate shannon;
 extern crate tokio_core;
 extern crate tokio_io;
+extern crate url;
 extern crate uuid;
 
 extern crate librespot_protocol as protocol;
@@ -44,6 +47,7 @@ mod connection;
 pub mod diffie_hellman;
 pub mod keymaster;
 pub mod mercury;
+mod proxytunnel;
 pub mod session;
 pub mod spotify_id;
 pub mod util;

core/src/mercury/mod.rs

@@ -1,7 +1,7 @@
 use byteorder::{BigEndian, ByteOrder};
 use bytes::Bytes;
-use futures::{Async, Future, Poll};
 use futures::sync::{mpsc, oneshot};
+use futures::{Async, Future, Poll};
 use protobuf;
 use protocol;
 use std::collections::HashMap;

core/src/proxytunnel.rs

@@ -0,0 +1,108 @@
+use std::error::Error;
+use std::io;
+use std::str::FromStr;
+
+use futures::{Async, Future, Poll};
+use httparse;
+use hyper::Uri;
+use tokio_io::io::{read, write_all, Read, Window, WriteAll};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+pub struct ProxyTunnel<T> {
+    state: ProxyState<T>,
+}
+
+enum ProxyState<T> {
+    ProxyConnect(WriteAll<T, Vec<u8>>),
+    ProxyResponse(Read<T, Window<Vec<u8>>>),
+}
+
+pub fn connect<T: AsyncRead + AsyncWrite>(connection: T, connect_url: &str) -> ProxyTunnel<T> {
+    let proxy = proxy_connect(connection, connect_url);
+    ProxyTunnel {
+        state: ProxyState::ProxyConnect(proxy),
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> Future for ProxyTunnel<T> {
+    type Item = T;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, io::Error> {
+        use self::ProxyState::*;
+        loop {
+            self.state = match self.state {
+                ProxyConnect(ref mut write) => {
+                    let (connection, mut accumulator) = try_ready!(write.poll());
+
+                    let capacity = accumulator.capacity();
+                    accumulator.resize(capacity, 0);
+                    let window = Window::new(accumulator);
+
+                    let read = read(connection, window);
+                    ProxyResponse(read)
+                }
+
+                ProxyResponse(ref mut read_f) => {
+                    let (connection, mut window, bytes_read) = try_ready!(read_f.poll());
+
+                    if bytes_read == 0 {
+                        return Err(io::Error::new(io::ErrorKind::Other, "Early EOF from proxy"));
+                    }
+
+                    let data_end = window.start() + bytes_read;
+
+                    let buf = window.get_ref()[0..data_end].to_vec();
+                    let mut headers = [httparse::EMPTY_HEADER; 16];
+                    let mut response = httparse::Response::new(&mut headers);
+                    let status = match response.parse(&buf) {
+                        Ok(status) => status,
+                        Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err.description())),
+                    };
+
+                    if status.is_complete() {
+                        if let Some(code) = response.code {
+                            if code == 200 {
+                                // Proxy says all is well
+                                return Ok(Async::Ready(connection));
+                            } else {
+                                let reason = response.reason.unwrap_or("no reason");
+                                let msg = format!("Proxy responded with {}: {}", code, reason);
+
+                                return Err(io::Error::new(io::ErrorKind::Other, msg));
+                            }
+                        } else {
+                            return Err(io::Error::new(
+                                io::ErrorKind::Other,
+                                "Malformed response from proxy",
+                            ));
+                        }
+                    } else {
+                        if data_end >= window.end() {
+                            // Allocate some more buffer space
+                            let newsize = data_end + 100;
+                            window.get_mut().resize(newsize, 0);
+                            window.set_end(newsize);
+                        }
+                        // We did not get a full header
+                        window.set_start(data_end);
+                        let read = read(connection, window);
+                        ProxyResponse(read)
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn proxy_connect<T: AsyncWrite>(connection: T, connect_url: &str) -> WriteAll<T, Vec<u8>> {
+    let uri = Uri::from_str(connect_url).unwrap();
+    let buffer = format!(
+        "CONNECT {0}:{1} HTTP/1.1\r\n\
+         \r\n",
+        uri.host().expect(&format!("No host in {}", uri)),
+        uri.port().expect(&format!("No port in {}", uri))
+    ).into_bytes();
+
+    write_all(connection, buffer)
+}

core/src/session.rs

@@ -1,9 +1,9 @@
 use bytes::Bytes;
-use futures::{Async, Future, IntoFuture, Poll, Stream};
 use futures::sync::mpsc;
+use futures::{Async, Future, IntoFuture, Poll, Stream};
 use std::io;
-use std::sync::{Arc, RwLock, Weak};
 use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
+use std::sync::{Arc, RwLock, Weak};
 use tokio_core::reactor::{Handle, Remote};
 
 use apresolve::apresolve_or_fallback;
@@ -50,12 +50,13 @@ impl Session {
         cache: Option<Cache>,
         handle: Handle,
     ) -> Box<Future<Item = Session, Error = io::Error>> {
-        let access_point = apresolve_or_fallback::<io::Error>(&handle);
+        let access_point = apresolve_or_fallback::<io::Error>(&handle, &config.proxy);
 
         let handle_ = handle.clone();
+        let proxy = config.proxy.clone();
         let connection = access_point.and_then(move |addr| {
             info!("Connecting to AP \"{}\"", addr);
-            connection::connect::<&str>(&addr, &handle_)
+            connection::connect(addr, &handle_, &proxy)
         });
 
         let device_id = config.device_id.clone();
@@ -124,11 +125,7 @@ impl Session {
             .map(|_| ());
         let receiver_task = DispatchTask(stream, session.weak());
 
-        let task = Box::new(
-            (receiver_task, sender_task)
-                .into_future()
-                .map(|((), ())| ()),
-        );
+        let task = Box::new((receiver_task, sender_task).into_future().map(|((), ())| ()));
 
         (session, task)
     }

examples/play.rs

@@ -36,9 +36,7 @@ fn main() {
     let session = core.run(Session::connect(session_config, credentials, None, handle))
         .unwrap();
 
-    let player = Player::new(player_config, session.clone(), None, move || {
-        (backend)(None)
-    });
+    let player = Player::new(player_config, session.clone(), None, move || (backend)(None));
 
     println!("Playing...");
     core.run(player.load(track, true, 0)).unwrap();

playback/src/audio_backend/jackaudio.rs

@@ -44,12 +44,8 @@ impl Open for JackSink {
 
         let client_name = client_name.unwrap_or("librespot".to_string());
         let (client, _status) = Client::new(&client_name[..], client_options::NO_START_SERVER).unwrap();
-        let ch_r = client
-            .register_port("out_0", AudioOutSpec::default())
-            .unwrap();
-        let ch_l = client
-            .register_port("out_1", AudioOutSpec::default())
-            .unwrap();
+        let ch_r = client.register_port("out_0", AudioOutSpec::default()).unwrap();
+        let ch_l = client.register_port("out_1", AudioOutSpec::default()).unwrap();
         // buffer for samples from librespot (~10ms)
         let (tx, rx) = sync_channel(2 * 1024 * 4);
         let jack_data = JackData {

playback/src/audio_backend/pipe.rs

@@ -28,10 +28,7 @@ impl Sink for StdoutSink {
 
     fn write(&mut self, data: &[i16]) -> io::Result<()> {
         let data: &[u8] = unsafe {
-            slice::from_raw_parts(
-                data.as_ptr() as *const u8,
-                data.len() * mem::size_of::<i16>(),
-            )
+            slice::from_raw_parts(data.as_ptr() as *const u8, data.len() * mem::size_of::<i16>())
         };
 
         self.0.write_all(data)?;

playback/src/player.rs

@@ -1,7 +1,7 @@
 use byteorder::{LittleEndian, ReadBytesExt};
 use futures;
-use futures::{future, Future};
 use futures::sync::oneshot;
+use futures::{future, Future};
 use std;
 use std::borrow::Cow;
 use std::io::{Read, Result, Seek, SeekFrom};
@@ -93,10 +93,8 @@ impl NormalisationData {
     }
 
     fn get_factor(config: &PlayerConfig, data: NormalisationData) -> f32 {
-        let mut normalisation_factor = f32::powf(
-            10.0,
-            (data.track_gain_db + config.normalisation_pregain) / 20.0,
-        );
+        let mut normalisation_factor =
+            f32::powf(10.0, (data.track_gain_db + config.normalisation_pregain) / 20.0);
 
         if normalisation_factor * data.track_peak > 1.0 {
             warn!("Reducing normalisation factor to prevent clipping. Please add negative pregain to avoid.");
@@ -231,12 +229,7 @@ impl PlayerState {
         use self::PlayerState::*;
         match *self {
             Stopped | EndOfTrack { .. } => None,
-            Paused {
-                ref mut decoder, ..
-            }
-            | Playing {
-                ref mut decoder, ..
-            } => Some(decoder),
+            Paused { ref mut decoder, .. } | Playing { ref mut decoder, .. } => Some(decoder),
             Invalid => panic!("invalid state"),
         }
     }
@@ -525,10 +518,7 @@ impl PlayerInternal {
                 .map(|alt_id| Track::get(&self.session, *alt_id));
             let alternatives = future::join_all(alternatives).wait().unwrap();
 
-            alternatives
-                .into_iter()
-                .find(|alt| alt.available)
-                .map(Cow::Owned)
+            alternatives.into_iter().find(|alt| alt.available).map(Cow::Owned)
         }
     }
 
@@ -558,10 +548,7 @@ impl PlayerInternal {
         let file_id = match track.files.get(&format) {
             Some(&file_id) => file_id,
             None => {
-                warn!(
-                    "Track \"{}\" is not available in format {:?}",
-                    track.name, format
-                );
+                warn!("Track \"{}\" is not available in format {:?}", track.name, format);
                 return None;
             }
         };

protocol/build.rs

@@ -7,10 +7,7 @@ fn main() {
     for &(path, expected_checksum) in files::FILES {
         let actual = cksum_file(path).unwrap();
         if expected_checksum != actual {
-            panic!(
-                "Checksum for {:?} does not match. Try running build.sh",
-                path
-            );
+            panic!("Checksum for {:?} does not match. Try running build.sh", path);
         }
     }
 }

src/main.rs

@@ -9,12 +9,13 @@ extern crate rpassword;
 extern crate tokio_core;
 extern crate tokio_io;
 extern crate tokio_signal;
+extern crate url;
 
 use crypto::digest::Digest;
 use crypto::sha1::Sha1;
 use env_logger::LogBuilder;
-use futures::{Async, Future, Poll, Stream};
 use futures::sync::mpsc::UnboundedReceiver;
+use futures::{Async, Future, Poll, Stream};
 use std::env;
 use std::io::{self, stderr, Write};
 use std::mem;
@@ -23,6 +24,7 @@ use std::process::exit;
 use std::str::FromStr;
 use tokio_core::reactor::{Core, Handle};
 use tokio_io::IoStream;
+use url::Url;
 
 use librespot::core::authentication::{get_credentials, Credentials};
 use librespot::core::cache::Cache;
@@ -108,11 +110,7 @@ fn setup(args: &[String]) -> Setup {
         "cache",
         "Path to a directory where files will be cached.",
         "CACHE",
-    ).optflag(
-            "",
-            "disable-audio-cache",
-            "Disable caching of the audio data.",
-        )
+    ).optflag("", "disable-audio-cache", "Disable caching of the audio data.")
         .reqopt("n", "name", "Device name", "NAME")
         .optopt("", "device-type", "Displayed device type", "DEVICE_TYPE")
         .optopt(
@@ -130,6 +128,7 @@ fn setup(args: &[String]) -> Setup {
         .optflag("v", "verbose", "Enable verbose output")
         .optopt("u", "username", "Username to sign in with", "USERNAME")
         .optopt("p", "password", "Password", "PASSWORD")
+        .optopt("", "proxy", "HTTP proxy to use when connecting", "PROXY")
         .optflag("", "disable-discovery", "Disable discovery mode")
         .optopt(
             "",
@@ -176,12 +175,7 @@ fn setup(args: &[String]) -> Setup {
     let matches = match opts.parse(&args[1..]) {
         Ok(m) => m,
         Err(f) => {
-            writeln!(
-                stderr(),
-                "error: {}\n{}",
-                f.to_string(),
-                usage(&args[0], &opts)
-            ).unwrap();
+            writeln!(stderr(), "error: {}\n{}", f.to_string(), usage(&args[0], &opts)).unwrap();
             exit(1);
         }
     };
@@ -256,6 +250,23 @@ fn setup(args: &[String]) -> Setup {
         SessionConfig {
             user_agent: version::version_string(),
             device_id: device_id,
+            proxy: matches.opt_str("proxy").or(std::env::var("http_proxy").ok()).map(
+                |s| {
+                    match Url::parse(&s) {
+                Ok(url) => {
+                    if url.host().is_none() || url.port().is_none() {
+                        panic!("Invalid proxy url, only urls on the format \"http://host:port\" are allowed");
+                    }
+
+                    if url.scheme() != "http" {
+                        panic!("Only unsecure http:// proxies are supported");
+                    }
+                    url
+                },
+                Err(err) => panic!("Invalid proxy url: {}, only urls on the format \"http://host:port\" are allowed", err)
+            }
+                },
+            ),
         }
     };