Merge pull request #9 from brain0/fix_deprecated

Update dependencies and fix the use of deprecated functionality
This commit is contained in:
Colm 2018-01-26 21:17:53 +00:00 committed by GitHub
commit 0105c53af4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
23 changed files with 462 additions and 432 deletions

View file

@ -1,6 +1,6 @@
language: rust language: rust
rust: rust:
- 1.17.0 - 1.18.0
- stable - stable
- beta - beta
- nightly - nightly
@ -24,13 +24,11 @@ before_script:
script: script:
- cargo build --no-default-features - cargo build --no-default-features
- cargo build --no-default-features --features "with-tremor" - cargo build --no-default-features --features "with-tremor"
- cargo build --no-default-features --features "with-lewton";
- cargo build --no-default-features --features "portaudio-backend" - cargo build --no-default-features --features "portaudio-backend"
- cargo build --no-default-features --features "pulseaudio-backend" - cargo build --no-default-features --features "pulseaudio-backend"
- cargo build --no-default-features --features "alsa-backend" - cargo build --no-default-features --features "alsa-backend"
- cargo build --no-default-features --target armv7-unknown-linux-gnueabihf - cargo build --no-default-features --target armv7-unknown-linux-gnueabihf
- if [[ $TRAVIS_RUST_VERSION != *"1.17.0"* ]]; then
cargo build --no-default-features --features "with-lewton";
fi
notifications: notifications:
email: false email: false

584
Cargo.lock generated

File diff suppressed because it is too large Load diff

View file

@ -46,6 +46,7 @@ serde = "0.9.6"
serde_derive = "0.9.6" serde_derive = "0.9.6"
serde_json = "0.9.5" serde_json = "0.9.5"
tokio-core = "0.1.2" tokio-core = "0.1.2"
tokio-io = "0.1"
tokio-signal = "0.1.2" tokio-signal = "0.1.2"
url = "1.3" url = "1.3"

View file

@ -21,7 +21,7 @@ I've done noting more than make this pretty so big thanks to:
[brain0](https://github.com/brain0/) for [making pluseaudio more robust against audio failures](https://github.com/ComlOnline/librespot/pull/6) [brain0](https://github.com/brain0/) for [making pluseaudio more robust against audio failures](https://github.com/ComlOnline/librespot/pull/6)
## Building ## Building
Rust 1.17.0 or later is required to build librespot. Rust 1.18.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.** **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.**

View file

@ -340,7 +340,7 @@ impl Seek for AudioFileStreaming {
// Notify the fetch thread to get the correct block // Notify the fetch thread to get the correct block
// This can fail if fetch thread has completed, in which case the // This can fail if fetch thread has completed, in which case the
// block is ready. Just ignore the error. // block is ready. Just ignore the error.
let _ = self.seek.send(self.position); let _ = self.seek.unbounded_send(self.position);
Ok(self.position) Ok(self.position)
} }
} }

View file

@ -10,6 +10,7 @@ path = "../protocol"
[dependencies] [dependencies]
base64 = "0.5.0" base64 = "0.5.0"
byteorder = "1.0" byteorder = "1.0"
bytes = "0.4"
error-chain = { version = "0.9.0", default_features = false } error-chain = { version = "0.9.0", default_features = false }
futures = "0.1.8" futures = "0.1.8"
hyper = "0.11.2" hyper = "0.11.2"
@ -27,6 +28,7 @@ serde_derive = "0.9.6"
serde_json = "0.9.5" serde_json = "0.9.5"
shannon = "0.2.0" shannon = "0.2.0"
tokio-core = "0.1.2" tokio-core = "0.1.2"
tokio-io = "0.1"
uuid = { version = "0.4", features = ["v4"] } uuid = { version = "0.4", features = ["v4"] }
[build-dependencies] [build-dependencies]

View file

@ -39,5 +39,7 @@ pub fn build_id() -> &'static str {{
protobuf_macros::expand("src/lib.in.rs", &out.join("lib.rs")).unwrap(); protobuf_macros::expand("src/lib.in.rs", &out.join("lib.rs")).unwrap();
println!("cargo:rerun-if-changed=src/lib.in.rs"); println!("cargo:rerun-if-changed=src/lib.in.rs");
println!("cargo:rerun-if-changed=src/connection"); println!("cargo:rerun-if-changed=src/connection/mod.rs");
println!("cargo:rerun-if-changed=src/connection/codec.rs");
println!("cargo:rerun-if-changed=src/connection/handshake.rs");
} }

View file

@ -1,9 +1,9 @@
use byteorder::{BigEndian, ByteOrder, WriteBytesExt}; use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
use bytes::Bytes;
use futures::sync::oneshot; use futures::sync::oneshot;
use futures::{Async, Future, Poll}; use futures::{Async, Future, Poll};
use std::collections::HashMap; use std::collections::HashMap;
use std::io::Write; use std::io::Write;
use tokio_core::io::EasyBuf;
use util::SeqGenerator; use util::SeqGenerator;
use util::{SpotifyId, FileId}; use util::{SpotifyId, FileId};
@ -22,8 +22,8 @@ component! {
} }
impl AudioKeyManager { impl AudioKeyManager {
pub fn dispatch(&self, cmd: u8, mut data: EasyBuf) { pub fn dispatch(&self, cmd: u8, mut data: Bytes) {
let seq = BigEndian::read_u32(data.drain_to(4).as_ref()); let seq = BigEndian::read_u32(data.split_to(4).as_ref());
let sender = self.lock(|inner| inner.pending.remove(&seq)); let sender = self.lock(|inner| inner.pending.remove(&seq));
@ -32,11 +32,11 @@ impl AudioKeyManager {
0xd => { 0xd => {
let mut key = [0u8; 16]; let mut key = [0u8; 16];
key.copy_from_slice(data.as_ref()); key.copy_from_slice(data.as_ref());
sender.complete(Ok(AudioKey(key))); let _ = sender.send(Ok(AudioKey(key)));
} }
0xe => { 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]);
sender.complete(Err(AudioKeyError)); let _ = sender.send(Err(AudioKeyError));
} }
_ => (), _ => (),
} }

View file

@ -1,15 +1,15 @@
use byteorder::{BigEndian, ByteOrder}; use byteorder::{BigEndian, ByteOrder};
use bytes::Bytes;
use futures::sync::{BiLock, mpsc}; use futures::sync::{BiLock, mpsc};
use futures::{Poll, Async, Stream}; use futures::{Poll, Async, Stream};
use std::collections::HashMap; use std::collections::HashMap;
use tokio_core::io::EasyBuf;
use util::SeqGenerator; use util::SeqGenerator;
component! { component! {
ChannelManager : ChannelManagerInner { ChannelManager : ChannelManagerInner {
sequence: SeqGenerator<u16> = SeqGenerator::new(0), sequence: SeqGenerator<u16> = SeqGenerator::new(0),
channels: HashMap<u16, mpsc::UnboundedSender<(u8, EasyBuf)>> = HashMap::new(), channels: HashMap<u16, mpsc::UnboundedSender<(u8, Bytes)>> = HashMap::new(),
} }
} }
@ -17,7 +17,7 @@ component! {
pub struct ChannelError; pub struct ChannelError;
pub struct Channel { pub struct Channel {
receiver: mpsc::UnboundedReceiver<(u8, EasyBuf)>, receiver: mpsc::UnboundedReceiver<(u8, Bytes)>,
state: ChannelState, state: ChannelState,
} }
@ -26,12 +26,12 @@ pub struct ChannelData(BiLock<Channel>);
pub enum ChannelEvent { pub enum ChannelEvent {
Header(u8, Vec<u8>), Header(u8, Vec<u8>),
Data(EasyBuf), Data(Bytes),
} }
#[derive(Clone)] #[derive(Clone)]
enum ChannelState { enum ChannelState {
Header(EasyBuf), Header(Bytes),
Data, Data,
Closed, Closed,
} }
@ -48,27 +48,27 @@ impl ChannelManager {
let channel = Channel { let channel = Channel {
receiver: rx, receiver: rx,
state: ChannelState::Header(EasyBuf::new()), state: ChannelState::Header(Bytes::new()),
}; };
(seq, channel) (seq, channel)
} }
pub fn dispatch(&self, cmd: u8, mut data: EasyBuf) { pub fn dispatch(&self, cmd: u8, mut data: Bytes) {
use std::collections::hash_map::Entry; use std::collections::hash_map::Entry;
let id: u16 = BigEndian::read_u16(data.drain_to(2).as_ref()); let id: u16 = BigEndian::read_u16(data.split_to(2).as_ref());
self.lock(|inner| { self.lock(|inner| {
if let Entry::Occupied(entry) = inner.channels.entry(id) { if let Entry::Occupied(entry) = inner.channels.entry(id) {
let _ = entry.get().send((cmd, data)); let _ = entry.get().unbounded_send((cmd, data));
} }
}); });
} }
} }
impl Channel { impl Channel {
fn recv_packet(&mut self) -> Poll<EasyBuf, ChannelError> { fn recv_packet(&mut self) -> Poll<Bytes, ChannelError> {
let (cmd, packet) = match self.receiver.poll() { let (cmd, packet) = match self.receiver.poll() {
Ok(Async::Ready(t)) => t.expect("channel closed"), Ok(Async::Ready(t)) => t.expect("channel closed"),
Ok(Async::NotReady) => return Ok(Async::NotReady), Ok(Async::NotReady) => return Ok(Async::NotReady),
@ -107,13 +107,13 @@ impl Stream for Channel {
data = try_ready!(self.recv_packet()); data = try_ready!(self.recv_packet());
} }
let length = BigEndian::read_u16(data.drain_to(2).as_ref()) as usize; let length = BigEndian::read_u16(data.split_to(2).as_ref()) as usize;
if length == 0 { if length == 0 {
assert_eq!(data.len(), 0); assert_eq!(data.len(), 0);
self.state = ChannelState::Data; self.state = ChannelState::Data;
} else { } else {
let header_id = data.drain_to(1).as_ref()[0]; let header_id = data.split_to(1).as_ref()[0];
let header_data = data.drain_to(length - 1).as_ref().to_owned(); let header_data = data.split_to(length - 1).as_ref().to_owned();
self.state = ChannelState::Header(data); self.state = ChannelState::Header(data);
@ -139,7 +139,7 @@ impl Stream for Channel {
} }
impl Stream for ChannelData { impl Stream for ChannelData {
type Item = EasyBuf; type Item = Bytes;
type Error = ChannelError; type Error = ChannelError;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {

View file

@ -1,7 +1,8 @@
use byteorder::{BigEndian, ByteOrder, WriteBytesExt}; use byteorder::{BigEndian, ByteOrder};
use bytes::{Bytes, BytesMut, BufMut};
use shannon::Shannon; use shannon::Shannon;
use std::io; use std::io;
use tokio_core::io::{Codec, EasyBuf}; use tokio_io::codec::{Decoder, Encoder};
const HEADER_SIZE: usize = 3; const HEADER_SIZE: usize = 3;
const MAC_SIZE: usize = 4; const MAC_SIZE: usize = 4;
@ -34,16 +35,17 @@ impl APCodec {
} }
} }
impl Codec for APCodec { impl Encoder for APCodec {
type Out = (u8, Vec<u8>); type Item = (u8, Vec<u8>);
type In = (u8, EasyBuf); type Error = io::Error;
fn encode(&mut self, item: (u8, Vec<u8>), buf: &mut Vec<u8>) -> io::Result<()> { fn encode(&mut self, item: (u8, Vec<u8>), buf: &mut BytesMut) -> io::Result<()> {
let (cmd, payload) = item; let (cmd, payload) = item;
let offset = buf.len(); let offset = buf.len();
buf.write_u8(cmd).unwrap(); buf.reserve(3 + payload.len());
buf.write_u16::<BigEndian>(payload.len() as u16).unwrap(); buf.put_u8(cmd);
buf.put_u16::<BigEndian>(payload.len() as u16);
buf.extend_from_slice(&payload); buf.extend_from_slice(&payload);
self.encode_cipher.nonce_u32(self.encode_nonce); self.encode_cipher.nonce_u32(self.encode_nonce);
@ -57,12 +59,17 @@ impl Codec for APCodec {
Ok(()) Ok(())
} }
}
fn decode(&mut self, buf: &mut EasyBuf) -> io::Result<Option<(u8, EasyBuf)>> { impl Decoder for APCodec {
type Item = (u8, Bytes);
type Error = io::Error;
fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<(u8, Bytes)>> {
if let DecodeState::Header = self.decode_state { if let DecodeState::Header = self.decode_state {
if buf.len() >= HEADER_SIZE { if buf.len() >= HEADER_SIZE {
let mut header = [0u8; HEADER_SIZE]; let mut header = [0u8; HEADER_SIZE];
header.copy_from_slice(buf.drain_to(HEADER_SIZE).as_slice()); header.copy_from_slice(buf.split_to(HEADER_SIZE).as_ref());
self.decode_cipher.nonce_u32(self.decode_nonce); self.decode_cipher.nonce_u32(self.decode_nonce);
self.decode_nonce += 1; self.decode_nonce += 1;
@ -79,13 +86,13 @@ impl Codec for APCodec {
if buf.len() >= size + MAC_SIZE { if buf.len() >= size + MAC_SIZE {
self.decode_state = DecodeState::Header; self.decode_state = DecodeState::Header;
let mut payload = buf.drain_to(size + MAC_SIZE); let mut payload = buf.split_to(size + MAC_SIZE);
self.decode_cipher.decrypt(&mut payload.get_mut()[..size]); self.decode_cipher.decrypt(&mut payload.get_mut(..size).unwrap());
let mac = payload.split_off(size); let mac = payload.split_off(size);
self.decode_cipher.check_mac(mac.as_slice())?; self.decode_cipher.check_mac(mac.as_ref())?;
return Ok(Some((cmd, payload))); return Ok(Some((cmd, payload.freeze())));
} }
} }

View file

@ -3,9 +3,11 @@ use crypto::hmac::Hmac;
use crypto::mac::Mac;use byteorder::{BigEndian, ByteOrder, WriteBytesExt}; use crypto::mac::Mac;use byteorder::{BigEndian, ByteOrder, WriteBytesExt};
use protobuf::{self, Message, MessageStatic}; use protobuf::{self, Message, MessageStatic};
use rand::thread_rng; use rand::thread_rng;
use std::io::{self, Read, Write}; use std::io::{self, Read};
use std::marker::PhantomData; use std::marker::PhantomData;
use tokio_core::io::{Io, Framed, write_all, WriteAll, read_exact, ReadExact, Window}; use tokio_io::{AsyncRead, AsyncWrite};
use tokio_io::codec::Framed;
use tokio_io::io::{write_all, WriteAll, read_exact, ReadExact, Window};
use futures::{Poll, Async, Future}; use futures::{Poll, Async, Future};
use diffie_hellman::DHLocalKeys; use diffie_hellman::DHLocalKeys;
@ -25,7 +27,7 @@ enum HandshakeState<T> {
ClientResponse(Option<APCodec>, WriteAll<T, Vec<u8>>), ClientResponse(Option<APCodec>, WriteAll<T, Vec<u8>>),
} }
pub fn handshake<T: Io>(connection: T) -> Handshake<T> { pub fn handshake<T: AsyncRead + AsyncWrite>(connection: T) -> Handshake<T> {
let local_keys = DHLocalKeys::random(&mut thread_rng()); let local_keys = DHLocalKeys::random(&mut thread_rng());
let client_hello = client_hello(connection, local_keys.public_key()); let client_hello = client_hello(connection, local_keys.public_key());
@ -35,7 +37,7 @@ pub fn handshake<T: Io>(connection: T) -> Handshake<T> {
} }
} }
impl <T: Io> Future for Handshake<T> { impl <T: AsyncRead + AsyncWrite> Future for Handshake<T> {
type Item = Framed<T, APCodec>; type Item = Framed<T, APCodec>;
type Error = io::Error; type Error = io::Error;
@ -78,7 +80,7 @@ impl <T: Io> Future for Handshake<T> {
} }
} }
fn client_hello<T: Write>(connection: T, gc: Vec<u8>) -> WriteAll<T, Vec<u8>> { fn client_hello<T: AsyncWrite>(connection: T, gc: Vec<u8>) -> WriteAll<T, Vec<u8>> {
let packet = protobuf_init!(ClientHello::new(), { let packet = protobuf_init!(ClientHello::new(), {
build_info => { build_info => {
product: protocol::keyexchange::Product::PRODUCT_PARTNER, product: protocol::keyexchange::Product::PRODUCT_PARTNER,
@ -104,7 +106,7 @@ fn client_hello<T: Write>(connection: T, gc: Vec<u8>) -> WriteAll<T, Vec<u8>> {
write_all(connection, buffer) write_all(connection, buffer)
} }
fn client_response<T: Write>(connection: T, challenge: Vec<u8>) -> WriteAll<T, Vec<u8>> { fn client_response<T: AsyncWrite>(connection: T, challenge: Vec<u8>) -> WriteAll<T, Vec<u8>> {
let packet = protobuf_init!(ClientResponsePlaintext::new(), { let packet = protobuf_init!(ClientResponsePlaintext::new(), {
login_crypto_response.diffie_hellman => { login_crypto_response.diffie_hellman => {
hmac: challenge hmac: challenge
@ -126,14 +128,14 @@ enum RecvPacket<T, M: MessageStatic> {
Body(ReadExact<T, Window<Vec<u8>>>, PhantomData<M>), Body(ReadExact<T, Window<Vec<u8>>>, PhantomData<M>),
} }
fn recv_packet<T, M>(connection: T, acc: Vec<u8>) -> RecvPacket<T, M> fn recv_packet<T: AsyncRead, M>(connection: T, acc: Vec<u8>) -> RecvPacket<T, M>
where T: Read, where T: Read,
M: MessageStatic M: MessageStatic
{ {
RecvPacket::Header(read_into_accumulator(connection, 4, acc), PhantomData) RecvPacket::Header(read_into_accumulator(connection, 4, acc), PhantomData)
} }
impl <T, M> Future for RecvPacket<T, M> impl <T: AsyncRead, M> Future for RecvPacket<T, M>
where T: Read, where T: Read,
M: MessageStatic M: MessageStatic
{ {
@ -165,7 +167,7 @@ impl <T, M> Future for RecvPacket<T, M>
} }
} }
fn read_into_accumulator<T: Read>(connection: T, size: usize, mut acc: Vec<u8>) -> ReadExact<T, Window<Vec<u8>>> { fn read_into_accumulator<T: AsyncRead>(connection: T, size: usize, mut acc: Vec<u8>) -> ReadExact<T, Window<Vec<u8>>> {
let offset = acc.len(); let offset = acc.len();
acc.resize(offset + size, 0); acc.resize(offset + size, 0);

View file

@ -4,12 +4,12 @@ mod handshake;
pub use self::codec::APCodec; pub use self::codec::APCodec;
pub use self::handshake::handshake; pub use self::handshake::handshake;
use futures::{Future, Sink, Stream, BoxFuture}; use futures::{Future, Sink, Stream};
use std::io; use std::io;
use std::net::ToSocketAddrs; use std::net::ToSocketAddrs;
use tokio_core::net::TcpStream; use tokio_core::net::TcpStream;
use tokio_core::reactor::Handle; use tokio_core::reactor::Handle;
use tokio_core::io::Framed; use tokio_io::codec::Framed;
use protobuf::{self, Message}; use protobuf::{self, Message};
use authentication::Credentials; use authentication::Credentials;
@ -17,18 +17,18 @@ use version;
pub type Transport = Framed<TcpStream, APCodec>; pub type Transport = Framed<TcpStream, APCodec>;
pub fn connect<A: ToSocketAddrs>(addr: A, handle: &Handle) -> BoxFuture<Transport, io::Error> { pub fn connect<A: ToSocketAddrs>(addr: A, handle: &Handle) -> Box<Future<Item = Transport, Error = io::Error>> {
let addr = addr.to_socket_addrs().unwrap().next().unwrap(); let addr = addr.to_socket_addrs().unwrap().next().unwrap();
let socket = TcpStream::connect(&addr, handle); let socket = TcpStream::connect(&addr, handle);
let connection = socket.and_then(|socket| { let connection = socket.and_then(|socket| {
handshake(socket) handshake(socket)
}); });
connection.boxed() Box::new(connection)
} }
pub fn authenticate(transport: Transport, credentials: Credentials, device_id: String) pub fn authenticate(transport: Transport, credentials: Credentials, device_id: String)
-> BoxFuture<(Transport, Credentials), io::Error> -> Box<Future<Item = (Transport, Credentials), Error = io::Error>>
{ {
use protocol::authentication::{APWelcome, ClientResponseEncrypted, CpuFamily, Os}; use protocol::authentication::{APWelcome, ClientResponseEncrypted, CpuFamily, Os};
@ -50,7 +50,7 @@ pub fn authenticate(transport: Transport, credentials: Credentials, device_id: S
let cmd = 0xab; let cmd = 0xab;
let data = packet.write_to_bytes().unwrap(); let data = packet.write_to_bytes().unwrap();
transport.send((cmd, data)).and_then(|transport| { Box::new(transport.send((cmd, data)).and_then(|transport| {
transport.into_future().map_err(|(err, _stream)| err) transport.into_future().map_err(|(err, _stream)| err)
}).and_then(|(packet, transport)| { }).and_then(|(packet, transport)| {
match packet { match packet {
@ -71,5 +71,5 @@ pub fn authenticate(transport: Transport, credentials: Credentials, device_id: S
Some((cmd, _)) => panic!("Unexpected packet {:?}", cmd), Some((cmd, _)) => panic!("Unexpected packet {:?}", cmd),
None => panic!("EOF"), None => panic!("EOF"),
} }
}).boxed() }))
} }

View file

@ -1,8 +1,5 @@
#![cfg_attr(feature = "cargo-clippy", allow(unused_io_amount))] #![cfg_attr(feature = "cargo-clippy", allow(unused_io_amount))]
// TODO: many items from tokio-core::io have been deprecated in favour of tokio-io
#![allow(deprecated)]
#[macro_use] extern crate error_chain; #[macro_use] extern crate error_chain;
#[macro_use] extern crate futures; #[macro_use] extern crate futures;
#[macro_use] extern crate lazy_static; #[macro_use] extern crate lazy_static;
@ -11,6 +8,7 @@
extern crate base64; extern crate base64;
extern crate byteorder; extern crate byteorder;
extern crate bytes;
extern crate crypto; extern crate crypto;
extern crate hyper; extern crate hyper;
extern crate num_bigint; extern crate num_bigint;
@ -23,6 +21,7 @@ extern crate serde;
extern crate serde_json; extern crate serde_json;
extern crate shannon; extern crate shannon;
extern crate tokio_core; extern crate tokio_core;
extern crate tokio_io;
extern crate uuid; extern crate uuid;
extern crate librespot_protocol as protocol; extern crate librespot_protocol as protocol;

View file

@ -1,11 +1,11 @@
use byteorder::{BigEndian, ByteOrder}; use byteorder::{BigEndian, ByteOrder};
use bytes::Bytes;
use futures::sync::{oneshot, mpsc}; use futures::sync::{oneshot, mpsc};
use futures::{Async, Poll, BoxFuture, Future}; use futures::{Async, Poll, Future};
use protobuf; use protobuf;
use protocol; use protocol;
use std::collections::HashMap; use std::collections::HashMap;
use std::mem; use std::mem;
use tokio_core::io::EasyBuf;
use util::SeqGenerator; use util::SeqGenerator;
@ -99,7 +99,7 @@ impl MercuryManager {
} }
pub fn subscribe<T: Into<String>>(&self, uri: T) pub fn subscribe<T: Into<String>>(&self, uri: T)
-> BoxFuture<mpsc::UnboundedReceiver<MercuryResponse>, MercuryError> -> Box<Future<Item = mpsc::UnboundedReceiver<MercuryResponse>, Error = MercuryError>>
{ {
let uri = uri.into(); let uri = uri.into();
let request = self.request(MercuryRequest { let request = self.request(MercuryRequest {
@ -110,7 +110,7 @@ impl MercuryManager {
}); });
let manager = self.clone(); let manager = self.clone();
request.map(move |response| { Box::new(request.map(move |response| {
let (tx, rx) = mpsc::unbounded(); let (tx, rx) = mpsc::unbounded();
manager.lock(move |inner| { manager.lock(move |inner| {
@ -133,15 +133,15 @@ impl MercuryManager {
}); });
rx rx
}).boxed() }))
} }
pub fn dispatch(&self, cmd: u8, mut data: EasyBuf) { pub fn dispatch(&self, cmd: u8, mut data: Bytes) {
let seq_len = BigEndian::read_u16(data.drain_to(2).as_ref()) as usize; let seq_len = BigEndian::read_u16(data.split_to(2).as_ref()) as usize;
let seq = data.drain_to(seq_len).as_ref().to_owned(); let seq = data.split_to(seq_len).as_ref().to_owned();
let flags = data.drain_to(1).as_ref()[0]; let flags = data.split_to(1).as_ref()[0];
let count = BigEndian::read_u16(data.drain_to(2).as_ref()) as usize; let count = BigEndian::read_u16(data.split_to(2).as_ref()) as usize;
let pending = self.lock(|inner| inner.pending.remove(&seq)); let pending = self.lock(|inner| inner.pending.remove(&seq));
@ -181,9 +181,9 @@ impl MercuryManager {
} }
} }
fn parse_part(data: &mut EasyBuf) -> Vec<u8> { fn parse_part(data: &mut Bytes) -> Vec<u8> {
let size = BigEndian::read_u16(data.drain_to(2).as_ref()) as usize; let size = BigEndian::read_u16(data.split_to(2).as_ref()) as usize;
data.drain_to(size).as_ref().to_owned() data.split_to(size).as_ref().to_owned()
} }
fn complete_request(&self, cmd: u8, mut pending: MercuryPending) { fn complete_request(&self, cmd: u8, mut pending: MercuryPending) {
@ -199,7 +199,7 @@ impl MercuryManager {
if response.status_code >= 400 { if response.status_code >= 400 {
warn!("error {} for uri {}", response.status_code, &response.uri); warn!("error {} for uri {}", response.status_code, &response.uri);
if let Some(cb) = pending.callback { if let Some(cb) = pending.callback {
cb.complete(Err(MercuryError)); let _ = cb.send(Err(MercuryError));
} }
} else { } else {
if cmd == 0xb5 { if cmd == 0xb5 {
@ -211,7 +211,7 @@ impl MercuryManager {
// if send fails, remove from list of subs // if send fails, remove from list of subs
// TODO: send unsub message // TODO: send unsub message
sub.send(response.clone()).is_ok() sub.unbounded_send(response.clone()).is_ok()
} else { } else {
// URI doesn't match // URI doesn't match
true true
@ -223,7 +223,7 @@ impl MercuryManager {
} }
}) })
} else if let Some(cb) = pending.callback { } else if let Some(cb) = pending.callback {
cb.complete(Ok(response)); let _ = cb.send(Ok(response));
} }
} }
} }

View file

@ -1,10 +1,10 @@
use bytes::Bytes;
use crypto::digest::Digest; use crypto::digest::Digest;
use crypto::sha1::Sha1; use crypto::sha1::Sha1;
use futures::sync::mpsc; use futures::sync::mpsc;
use futures::{Future, Stream, BoxFuture, IntoFuture, Poll, Async}; use futures::{Future, Stream, IntoFuture, Poll, Async};
use std::io; use std::io;
use std::sync::{RwLock, Arc, Weak}; use std::sync::{RwLock, Arc, Weak};
use tokio_core::io::EasyBuf;
use tokio_core::reactor::{Handle, Remote}; use tokio_core::reactor::{Handle, Remote};
use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering}; use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering};
@ -90,7 +90,7 @@ impl Session {
fn create(handle: &Handle, transport: connection::Transport, fn create(handle: &Handle, transport: connection::Transport,
config: SessionConfig, cache: Option<Cache>, username: String) config: SessionConfig, cache: Option<Cache>, username: String)
-> (Session, BoxFuture<(), io::Error>) -> (Session, Box<Future<Item = (), Error = io::Error>>)
{ {
let (sink, stream) = transport.split(); let (sink, stream) = transport.split();
@ -124,8 +124,8 @@ impl Session {
.forward(sink).map(|_| ()); .forward(sink).map(|_| ());
let receiver_task = DispatchTask(stream, session.weak()); let receiver_task = DispatchTask(stream, session.weak());
let task = (receiver_task, sender_task).into_future() let task = Box::new((receiver_task, sender_task).into_future()
.map(|((), ())| ()).boxed(); .map(|((), ())| ()));
(session, task) (session, task)
} }
@ -156,7 +156,7 @@ impl Session {
} }
#[cfg_attr(feature = "cargo-clippy", allow(match_same_arms))] #[cfg_attr(feature = "cargo-clippy", allow(match_same_arms))]
fn dispatch(&self, cmd: u8, data: EasyBuf) { fn dispatch(&self, cmd: u8, data: Bytes) {
match cmd { match cmd {
0x4 => { 0x4 => {
self.debug_info(); self.debug_info();
@ -177,7 +177,7 @@ impl Session {
} }
pub fn send_packet(&self, cmd: u8, data: Vec<u8>) { pub fn send_packet(&self, cmd: u8, data: Vec<u8>) {
self.0.tx_connection.send((cmd, data)).unwrap(); self.0.tx_connection.unbounded_send((cmd, data)).unwrap();
} }
pub fn cache(&self) -> Option<&Arc<Cache>> { pub fn cache(&self) -> Option<&Arc<Cache>> {
@ -229,10 +229,10 @@ impl Drop for SessionInternal {
} }
struct DispatchTask<S>(S, SessionWeak) struct DispatchTask<S>(S, SessionWeak)
where S: Stream<Item = (u8, EasyBuf)>; where S: Stream<Item = (u8, Bytes)>;
impl <S> Future for DispatchTask<S> impl <S> Future for DispatchTask<S>
where S: Stream<Item = (u8, EasyBuf)> where S: Stream<Item = (u8, Bytes)>
{ {
type Item = (); type Item = ();
type Error = S::Error; type Error = S::Error;
@ -253,7 +253,7 @@ impl <S> Future for DispatchTask<S>
} }
impl <S> Drop for DispatchTask<S> impl <S> Drop for DispatchTask<S>
where S: Stream<Item = (u8, EasyBuf)> where S: Stream<Item = (u8, Bytes)>
{ {
fn drop(&mut self) { fn drop(&mut self) {
debug!("drop Dispatch"); debug!("drop Dispatch");

View file

@ -2,6 +2,8 @@ use std;
use std::fmt; use std::fmt;
use util::u128; use util::u128;
use byteorder::{BigEndian, ByteOrder}; use byteorder::{BigEndian, ByteOrder};
// Unneeded since 1.21
#[allow(unused_imports)]
use std::ascii::AsciiExt; use std::ascii::AsciiExt;
#[derive(Debug,Copy,Clone,PartialEq,Eq,Hash)] #[derive(Debug,Copy,Clone,PartialEq,Eq,Hash)]

View file

@ -8,7 +8,7 @@ extern crate librespot_protocol as protocol;
pub mod cover; pub mod cover;
use futures::{Future, BoxFuture}; use futures::Future;
use linear_map::LinearMap; use linear_map::LinearMap;
use core::mercury::MercuryError; use core::mercury::MercuryError;
@ -57,17 +57,17 @@ pub trait Metadata : Send + Sized + 'static {
fn base_url() -> &'static str; fn base_url() -> &'static str;
fn parse(msg: &Self::Message, session: &Session) -> Self; fn parse(msg: &Self::Message, session: &Session) -> Self;
fn get(session: &Session, id: SpotifyId) -> BoxFuture<Self, MercuryError> { fn get(session: &Session, id: SpotifyId) -> Box<Future<Item = Self, Error = MercuryError>> {
let uri = format!("{}/{}", Self::base_url(), id.to_base16()); let uri = format!("{}/{}", Self::base_url(), id.to_base16());
let request = session.mercury().get(uri); let request = session.mercury().get(uri);
let session = session.clone(); let session = session.clone();
request.and_then(move |response| { Box::new(request.and_then(move |response| {
let data = response.payload.first().expect("Empty payload"); let data = response.payload.first().expect("Empty payload");
let msg: Self::Message = protobuf::parse_from_bytes(data).unwrap(); let msg: Self::Message = protobuf::parse_from_bytes(data).unwrap();
Ok(Self::parse(&msg, &session)) Ok(Self::parse(&msg, &session))
}).boxed() }))
} }
} }

View file

@ -3,8 +3,8 @@ use crypto::digest::Digest;
use crypto::mac::Mac; use crypto::mac::Mac;
use crypto; use crypto;
use futures::sync::mpsc; use futures::sync::mpsc;
use futures::{Future, Stream, BoxFuture, Poll, Async}; use futures::{Future, Stream, Poll};
use hyper::server::{Service, NewService, Request, Response, Http}; use hyper::server::{Service, Request, Response, Http};
use hyper::{self, Get, Post, StatusCode}; use hyper::{self, Get, Post, StatusCode};
use mdns; use mdns;
use num_bigint::BigUint; use num_bigint::BigUint;
@ -12,7 +12,6 @@ use rand;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::io; use std::io;
use std::sync::Arc; use std::sync::Arc;
use tokio_core::net::TcpListener;
use tokio_core::reactor::Handle; use tokio_core::reactor::Handle;
use url; use url;
@ -32,7 +31,7 @@ struct DiscoveryInner {
} }
impl Discovery { impl Discovery {
pub fn new(config: ConnectConfig, device_id: String) fn new(config: ConnectConfig, device_id: String)
-> (Discovery, mpsc::UnboundedReceiver<Credentials>) -> (Discovery, mpsc::UnboundedReceiver<Credentials>)
{ {
let (tx, rx) = mpsc::unbounded(); let (tx, rx) = mpsc::unbounded();
@ -136,7 +135,7 @@ impl Discovery {
let credentials = Credentials::with_blob(username.to_owned(), &decrypted, &self.0.device_id); let credentials = Credentials::with_blob(username.to_owned(), &decrypted, &self.0.device_id);
self.0.tx.send(credentials).unwrap(); self.0.tx.unbounded_send(credentials).unwrap();
let result = json!({ let result = json!({
"status": 101, "status": 101,
@ -159,7 +158,7 @@ impl Service for Discovery {
type Request = Request; type Request = Request;
type Response = Response; type Response = Response;
type Error = hyper::Error; type Error = hyper::Error;
type Future = BoxFuture<Response, hyper::Error>; type Future = Box<Future<Item = Response, Error = hyper::Error>>;
fn call(&self, request: Request) -> Self::Future { fn call(&self, request: Request) -> Self::Future {
let mut params = BTreeMap::new(); let mut params = BTreeMap::new();
@ -174,7 +173,7 @@ impl Service for Discovery {
} }
let this = self.clone(); let this = self.clone();
body.fold(Vec::new(), |mut acc, chunk| { Box::new(body.fold(Vec::new(), |mut acc, chunk| {
acc.extend_from_slice(chunk.as_ref()); acc.extend_from_slice(chunk.as_ref());
Ok::<_, hyper::Error>(acc) Ok::<_, hyper::Error>(acc)
}).map(move |body| { }).map(move |body| {
@ -186,25 +185,13 @@ impl Service for Discovery {
(Post, Some("addUser")) => this.handle_add_user(&params), (Post, Some("addUser")) => this.handle_add_user(&params),
_ => this.not_found(), _ => this.not_found(),
} }
}).boxed() }))
}
}
impl NewService for Discovery {
type Request = Request;
type Response = Response;
type Error = hyper::Error;
type Instance = Self;
fn new_service(&self) -> io::Result<Self::Instance> {
Ok(self.clone())
} }
} }
pub struct DiscoveryStream { pub struct DiscoveryStream {
credentials: mpsc::UnboundedReceiver<Credentials>, credentials: mpsc::UnboundedReceiver<Credentials>,
_svc: mdns::Service, _svc: mdns::Service,
task: Box<Future<Item=(), Error=io::Error>>,
} }
pub fn discovery(handle: &Handle, config: ConnectConfig, device_id: String) pub fn discovery(handle: &Handle, config: ConnectConfig, device_id: String)
@ -212,15 +199,20 @@ pub fn discovery(handle: &Handle, config: ConnectConfig, device_id: String)
{ {
let (discovery, creds_rx) = Discovery::new(config.clone(), device_id); let (discovery, creds_rx) = Discovery::new(config.clone(), device_id);
let listener = TcpListener::bind(&"0.0.0.0:0".parse().unwrap(), handle)?; let serve = {
let addr = listener.local_addr()?; let http = Http::new();
http.serve_addr_handle(&"0.0.0.0:0".parse().unwrap(), &handle, move || Ok(discovery.clone())).unwrap()
let http = Http::new(); };
let handle_ = handle.clone(); let addr = serve.incoming_ref().local_addr();
let task = Box::new(listener.incoming().for_each(move |(socket, addr)| { let server_future = {
http.bind_connection(&handle_, socket, addr, discovery.clone()); let handle = handle.clone();
Ok(()) serve.for_each(move |connection| {
})); handle.spawn(connection.then(|_| Ok(())));
Ok(())
})
.then(|_| Ok(()))
};
handle.spawn(server_future);
let responder = mdns::Responder::spawn(&handle)?; let responder = mdns::Responder::spawn(&handle)?;
let svc = responder.register( let svc = responder.register(
@ -232,20 +224,14 @@ pub fn discovery(handle: &Handle, config: ConnectConfig, device_id: String)
Ok(DiscoveryStream { Ok(DiscoveryStream {
credentials: creds_rx, credentials: creds_rx,
_svc: svc, _svc: svc,
task: task,
}) })
} }
impl Stream for DiscoveryStream { impl Stream for DiscoveryStream {
type Item = Credentials; type Item = Credentials;
type Error = io::Error; type Error = ();
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
match self.task.poll()? { self.credentials.poll()
Async::Ready(()) => unreachable!(),
Async::NotReady => (),
}
Ok(self.credentials.poll().unwrap())
} }
} }

View file

@ -1,4 +1,4 @@
use futures::{Future, BoxFuture}; use futures::Future;
use serde_json; use serde_json;
use core::mercury::MercuryError; use core::mercury::MercuryError;
@ -13,14 +13,14 @@ pub struct Token {
pub scope: Vec<String>, pub scope: Vec<String>,
} }
pub fn get_token(session: &Session, client_id: &str, scopes: &str) -> BoxFuture<Token, MercuryError> { pub fn get_token(session: &Session, client_id: &str, scopes: &str) -> Box<Future<Item = Token, Error = MercuryError>> {
let url = format!("hm://keymaster/token/authenticated?client_id={}&scope={}", let url = format!("hm://keymaster/token/authenticated?client_id={}&scope={}",
client_id, scopes); client_id, scopes);
session.mercury().get(url).map(move |response| { Box::new(session.mercury().get(url).map(move |response| {
let data = response.payload.first().expect("Empty payload"); let data = response.payload.first().expect("Empty payload");
let data = String::from_utf8(data.clone()).unwrap(); let data = String::from_utf8(data.clone()).unwrap();
let token : Token = serde_json::from_str(&data).unwrap(); let token : Token = serde_json::from_str(&data).unwrap();
token token
}).boxed() }))
} }

View file

@ -2,9 +2,6 @@
#![cfg_attr(feature = "cargo-clippy", allow(unused_io_amount))] #![cfg_attr(feature = "cargo-clippy", allow(unused_io_amount))]
// TODO: many items from tokio-core::io have been deprecated in favour of tokio-io
#![allow(deprecated)]
#[macro_use] extern crate log; #[macro_use] extern crate log;
#[macro_use] extern crate serde_json; #[macro_use] extern crate serde_json;
#[macro_use] extern crate serde_derive; #[macro_use] extern crate serde_derive;

View file

@ -1,12 +1,10 @@
// TODO: many items from tokio-core::io have been deprecated in favour of tokio-io
#![allow(deprecated)]
#[macro_use] extern crate log; #[macro_use] extern crate log;
extern crate env_logger; extern crate env_logger;
extern crate futures; extern crate futures;
extern crate getopts; extern crate getopts;
extern crate librespot; extern crate librespot;
extern crate tokio_core; extern crate tokio_core;
extern crate tokio_io;
extern crate tokio_signal; extern crate tokio_signal;
use env_logger::LogBuilder; use env_logger::LogBuilder;
@ -17,7 +15,7 @@ use std::path::PathBuf;
use std::process::exit; use std::process::exit;
use std::str::FromStr; use std::str::FromStr;
use tokio_core::reactor::{Handle, Core}; use tokio_core::reactor::{Handle, Core};
use tokio_core::io::IoStream; use tokio_io::IoStream;
use std::mem; use std::mem;
use librespot::core::authentication::{get_credentials, Credentials}; use librespot::core::authentication::{get_credentials, Credentials};
@ -264,7 +262,7 @@ impl Main {
spirc: None, spirc: None,
spirc_task: None, spirc_task: None,
shutdown: false, shutdown: false,
signal: tokio_signal::ctrl_c(&handle).flatten_stream().boxed(), signal: Box::new(tokio_signal::ctrl_c(&handle).flatten_stream()),
}; };
if setup.enable_discovery { if setup.enable_discovery {

View file

@ -155,7 +155,7 @@ impl PlayerState {
match self { match self {
Paused { end_of_track, .. } | Paused { end_of_track, .. } |
Playing { end_of_track, .. } => { Playing { end_of_track, .. } => {
end_of_track.complete(()) let _ = end_of_track.send(());
} }
Stopped => warn!("signal_end_of_track from stopped state"), Stopped => warn!("signal_end_of_track from stopped state"),
@ -313,7 +313,7 @@ impl PlayerInternal {
} }
None => { None => {
end_of_track.complete(()); let _ = end_of_track.send(());
if self.state.is_playing() { if self.state.is_playing() {
self.run_onstop(); self.run_onstop();
} }

View file

@ -1,8 +1,6 @@
use futures::future; use futures::future;
use futures::sink::BoxSink;
use futures::stream::BoxStream;
use futures::sync::{oneshot, mpsc}; use futures::sync::{oneshot, mpsc};
use futures::{Future, Stream, Sink, Async, Poll, BoxFuture}; use futures::{Future, Stream, Sink, Async, Poll};
use protobuf::{self, Message}; use protobuf::{self, Message};
use core::config::ConnectConfig; use core::config::ConnectConfig;
@ -30,10 +28,10 @@ pub struct SpircTask {
device: DeviceState, device: DeviceState,
state: State, state: State,
subscription: BoxStream<Frame, MercuryError>, subscription: Box<Stream<Item = Frame, Error = MercuryError>>,
sender: BoxSink<Frame, MercuryError>, sender: Box<Sink<SinkItem = Frame, SinkError = MercuryError>>,
commands: mpsc::UnboundedReceiver<SpircCommand>, commands: mpsc::UnboundedReceiver<SpircCommand>,
end_of_track: BoxFuture<(), oneshot::Canceled>, end_of_track: Box<Future<Item = (), Error = oneshot::Canceled>>,
shutdown: bool, shutdown: bool,
session: Session, session: Session,
@ -134,10 +132,10 @@ impl Spirc {
let subscription = session.mercury().subscribe(&uri as &str); let subscription = session.mercury().subscribe(&uri as &str);
let subscription = subscription.map(|stream| stream.map_err(|_| MercuryError)).flatten_stream(); let subscription = subscription.map(|stream| stream.map_err(|_| MercuryError)).flatten_stream();
let subscription = subscription.map(|response| -> Frame { let subscription = Box::new(subscription.map(|response| -> Frame {
let data = response.payload.first().unwrap(); let data = response.payload.first().unwrap();
protobuf::parse_from_bytes(data).unwrap() protobuf::parse_from_bytes(data).unwrap()
}).boxed(); }));
let sender = Box::new(session.mercury().sender(uri).with(|frame: Frame| { let sender = Box::new(session.mercury().sender(uri).with(|frame: Frame| {
Ok(frame.write_to_bytes().unwrap()) Ok(frame.write_to_bytes().unwrap())
@ -163,7 +161,7 @@ impl Spirc {
subscription: subscription, subscription: subscription,
sender: sender, sender: sender,
commands: cmd_rx, commands: cmd_rx,
end_of_track: future::empty().boxed(), end_of_track: Box::new(future::empty()),
shutdown: false, shutdown: false,
session: session.clone(), session: session.clone(),
@ -179,28 +177,28 @@ impl Spirc {
} }
pub fn play(&self) { pub fn play(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::Play); let _ = self.commands.unbounded_send(SpircCommand::Play);
} }
pub fn play_pause(&self) { pub fn play_pause(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::PlayPause); let _ = self.commands.unbounded_send(SpircCommand::PlayPause);
} }
pub fn pause(&self) { pub fn pause(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::Pause); let _ = self.commands.unbounded_send(SpircCommand::Pause);
} }
pub fn prev(&self) { pub fn prev(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::Prev); let _ = self.commands.unbounded_send(SpircCommand::Prev);
} }
pub fn next(&self) { pub fn next(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::Next); let _ = self.commands.unbounded_send(SpircCommand::Next);
} }
pub fn volume_up(&self) { pub fn volume_up(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::VolumeUp); let _ = self.commands.unbounded_send(SpircCommand::VolumeUp);
} }
pub fn volume_down(&self) { pub fn volume_down(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::VolumeDown); let _ = self.commands.unbounded_send(SpircCommand::VolumeDown);
} }
pub fn shutdown(&self) { pub fn shutdown(&self) {
let _ = mpsc::UnboundedSender::send(&self.commands, SpircCommand::Shutdown); let _ = self.commands.unbounded_send(SpircCommand::Shutdown);
} }
} }
@ -238,7 +236,7 @@ impl Future for SpircTask {
} }
Ok(Async::NotReady) => (), Ok(Async::NotReady) => (),
Err(oneshot::Canceled) => { Err(oneshot::Canceled) => {
self.end_of_track = future::empty().boxed() self.end_of_track = Box::new(future::empty())
} }
} }
} }
@ -587,7 +585,7 @@ impl SpircTask {
self.state.set_status(PlayStatus::kPlayStatusPause); self.state.set_status(PlayStatus::kPlayStatusPause);
} }
self.end_of_track = end_of_track.boxed(); self.end_of_track = Box::new(end_of_track);
} }
fn hello(&mut self) { fn hello(&mut self) {