OMGeeky/tarpc
1
0
Fork 0
mirror of https://github.com/OMGeeky/tarpc.git synced 2026-02-23 15:49:54 +01:00
Code Issues Packages Projects Releases Wiki Activity

format with rustfmt 0.8.3 (#148)

Browse Source
This commit is contained in:
Jon Gjengset
2017-04-26 15:25:49 -04:00
committed by Tim
parent e2728d84f3
commit 8bc01a993b
21 changed files with 363 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
benches/latency.rs
View File

@@ -40,7 +40,8 @@ impl FutureService for Server {
fn latency(bencher: &mut Bencher) { fn latency(bencher: &mut Bencher) {
let _ = env_logger::init(); let _ = env_logger::init();
let mut reactor = reactor::Core::new().unwrap(); let mut reactor = reactor::Core::new().unwrap();
let (handle, server) = Server.listen("localhost:0".first_socket_addr(), let (handle, server) = Server
.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();

111
examples/concurrency.rs
View File

@@ -57,13 +57,13 @@ impl FutureService for Server {
debug!("Server received read({}) no. {}", size, request_number); debug!("Server received read({}) no. {}", size, request_number);
self.pool self.pool
.spawn(futures::lazy(move || { .spawn(futures::lazy(move || {
let mut vec = Vec::with_capacity(size as usize); let mut vec = Vec::with_capacity(size as usize);
for i in 0..size { for i in 0..size {
vec.push(((i % 2) << 8) as u8); vec.push(((i % 2) << 8) as u8);
} }
debug!("Server sending response no. {}", request_number); debug!("Server sending response no. {}", request_number);
Ok(vec.into()) Ok(vec.into())
})) }))
} }
} }
@@ -94,12 +94,12 @@ struct Stats {
fn spawn_core() -> reactor::Remote { fn spawn_core() -> reactor::Remote {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
thread::spawn(move || { thread::spawn(move || {
let mut core = reactor::Core::new().unwrap(); let mut core = reactor::Core::new().unwrap();
tx.send(core.handle().remote().clone()).unwrap(); tx.send(core.handle().remote().clone()).unwrap();
// Run forever // Run forever
core.run(futures::empty::<(), !>()).unwrap(); core.run(futures::empty::<(), !>()).unwrap();
}); });
rx.recv().unwrap() rx.recv().unwrap()
} }
@@ -108,36 +108,37 @@ fn run_once(clients: Vec<FutureClient>,
-> impl Future<Item = (), Error = ()> + 'static { -> impl Future<Item = (), Error = ()> + 'static {
let start = Instant::now(); let start = Instant::now();
futures::stream::futures_unordered((0..concurrency as usize) futures::stream::futures_unordered((0..concurrency as usize)
.zip(clients.iter().enumerate().cycle()) .zip(clients.iter().enumerate().cycle())
.map(|(iteration, (client_idx, client))| { .map(|(iteration, (client_idx, client))| {
let start = Instant::now(); let start = Instant::now();
debug!("Client {} reading (iteration {})...", client_idx, iteration); debug!("Client {} reading (iteration {})...", client_idx, iteration);
client.read(CHUNK_SIZE) client
.map(move |_| (client_idx, iteration, start)) .read(CHUNK_SIZE)
})) .map(move |_| (client_idx, iteration, start))
.map(|(client_idx, iteration, start)| { }))
let elapsed = start.elapsed(); .map(|(client_idx, iteration, start)| {
debug!("Client {} received reply (iteration {}).", let elapsed = start.elapsed();
client_idx, debug!("Client {} received reply (iteration {}).",
iteration); client_idx,
elapsed iteration);
}) elapsed
.map_err(|e| panic!(e)) })
.fold(Stats::default(), move |mut stats, elapsed| { .map_err(|e| panic!(e))
stats.sum += elapsed; .fold(Stats::default(), move |mut stats, elapsed| {
stats.count += 1; stats.sum += elapsed;
stats.min = Some(cmp::min(stats.min.unwrap_or(elapsed), elapsed)); stats.count += 1;
stats.max = Some(cmp::max(stats.max.unwrap_or(elapsed), elapsed)); stats.min = Some(cmp::min(stats.min.unwrap_or(elapsed), elapsed));
Ok(stats) stats.max = Some(cmp::max(stats.max.unwrap_or(elapsed), elapsed));
}) Ok(stats)
.map(move |stats| { })
info!("{} requests => Mean={}µs, Min={}µs, Max={}µs, Total={}µs", .map(move |stats| {
stats.count, info!("{} requests => Mean={}µs, Min={}µs, Max={}µs, Total={}µs",
stats.sum.microseconds() as f64 / stats.count as f64, stats.count,
stats.min.unwrap().microseconds(), stats.sum.microseconds() as f64 / stats.count as f64,
stats.max.unwrap().microseconds(), stats.min.unwrap().microseconds(),
start.elapsed().microseconds()); stats.max.unwrap().microseconds(),
}) start.elapsed().microseconds());
})
} }
fn main() { fn main() {
@@ -145,23 +146,25 @@ fn main() {
let matches = App::new("Tarpc Concurrency") let matches = App::new("Tarpc Concurrency")
.about("Demonstrates making concurrent requests to a tarpc service.") .about("Demonstrates making concurrent requests to a tarpc service.")
.arg(Arg::with_name("concurrency") .arg(Arg::with_name("concurrency")
.short("c") .short("c")
.long("concurrency") .long("concurrency")
.value_name("LEVEL") .value_name("LEVEL")
.help("Sets a custom concurrency level") .help("Sets a custom concurrency level")
.takes_value(true)) .takes_value(true))
.arg(Arg::with_name("clients") .arg(Arg::with_name("clients")
.short("n") .short("n")
.long("num_clients") .long("num_clients")
.value_name("AMOUNT") .value_name("AMOUNT")
.help("How many clients to distribute requests between") .help("How many clients to distribute requests between")
.takes_value(true)) .takes_value(true))
.get_matches(); .get_matches();
let concurrency = matches.value_of("concurrency") let concurrency = matches
.value_of("concurrency")
.map(&str::parse) .map(&str::parse)
.map(Result::unwrap) .map(Result::unwrap)
.unwrap_or(10); .unwrap_or(10);
let num_clients = matches.value_of("clients") let num_clients = matches
.value_of("clients")
.map(&str::parse) .map(&str::parse)
.map(Result::unwrap) .map(Result::unwrap)
.unwrap_or(4); .unwrap_or(4);

44
examples/pubsub.rs
View File

@@ -58,10 +58,7 @@ impl subscriber::FutureService for Subscriber {
} }
impl Subscriber { impl Subscriber {
fn listen(id: u32, fn listen(id: u32, handle: &reactor::Handle, options: server::Options) -> server::Handle {
handle: &reactor::Handle,
options: server::Options)
-> server::Handle {
let (server_handle, server) = Subscriber { id: id } let (server_handle, server) = Subscriber { id: id }
.listen("localhost:0".first_socket_addr(), handle, options) .listen("localhost:0".first_socket_addr(), handle, options)
.unwrap(); .unwrap();
@@ -103,12 +100,12 @@ impl publisher::FutureService for Publisher {
fn subscribe(&self, id: u32, address: SocketAddr) -> Self::SubscribeFut { fn subscribe(&self, id: u32, address: SocketAddr) -> Self::SubscribeFut {
let clients = self.clients.clone(); let clients = self.clients.clone();
Box::new(subscriber::FutureClient::connect(address, client::Options::default()) Box::new(subscriber::FutureClient::connect(address, client::Options::default())
.map(move |subscriber| { .map(move |subscriber| {
println!("Subscribing {}.", id); println!("Subscribing {}.", id);
clients.borrow_mut().insert(id, subscriber); clients.borrow_mut().insert(id, subscriber);
() ()
}) })
.map_err(|e| e.to_string().into())) .map_err(|e| e.to_string().into()))
} }
type UnsubscribeFut = Box<Future<Item = (), Error = Never>>; type UnsubscribeFut = Box<Future<Item = (), Error = Never>>;
@@ -133,19 +130,20 @@ fn main() {
let subscriber1 = Subscriber::listen(0, &reactor.handle(), server::Options::default()); let subscriber1 = Subscriber::listen(0, &reactor.handle(), server::Options::default());
let subscriber2 = Subscriber::listen(1, &reactor.handle(), server::Options::default()); let subscriber2 = Subscriber::listen(1, &reactor.handle(), server::Options::default());
let publisher = let publisher = reactor
reactor.run(publisher::FutureClient::connect(publisher_handle.addr(), .run(publisher::FutureClient::connect(publisher_handle.addr(), client::Options::default()))
client::Options::default())) .unwrap();
.unwrap(); reactor
reactor.run(publisher.subscribe(0, subscriber1.addr()) .run(publisher
.and_then(|_| publisher.subscribe(1, subscriber2.addr())) .subscribe(0, subscriber1.addr())
.map_err(|e| panic!(e)) .and_then(|_| publisher.subscribe(1, subscriber2.addr()))
.and_then(|_| { .map_err(|e| panic!(e))
println!("Broadcasting..."); .and_then(|_| {
publisher.broadcast("hello to all".to_string()) println!("Broadcasting...");
}) publisher.broadcast("hello to all".to_string())
.and_then(|_| publisher.unsubscribe(1)) })
.and_then(|_| publisher.broadcast("hi again".to_string()))) .and_then(|_| publisher.unsubscribe(1))
.and_then(|_| publisher.broadcast("hi again".to_string())))
.unwrap(); .unwrap();
thread::sleep(Duration::from_millis(300)); thread::sleep(Duration::from_millis(300));
} }

10
examples/readme_errors.rs
View File

@@ -55,10 +55,12 @@ impl SyncService for HelloServer {
fn main() { fn main() {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
thread::spawn(move || { thread::spawn(move || {
let handle = HelloServer.listen("localhost:10000", server::Options::default()).unwrap(); let handle = HelloServer
tx.send(handle.addr()).unwrap(); .listen("localhost:10000", server::Options::default())
handle.run(); .unwrap();
}); tx.send(handle.addr()).unwrap();
handle.run();
});
let client = SyncClient::connect(rx.recv().unwrap(), client::Options::default()).unwrap(); let client = SyncClient::connect(rx.recv().unwrap(), client::Options::default()).unwrap();
println!("{}", client.hello("Mom".to_string()).unwrap()); println!("{}", client.hello("Mom".to_string()).unwrap());
println!("{}", client.hello("".to_string()).unwrap_err()); println!("{}", client.hello("".to_string()).unwrap_err());

12
examples/readme_futures.rs
View File

@@ -34,16 +34,18 @@ impl FutureService for HelloServer {
fn main() { fn main() {
let mut reactor = reactor::Core::new().unwrap(); let mut reactor = reactor::Core::new().unwrap();
let (handle, server) = HelloServer.listen("localhost:10000".first_socket_addr(), let (handle, server) = HelloServer
.listen("localhost:10000".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();
reactor.handle().spawn(server); reactor.handle().spawn(server);
let options = client::Options::default().handle(reactor.handle()); let options = client::Options::default().handle(reactor.handle());
reactor.run(FutureClient::connect(handle.addr(), options) reactor
.map_err(tarpc::Error::from) .run(FutureClient::connect(handle.addr(), options)
.and_then(|client| client.hello("Mom".to_string())) .map_err(tarpc::Error::from)
.map(|resp| println!("{}", resp))) .and_then(|client| client.hello("Mom".to_string()))
.map(|resp| println!("{}", resp)))
.unwrap(); .unwrap();
} }

14
examples/readme_sync.rs
View File

@@ -27,17 +27,21 @@ struct HelloServer;
impl SyncService for HelloServer { impl SyncService for HelloServer {
fn hello(&self, name: String) -> Result<String, Never> { fn hello(&self, name: String) -> Result<String, Never> {
Ok(format!("Hello from thread {}, {}!", thread::current().name().unwrap(), name)) Ok(format!("Hello from thread {}, {}!",
thread::current().name().unwrap(),
name))
} }
} }
fn main() { fn main() {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
thread::spawn(move || { thread::spawn(move || {
let handle = HelloServer.listen("localhost:0", server::Options::default()).unwrap(); let handle = HelloServer
tx.send(handle.addr()).unwrap(); .listen("localhost:0", server::Options::default())
handle.run(); .unwrap();
}); tx.send(handle.addr()).unwrap();
handle.run();
});
let client = SyncClient::connect(rx.recv().unwrap(), client::Options::default()).unwrap(); let client = SyncClient::connect(rx.recv().unwrap(), client::Options::default()).unwrap();
println!("{}", client.hello("Mom".to_string()).unwrap()); println!("{}", client.hello("Mom".to_string()).unwrap());
} }

26
examples/server_calling_server.rs
View File

@@ -72,14 +72,17 @@ impl DoubleFutureService for DoubleServer {
fn main() { fn main() {
let _ = env_logger::init(); let _ = env_logger::init();
let mut reactor = reactor::Core::new().unwrap(); let mut reactor = reactor::Core::new().unwrap();
let (add, server) = AddServer.listen("localhost:0".first_socket_addr(), let (add, server) = AddServer
.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();
reactor.handle().spawn(server); reactor.handle().spawn(server);
let options = client::Options::default().handle(reactor.handle()); let options = client::Options::default().handle(reactor.handle());
let add_client = reactor.run(add::FutureClient::connect(add.addr(), options)).unwrap(); let add_client = reactor
.run(add::FutureClient::connect(add.addr(), options))
.unwrap();
let (double, server) = DoubleServer::new(add_client) let (double, server) = DoubleServer::new(add_client)
.listen("localhost:0".first_socket_addr(), .listen("localhost:0".first_socket_addr(),
@@ -88,14 +91,15 @@ fn main() {
.unwrap(); .unwrap();
reactor.handle().spawn(server); reactor.handle().spawn(server);
let double_client = let double_client = reactor
reactor.run(double::FutureClient::connect(double.addr(), client::Options::default())) .run(double::FutureClient::connect(double.addr(), client::Options::default()))
.unwrap(); .unwrap();
reactor.run(futures::stream::futures_unordered((0..5).map(|i| double_client.double(i))) reactor
.map_err(|e| println!("{}", e)) .run(futures::stream::futures_unordered((0..5).map(|i| double_client.double(i)))
.for_each(|i| { .map_err(|e| println!("{}", e))
println!("{:?}", i); .for_each(|i| {
Ok(()) println!("{:?}", i);
})) Ok(())
}))
.unwrap(); .unwrap();
} }

17
examples/sync_server_calling_server.rs
View File

@@ -58,9 +58,7 @@ impl DoubleServer {
impl DoubleSyncService for DoubleServer { impl DoubleSyncService for DoubleServer {
fn double(&self, x: i32) -> Result<i32, Message> { fn double(&self, x: i32) -> Result<i32, Message> {
self.client self.client.add(x, x).map_err(|e| e.to_string().into())
.add(x, x)
.map_err(|e| e.to_string().into())
} }
} }
@@ -68,12 +66,13 @@ fn main() {
let _ = env_logger::init(); let _ = env_logger::init();
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
thread::spawn(move || { thread::spawn(move || {
let handle = AddServer.listen("localhost:0".first_socket_addr(), let handle = AddServer
server::Options::default()) .listen("localhost:0".first_socket_addr(),
.unwrap(); server::Options::default())
tx.send(handle.addr()).unwrap(); .unwrap();
handle.run(); tx.send(handle.addr()).unwrap();
}); handle.run();
});
let add = rx.recv().unwrap(); let add = rx.recv().unwrap();

9
examples/throughput.rs
View File

@@ -59,7 +59,8 @@ fn bench_tarpc(target: u64) {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
thread::spawn(move || { thread::spawn(move || {
let mut reactor = reactor::Core::new().unwrap(); let mut reactor = reactor::Core::new().unwrap();
let (addr, server) = Server.listen("localhost:0".first_socket_addr(), let (addr, server) = Server
.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();
@@ -86,9 +87,9 @@ fn bench_tcp(target: u64) {
let l = net::TcpListener::bind("localhost:0").unwrap(); let l = net::TcpListener::bind("localhost:0").unwrap();
let addr = l.local_addr().unwrap(); let addr = l.local_addr().unwrap();
thread::spawn(move || { thread::spawn(move || {
let (mut stream, _) = l.accept().unwrap(); let (mut stream, _) = l.accept().unwrap();
while let Ok(_) = stream.write_all(&*BUF) {} while let Ok(_) = stream.write_all(&*BUF) {}
}); });
let mut stream = net::TcpStream::connect(&addr).unwrap(); let mut stream = net::TcpStream::connect(&addr).unwrap();
let mut buf = vec![0; CHUNK_SIZE as usize]; let mut buf = vec![0; CHUNK_SIZE as usize];
let start = time::Instant::now(); let start = time::Instant::now();

8
examples/two_clients.rs
View File

@@ -68,8 +68,8 @@ fn main() {
thread::spawn(move || { thread::spawn(move || {
let mut reactor = reactor::Core::new().unwrap(); let mut reactor = reactor::Core::new().unwrap();
let (handle, server) = Bar.listen("localhost:0".first_socket_addr(), let (handle, server) = Bar.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();
tx.send(handle).unwrap(); tx.send(handle).unwrap();
reactor.run(server).unwrap(); reactor.run(server).unwrap();
@@ -83,8 +83,8 @@ fn main() {
thread::spawn(move || { thread::spawn(move || {
let mut reactor = reactor::Core::new().unwrap(); let mut reactor = reactor::Core::new().unwrap();
let (handle, server) = Baz.listen("localhost:0".first_socket_addr(), let (handle, server) = Baz.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();
tx.send(handle).unwrap(); tx.send(handle).unwrap();
reactor.run(server).unwrap(); reactor.run(server).unwrap();

9
src/future/client.rs
View File

@@ -212,10 +212,11 @@ impl<Req, Resp, E> ClientExt for Client<Req, Resp, E>
#[cfg(feature = "tls")] #[cfg(feature = "tls")]
match tls_ctx { match tls_ctx {
Some(tls_ctx) => { Some(tls_ctx) => {
future::Either::A(tls_ctx.tls_connector future::Either::A(tls_ctx
.connect_async(&tls_ctx.domain, socket) .tls_connector
.map(StreamType::Tls) .connect_async(&tls_ctx.domain, socket)
.map_err(native_to_io)) .map(StreamType::Tls)
.map_err(native_to_io))
} }
None => future::Either::B(future::ok(StreamType::Tcp(socket))), None => future::Either::B(future::ok(StreamType::Tcp(socket))),
} }

6
src/future/server/connection.rs
View File

@@ -69,8 +69,8 @@ impl<S: NewService> NewService for TrackingNewService<S> {
fn new_service(&self) -> io::Result<Self::Instance> { fn new_service(&self) -> io::Result<Self::Instance> {
self.connection_tracker.increment(); self.connection_tracker.increment();
Ok(TrackingService { Ok(TrackingService {
service: self.new_service.new_service()?, service: self.new_service.new_service()?,
tracker: self.connection_tracker.clone(), tracker: self.connection_tracker.clone(),
}) })
} }
} }

76
src/future/server/mod.rs
View File

@@ -12,9 +12,9 @@ use std::fmt;
use std::io; use std::io;
use std::net::SocketAddr; use std::net::SocketAddr;
use stream_type::StreamType; use stream_type::StreamType;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_core::net::{Incoming, TcpListener, TcpStream}; use tokio_core::net::{Incoming, TcpListener, TcpStream};
use tokio_core::reactor; use tokio_core::reactor;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_proto::BindServer; use tokio_proto::BindServer;
use tokio_service::NewService; use tokio_service::NewService;
@@ -59,9 +59,9 @@ enum Acceptor {
struct Accept { struct Accept {
#[cfg(feature = "tls")] #[cfg(feature = "tls")]
inner: futures::Either<futures::MapErr<futures::Map<AcceptAsync<TcpStream>, inner: futures::Either<futures::MapErr<futures::Map<AcceptAsync<TcpStream>,
fn(TlsStream<TcpStream>) -> StreamType>, fn(TlsStream<TcpStream>) -> StreamType>,
fn(native_tls::Error) -> io::Error>, fn(native_tls::Error) -> io::Error>,
futures::FutureResult<StreamType, io::Error>>, futures::FutureResult<StreamType, io::Error>>,
#[cfg(not(feature = "tls"))] #[cfg(not(feature = "tls"))]
inner: futures::FutureResult<StreamType, io::Error>, inner: futures::FutureResult<StreamType, io::Error>,
} }
@@ -82,20 +82,19 @@ impl Acceptor {
Accept { Accept {
inner: match *self { inner: match *self {
Acceptor::Tls(ref tls_acceptor) => { Acceptor::Tls(ref tls_acceptor) => {
futures::Either::A(tls_acceptor.accept_async(socket) futures::Either::A(tls_acceptor
.map(StreamType::Tls as _) .accept_async(socket)
.map_err(native_to_io)) .map(StreamType::Tls as _)
.map_err(native_to_io))
} }
Acceptor::Tcp => futures::Either::B(futures::ok(StreamType::Tcp(socket))), Acceptor::Tcp => futures::Either::B(futures::ok(StreamType::Tcp(socket))),
} },
} }
} }
#[cfg(not(feature = "tls"))] #[cfg(not(feature = "tls"))]
fn accept(&self, socket: TcpStream) -> Accept { fn accept(&self, socket: TcpStream) -> Accept {
Accept { Accept { inner: futures::ok(StreamType::Tcp(socket)) }
inner: futures::ok(StreamType::Tcp(socket))
}
} }
} }
@@ -144,7 +143,7 @@ struct AcceptStream<S> {
} }
impl<S> Stream for AcceptStream<S> impl<S> Stream for AcceptStream<S>
where S: Stream<Item=(TcpStream, SocketAddr), Error = io::Error>, where S: Stream<Item = (TcpStream, SocketAddr), Error = io::Error>
{ {
type Item = <Accept as Future>::Item; type Item = <Accept as Future>::Item;
type Error = io::Error; type Error = io::Error;
@@ -167,7 +166,7 @@ impl<S> Stream for AcceptStream<S>
self.future = None; self.future = None;
Err(e) Err(e)
} }
Ok(Async::NotReady) => Ok(Async::NotReady) Ok(Async::NotReady) => Ok(Async::NotReady),
} }
} }
} }
@@ -183,9 +182,7 @@ pub struct Options {
impl Default for Options { impl Default for Options {
#[cfg(not(feature = "tls"))] #[cfg(not(feature = "tls"))]
fn default() -> Self { fn default() -> Self {
Options { Options { max_payload_size: 2 << 20 }
max_payload_size: 2 << 20,
}
} }
#[cfg(feature = "tls")] #[cfg(feature = "tls")]
@@ -221,7 +218,12 @@ impl fmt::Debug for Options {
let mut debug_struct = fmt.debug_struct("Options"); let mut debug_struct = fmt.debug_struct("Options");
#[cfg(feature = "tls")] #[cfg(feature = "tls")]
debug_struct.field("tls_acceptor", if self.tls_acceptor.is_some() { SOME } else { NONE }); debug_struct.field("tls_acceptor",
if self.tls_acceptor.is_some() {
SOME
} else {
NONE
});
debug_struct.finish() debug_struct.finish()
} }
} }
@@ -243,8 +245,11 @@ pub fn listen<S, Req, Resp, E>(new_service: S,
Resp: Serialize + 'static, Resp: Serialize + 'static,
E: Serialize + 'static E: Serialize + 'static
{ {
let (addr, shutdown, server) = listen_with( let (addr, shutdown, server) = listen_with(new_service,
new_service, addr, handle, options.max_payload_size, Acceptor::from(options))?; addr,
handle,
options.max_payload_size,
Acceptor::from(options))?;
Ok((Handle { Ok((Handle {
addr: addr, addr: addr,
shutdown: shutdown, shutdown: shutdown,
@@ -299,7 +304,8 @@ fn listener(addr: &SocketAddr, handle: &reactor::Handle) -> io::Result<TcpListen
}?; }?;
configure_tcp(&builder)?; configure_tcp(&builder)?;
builder.reuse_address(true)?; builder.reuse_address(true)?;
builder.bind(addr)? builder
.bind(addr)?
.listen(PENDING_CONNECTION_BACKLOG) .listen(PENDING_CONNECTION_BACKLOG)
.and_then(|l| TcpListener::from_listener(l, addr, handle)) .and_then(|l| TcpListener::from_listener(l, addr, handle))
} }
@@ -325,15 +331,15 @@ struct BindStream<S, St> {
impl<S, St> fmt::Debug for BindStream<S, St> impl<S, St> fmt::Debug for BindStream<S, St>
where S: fmt::Debug, where S: fmt::Debug,
St: fmt::Debug, St: fmt::Debug
{ {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> { fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
const HANDLE: &'static &'static str = &"Handle { .. }"; const HANDLE: &'static &'static str = &"Handle { .. }";
f.debug_struct("BindStream") f.debug_struct("BindStream")
.field("handle", HANDLE) .field("handle", HANDLE)
.field("new_service", &self.new_service) .field("new_service", &self.new_service)
.field("stream", &self.stream) .field("stream", &self.stream)
.finish() .finish()
} }
} }
@@ -345,14 +351,15 @@ impl<S, Req, Resp, E, I, St> BindStream<S, St>
Resp: Serialize + 'static, Resp: Serialize + 'static,
E: Serialize + 'static, E: Serialize + 'static,
I: AsyncRead + AsyncWrite + 'static, I: AsyncRead + AsyncWrite + 'static,
St: Stream<Item=I, Error=io::Error>, St: Stream<Item = I, Error = io::Error>
{ {
fn bind_each(&mut self) -> Poll<(), io::Error> { fn bind_each(&mut self) -> Poll<(), io::Error> {
loop { loop {
match try!(self.stream.poll()) { match try!(self.stream.poll()) {
Async::Ready(Some(socket)) => { Async::Ready(Some(socket)) => {
Proto::new(self.max_payload_size) Proto::new(self.max_payload_size).bind_server(&self.handle,
.bind_server(&self.handle, socket, self.new_service.new_service()?); socket,
self.new_service.new_service()?);
} }
Async::Ready(None) => return Ok(Async::Ready(())), Async::Ready(None) => return Ok(Async::Ready(())),
Async::NotReady => return Ok(Async::NotReady), Async::NotReady => return Ok(Async::NotReady),
@@ -369,7 +376,7 @@ impl<S, Req, Resp, E, I, St> Future for BindStream<S, St>
Resp: Serialize + 'static, Resp: Serialize + 'static,
E: Serialize + 'static, E: Serialize + 'static,
I: AsyncRead + AsyncWrite + 'static, I: AsyncRead + AsyncWrite + 'static,
St: Stream<Item=I, Error=io::Error>, St: Stream<Item = I, Error = io::Error>
{ {
type Item = (); type Item = ();
type Error = (); type Error = ();
@@ -396,8 +403,7 @@ pub struct Listen<S, Req, Resp, E>
Resp: Serialize + 'static, Resp: Serialize + 'static,
E: Serialize + 'static E: Serialize + 'static
{ {
inner: AlwaysOkUnit<futures::Select<BindStream<S, AcceptStream<Incoming>>, inner: AlwaysOkUnit<futures::Select<BindStream<S, AcceptStream<Incoming>>, shutdown::Watcher>>,
shutdown::Watcher>>,
} }
impl<S, Req, Resp, E> Future for Listen<S, Req, Resp, E> impl<S, Req, Resp, E> Future for Listen<S, Req, Resp, E>
@@ -422,7 +428,7 @@ impl<S, Req, Resp, E> fmt::Debug for Listen<S, Req, Resp, E>
Error = io::Error> + 'static, Error = io::Error> + 'static,
Req: Deserialize + 'static, Req: Deserialize + 'static,
Resp: Serialize + 'static, Resp: Serialize + 'static,
E: Serialize + 'static, E: Serialize + 'static
{ {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> { fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
f.debug_struct("Listen").finish() f.debug_struct("Listen").finish()
@@ -433,16 +439,16 @@ impl<S, Req, Resp, E> fmt::Debug for Listen<S, Req, Resp, E>
struct AlwaysOkUnit<F>(F); struct AlwaysOkUnit<F>(F);
impl<F> Future for AlwaysOkUnit<F> impl<F> Future for AlwaysOkUnit<F>
where F: Future, where F: Future
{ {
type Item = (); type Item = ();
type Error = (); type Error = ();
fn poll(&mut self) -> Poll<(), ()> { fn poll(&mut self) -> Poll<(), ()> {
match self.0.poll() { match self.0.poll() {
Ok(Async::Ready(_)) | Err(_) => Ok(Async::Ready(())), Ok(Async::Ready(_)) |
Err(_) => Ok(Async::Ready(())),
Ok(Async::NotReady) => Ok(Async::NotReady), Ok(Async::NotReady) => Ok(Async::NotReady),
} }
} }
} }

51
src/future/server/shutdown.rs
View File

@@ -1,9 +1,10 @@
use super::{AlwaysOkUnit, connection};
use futures::{Async, Future, Poll, Stream, future as futures, stream}; use futures::{Async, Future, Poll, Stream, future as futures, stream};
use futures::sync::{mpsc, oneshot}; use futures::sync::{mpsc, oneshot};
use futures::unsync; use futures::unsync;
use super::{AlwaysOkUnit, connection};
/// A hook to shut down a running server. /// A hook to shut down a running server.
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct Shutdown { pub struct Shutdown {
@@ -13,8 +14,7 @@ pub struct Shutdown {
/// A future that resolves when server shutdown completes. /// A future that resolves when server shutdown completes.
#[derive(Debug)] #[derive(Debug)]
pub struct ShutdownFuture { pub struct ShutdownFuture {
inner: futures::Either<futures::FutureResult<(), ()>, inner: futures::Either<futures::FutureResult<(), ()>, AlwaysOkUnit<oneshot::Receiver<()>>>,
AlwaysOkUnit<oneshot::Receiver<()>>>,
} }
impl Future for ShutdownFuture { impl Future for ShutdownFuture {
@@ -63,13 +63,13 @@ impl Watcher {
(connection_tx, (connection_tx,
Shutdown { tx: shutdown_tx }, Shutdown { tx: shutdown_tx },
Watcher { Watcher {
shutdown_rx: shutdown_rx.take(1), shutdown_rx: shutdown_rx.take(1),
connections: connections, connections: connections,
queued_error: None, queued_error: None,
shutdown: None, shutdown: None,
done: false, done: false,
num_connections: 0, num_connections: 0,
}) })
} }
fn process_connection(&mut self, action: connection::Action) { fn process_connection(&mut self, action: connection::Action) {
@@ -81,28 +81,28 @@ impl Watcher {
fn poll_shutdown_requests(&mut self) -> Poll<Option<()>, ()> { fn poll_shutdown_requests(&mut self) -> Poll<Option<()>, ()> {
Ok(Async::Ready(match try_ready!(self.shutdown_rx.poll()) { Ok(Async::Ready(match try_ready!(self.shutdown_rx.poll()) {
Some(tx) => { Some(tx) => {
debug!("Received shutdown request."); debug!("Received shutdown request.");
self.shutdown = Some(tx); self.shutdown = Some(tx);
Some(()) Some(())
} }
None => None, None => None,
})) }))
} }
fn poll_connections(&mut self) -> Poll<Option<()>, ()> { fn poll_connections(&mut self) -> Poll<Option<()>, ()> {
Ok(Async::Ready(match try_ready!(self.connections.poll()) { Ok(Async::Ready(match try_ready!(self.connections.poll()) {
Some(action) => { Some(action) => {
self.process_connection(action); self.process_connection(action);
Some(()) Some(())
} }
None => None, None => None,
})) }))
} }
fn poll_shutdown_requests_and_connections(&mut self) -> Poll<Option<()>, ()> { fn poll_shutdown_requests_and_connections(&mut self) -> Poll<Option<()>, ()> {
if let Some(e) = self.queued_error.take() { if let Some(e) = self.queued_error.take() {
return Err(e) return Err(e);
} }
match try!(self.poll_shutdown_requests()) { match try!(self.poll_shutdown_requests()) {
@@ -178,4 +178,3 @@ impl Future for Watcher {
} }
} }
} }

10
src/lib.rs
View File

@@ -189,12 +189,12 @@ lazy_static! {
fn spawn_core() -> reactor::Remote { fn spawn_core() -> reactor::Remote {
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
thread::spawn(move || { thread::spawn(move || {
let mut core = reactor::Core::new().unwrap(); let mut core = reactor::Core::new().unwrap();
tx.send(core.handle().remote().clone()).unwrap(); tx.send(core.handle().remote().clone()).unwrap();
// Run forever // Run forever
core.run(futures::empty::<(), !>()).unwrap(); core.run(futures::empty::<(), !>()).unwrap();
}); });
rx.recv().unwrap() rx.recv().unwrap()
} }

55
src/macros.rs
View File

@@ -1072,9 +1072,9 @@ mod functional_test {
#[test] #[test]
fn other_service() { fn other_service() {
let _ = env_logger::init(); let _ = env_logger::init();
let (_, client, _) = let (_, client, _) = unwrap!(
unwrap!(start_server_with_sync_client::<super::other_service::SyncClient, start_server_with_sync_client::<super::other_service::SyncClient, Server>(Server)
Server>(Server)); );
match client.foo().err().expect("failed unwrap") { match client.foo().err().expect("failed unwrap") {
::Error::RequestDeserialize(_) => {} // good ::Error::RequestDeserialize(_) => {} // good
bad => panic!("Expected Error::RequestDeserialize but got {}", bad), bad => panic!("Expected Error::RequestDeserialize but got {}", bad),
@@ -1111,7 +1111,9 @@ mod functional_test {
#[test] #[test]
fn bad_serialize() { fn bad_serialize() {
let handle = ().listen("localhost:0", server::Options::default()).unwrap(); let handle = ()
.listen("localhost:0", server::Options::default())
.unwrap();
let client = SyncClient::connect(handle.addr(), client::Options::default()).unwrap(); let client = SyncClient::connect(handle.addr(), client::Options::default()).unwrap();
client.bad(Bad).err().unwrap(); client.bad(Bad).err().unwrap();
} }
@@ -1209,12 +1211,15 @@ mod functional_test {
let _ = env_logger::init(); let _ = env_logger::init();
let reactor = reactor::Core::new().unwrap(); let reactor = reactor::Core::new().unwrap();
let handle = Server.listen("localhost:0".first_socket_addr(), let handle = Server
.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap() .unwrap()
.0; .0;
Server.listen(handle.addr(), &reactor.handle(), server::Options::default()).unwrap(); Server
.listen(handle.addr(), &reactor.handle(), server::Options::default())
.unwrap();
} }
#[test] #[test]
@@ -1226,20 +1231,23 @@ mod functional_test {
let _ = env_logger::init(); let _ = env_logger::init();
let mut reactor = reactor::Core::new().unwrap(); let mut reactor = reactor::Core::new().unwrap();
let (handle, server) = Server.listen("localhost:0".first_socket_addr(), let (handle, server) = Server
.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();
reactor.handle().spawn(server); reactor.handle().spawn(server);
let client = FutureClient::connect(handle.addr(), let client = FutureClient::connect(handle.addr(),
client::Options::default().handle(reactor.handle())); client::Options::default()
.handle(reactor.handle()));
let client = unwrap!(reactor.run(client)); let client = unwrap!(reactor.run(client));
assert_eq!(reactor.run(client.add(1, 2)).unwrap(), 3); assert_eq!(reactor.run(client.add(1, 2)).unwrap(), 3);
drop(client); drop(client);
let client = FutureClient::connect(handle.addr(), let client = FutureClient::connect(handle.addr(),
client::Options::default().handle(reactor.handle())); client::Options::default()
.handle(reactor.handle()));
let client = unwrap!(reactor.run(client)); let client = unwrap!(reactor.run(client));
assert_eq!(reactor.run(client.add(1, 2)).unwrap(), 3); assert_eq!(reactor.run(client.add(1, 2)).unwrap(), 3);
} }
@@ -1259,13 +1267,16 @@ mod functional_test {
assert_eq!("Hey, Tim.", assert_eq!("Hey, Tim.",
reactor.run(client.hey("Tim".to_string())).unwrap()); reactor.run(client.hey("Tim".to_string())).unwrap());
let (handle, server) = Server.listen("localhost:0".first_socket_addr(), let (handle, server) = Server
.listen("localhost:0".first_socket_addr(),
&reactor.handle(), &reactor.handle(),
server::Options::default()) server::Options::default())
.unwrap(); .unwrap();
reactor.handle().spawn(server); reactor.handle().spawn(server);
let options = client::Options::default().handle(reactor.handle()); let options = client::Options::default().handle(reactor.handle());
let client = reactor.run(FutureClient::connect(handle.addr(), options)).unwrap(); let client = reactor
.run(FutureClient::connect(handle.addr(), options))
.unwrap();
assert_eq!(3, reactor.run(client.add(1, 2)).unwrap()); assert_eq!(3, reactor.run(client.add(1, 2)).unwrap());
assert_eq!("Hey, Tim.", assert_eq!("Hey, Tim.",
reactor.run(client.hey("Tim".to_string())).unwrap()); reactor.run(client.hey("Tim".to_string())).unwrap());
@@ -1298,16 +1309,18 @@ mod functional_test {
let (_, mut reactor, client) = let (_, mut reactor, client) =
start_err_server_with_async_client::<FutureClient, ErrorServer>(ErrorServer).unwrap(); start_err_server_with_async_client::<FutureClient, ErrorServer>(ErrorServer).unwrap();
reactor.run(client.bar() reactor
.then(move |result| { .run(client
match result.err().unwrap() { .bar()
::Error::App(e) => { .then(move |result| {
assert_eq!(e.description(), "lol jk"); match result.err().unwrap() {
Ok::<_, ()>(()) ::Error::App(e) => {
} // good assert_eq!(e.description(), "lol jk");
bad => panic!("Expected Error::App but got {:?}", bad), Ok::<_, ()>(())
} } // good
})) bad => panic!("Expected Error::App but got {:?}", bad),
}
}))
.unwrap(); .unwrap();
} }

27
src/protocol.rs
View File

@@ -3,11 +3,11 @@
// Licensed under the MIT License, <LICENSE or http://opensource.org/licenses/MIT>. // Licensed under the MIT License, <LICENSE or http://opensource.org/licenses/MIT>.
// This file may not be copied, modified, or distributed except according to those terms. // This file may not be copied, modified, or distributed except according to those terms.
use serde;
use bincode::{self, Infinite}; use bincode::{self, Infinite};
use byteorder::{BigEndian, ReadBytesExt}; use byteorder::{BigEndian, ReadBytesExt};
use bytes::BytesMut; use bytes::BytesMut;
use bytes::buf::BufMut; use bytes::buf::BufMut;
use serde;
use std::io::{self, Cursor}; use std::io::{self, Cursor};
use std::marker::PhantomData; use std::marker::PhantomData;
use std::mem; use std::mem;
@@ -43,10 +43,12 @@ impl<Encode, Decode> Codec<Encode, Decode> {
fn too_big(payload_size: u64, max_payload_size: u64) -> io::Error { fn too_big(payload_size: u64, max_payload_size: u64) -> io::Error {
warn!("Not sending too-big packet of size {} (max is {})", warn!("Not sending too-big packet of size {} (max is {})",
payload_size, max_payload_size); payload_size,
max_payload_size);
io::Error::new(io::ErrorKind::InvalidData, io::Error::new(io::ErrorKind::InvalidData,
format!("Maximum payload size is {} bytes but got a payload of {}", format!("Maximum payload size is {} bytes but got a payload of {}",
max_payload_size, payload_size)) max_payload_size,
payload_size))
} }
impl<Encode, Decode> Encoder for Codec<Encode, Decode> impl<Encode, Decode> Encoder for Codec<Encode, Decode>
@@ -66,9 +68,7 @@ impl<Encode, Decode> Encoder for Codec<Encode, Decode>
buf.put_u64::<BigEndian>(id); buf.put_u64::<BigEndian>(id);
trace!("Encoded request id = {} as {:?}", id, buf); trace!("Encoded request id = {} as {:?}", id, buf);
buf.put_u64::<BigEndian>(payload_size); buf.put_u64::<BigEndian>(payload_size);
bincode::serialize_into(&mut buf.writer(), bincode::serialize_into(&mut buf.writer(), &message, Infinite)
&message,
Infinite)
.map_err(|serialize_err| io::Error::new(io::ErrorKind::Other, serialize_err))?; .map_err(|serialize_err| io::Error::new(io::ErrorKind::Other, serialize_err))?;
trace!("Encoded buffer: {:?}", buf); trace!("Encoded buffer: {:?}", buf);
Ok(()) Ok(())
@@ -121,8 +121,7 @@ impl<Encode, Decode> Decoder for Codec<Encode, Decode>
} }
Payload { id, len } => { Payload { id, len } => {
let payload = buf.split_to(len as usize); let payload = buf.split_to(len as usize);
let result = bincode::deserialize_from(&mut Cursor::new(payload), let result = bincode::deserialize_from(&mut Cursor::new(payload), Infinite);
Infinite);
// Reset the state machine because, either way, we're done processing this // Reset the state machine because, either way, we're done processing this
// message. // message.
self.state = Id; self.state = Id;
@@ -146,7 +145,7 @@ impl<Encode, Decode> Proto<Encode, Decode> {
pub fn new(max_payload_size: u64) -> Self { pub fn new(max_payload_size: u64) -> Self {
Proto { Proto {
max_payload_size: max_payload_size, max_payload_size: max_payload_size,
_phantom_data: PhantomData _phantom_data: PhantomData,
} }
} }
} }
@@ -190,8 +189,8 @@ fn serialize() {
for _ in 0..2 { for _ in 0..2 {
let mut codec: Codec<(char, char, char), (char, char, char)> = Codec::new(2_000_000); let mut codec: Codec<(char, char, char), (char, char, char)> = Codec::new(2_000_000);
codec.encode(MSG, &mut buf).unwrap(); codec.encode(MSG, &mut buf).unwrap();
let actual: Result<Option<(u64, Result<(char, char, char), bincode::Error>)>, io::Error> = let actual: Result<Option<(u64, Result<(char, char, char), bincode::Error>)>,
codec.decode(&mut buf); io::Error> = codec.decode(&mut buf);
match actual { match actual {
Ok(Some((id, ref v))) if id == MSG.0 && *v.as_ref().unwrap() == MSG.1 => {} Ok(Some((id, ref v))) if id == MSG.0 && *v.as_ref().unwrap() == MSG.1 => {}
@@ -207,7 +206,11 @@ fn deserialize_big() {
let mut codec: Codec<Vec<u8>, Vec<u8>> = Codec::new(24); let mut codec: Codec<Vec<u8>, Vec<u8>> = Codec::new(24);
let mut buf = BytesMut::with_capacity(40); let mut buf = BytesMut::with_capacity(40);
assert_eq!(codec.encode((0, vec![0; 24]), &mut buf).err().unwrap().kind(), assert_eq!(codec
.encode((0, vec![0; 24]), &mut buf)
.err()
.unwrap()
.kind(),
io::ErrorKind::InvalidData); io::ErrorKind::InvalidData);
// Header // Header

81
src/sync/client.rs
View File

@@ -8,12 +8,12 @@ use std::io;
use std::net::{SocketAddr, ToSocketAddrs}; use std::net::{SocketAddr, ToSocketAddrs};
use std::sync::mpsc; use std::sync::mpsc;
use std::thread; use std::thread;
#[cfg(feature = "tls")]
use tls::client::Context;
use tokio_core::reactor; use tokio_core::reactor;
use tokio_proto::util::client_proxy::{ClientProxy, Receiver, pair}; use tokio_proto::util::client_proxy::{ClientProxy, Receiver, pair};
use tokio_service::Service; use tokio_service::Service;
use util::FirstSocketAddr; use util::FirstSocketAddr;
#[cfg(feature = "tls")]
use tls::client::Context;
#[doc(hidden)] #[doc(hidden)]
pub struct Client<Req, Resp, E> { pub struct Client<Req, Resp, E> {
@@ -29,9 +29,7 @@ impl<Req, Resp, E> Clone for Client<Req, Resp, E> {
impl<Req, Resp, E> fmt::Debug for Client<Req, Resp, E> { impl<Req, Resp, E> fmt::Debug for Client<Req, Resp, E> {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> { fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
const PROXY: &'static &'static str = &"ClientProxy { .. }"; const PROXY: &'static &'static str = &"ClientProxy { .. }";
f.debug_struct("Client") f.debug_struct("Client").field("proxy", PROXY).finish()
.field("proxy", PROXY)
.finish()
} }
} }
@@ -47,7 +45,6 @@ impl<Req, Resp, E> Client<Req, Resp, E>
// oneshot send. // oneshot send.
self.proxy.call(request).wait() self.proxy.call(request).wait()
} }
} }
/// Additional options to configure how the client connects and operates. /// Additional options to configure how the client connects and operates.
@@ -61,9 +58,7 @@ pub struct Options {
impl Default for Options { impl Default for Options {
#[cfg(not(feature = "tls"))] #[cfg(not(feature = "tls"))]
fn default() -> Self { fn default() -> Self {
Options { Options { max_payload_size: 2_000_000 }
max_payload_size: 2_000_000,
}
} }
#[cfg(feature = "tls")] #[cfg(feature = "tls")]
@@ -137,15 +132,13 @@ impl<Req, Resp, E> ClientExt for Client<Req, Resp, E>
{ {
let addr = addr.try_first_socket_addr()?; let addr = addr.try_first_socket_addr()?;
let (connect_tx, connect_rx) = mpsc::channel(); let (connect_tx, connect_rx) = mpsc::channel();
thread::spawn(move || { thread::spawn(move || match RequestHandler::connect(addr, options) {
match RequestHandler::connect(addr, options) { Ok((proxy, mut handler)) => {
Ok((proxy, mut handler)) => { connect_tx.send(Ok(proxy)).unwrap();
connect_tx.send(Ok(proxy)).unwrap(); handler.handle_requests();
handler.handle_requests(); }
} Err(e) => connect_tx.send(Err(e)).unwrap(),
Err(e) => connect_tx.send(Err(e)).unwrap(), });
}
});
Ok(connect_rx.recv().unwrap()?) Ok(connect_rx.recv().unwrap()?)
} }
} }
@@ -166,14 +159,17 @@ impl<Req, Resp, E> RequestHandler<Req, Resp, E, FutureClient<Req, Resp, E>>
{ {
/// Creates a new `RequestHandler` by connecting a `FutureClient` to the given address /// Creates a new `RequestHandler` by connecting a `FutureClient` to the given address
/// using the given options. /// using the given options.
fn connect(addr: SocketAddr, options: Options) fn connect(addr: SocketAddr, options: Options) -> io::Result<(Client<Req, Resp, E>, Self)> {
-> io::Result<(Client<Req, Resp, E>, Self)>
{
let mut reactor = reactor::Core::new()?; let mut reactor = reactor::Core::new()?;
let options = (reactor.handle(), options).into(); let options = (reactor.handle(), options).into();
let client = reactor.run(FutureClient::connect(addr, options))?; let client = reactor.run(FutureClient::connect(addr, options))?;
let (proxy, requests) = pair(); let (proxy, requests) = pair();
Ok((Client { proxy }, RequestHandler { reactor, client, requests })) Ok((Client { proxy },
RequestHandler {
reactor,
client,
requests,
}))
} }
} }
@@ -182,21 +178,26 @@ impl<Req, Resp, E, S> RequestHandler<Req, Resp, E, S>
Resp: Deserialize + 'static, Resp: Deserialize + 'static,
E: Deserialize + 'static, E: Deserialize + 'static,
S: Service<Request = Req, Response = Resp, Error = ::Error<E>>, S: Service<Request = Req, Response = Resp, Error = ::Error<E>>,
S::Future: 'static, S::Future: 'static
{ {
fn handle_requests(&mut self) { fn handle_requests(&mut self) {
let RequestHandler { ref mut reactor, ref mut requests, ref mut client } = *self; let RequestHandler {
ref mut reactor,
ref mut requests,
ref mut client,
} = *self;
let handle = reactor.handle(); let handle = reactor.handle();
let requests = requests let requests =
.map(|result| { requests
match result { .map(|result| {
Ok(req) => req, match result {
// The ClientProxy never sends Err currently Ok(req) => req,
Err(e) => panic!("Unimplemented error handling in RequestHandler: {}", e), // The ClientProxy never sends Err currently
} Err(e) => panic!("Unimplemented error handling in RequestHandler: {}", e),
}) }
.for_each(|(request, response_tx)| { })
let request = client.call(request) .for_each(|(request, response_tx)| {
let request = client.call(request)
.then(move |response| { .then(move |response| {
// Safe to unwrap because clients always block on the response future. // Safe to unwrap because clients always block on the response future.
response_tx.send(response) response_tx.send(response)
@@ -204,9 +205,9 @@ impl<Req, Resp, E, S> RequestHandler<Req, Resp, E, S>
.expect("Client should block on response"); .expect("Client should block on response");
Ok(()) Ok(())
}); });
handle.spawn(request); handle.spawn(request);
Ok(()) Ok(())
}); });
reactor.run(requests).unwrap(); reactor.run(requests).unwrap();
} }
} }
@@ -230,7 +231,11 @@ fn handle_requests() {
let (request, requests) = ::futures::sync::mpsc::unbounded(); let (request, requests) = ::futures::sync::mpsc::unbounded();
let reactor = reactor::Core::new().unwrap(); let reactor = reactor::Core::new().unwrap();
let client = Client; let client = Client;
let mut request_handler = RequestHandler { reactor, client, requests }; let mut request_handler = RequestHandler {
reactor,
client,
requests,
};
// Test that `handle_requests` returns when all request senders are dropped. // Test that `handle_requests` returns when all request senders are dropped.
drop(request); drop(request);
request_handler.handle_requests(); request_handler.handle_requests();

85
src/sync/server.rs
View File

@@ -2,17 +2,17 @@ use {bincode, future, num_cpus};
use future::server::{Response, Shutdown}; use future::server::{Response, Shutdown};
use futures::{Future, future as futures}; use futures::{Future, future as futures};
use futures::sync::oneshot; use futures::sync::oneshot;
#[cfg(feature = "tls")]
use native_tls_inner::TlsAcceptor;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::io;
use std::fmt; use std::fmt;
use std::io;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::time::Duration; use std::time::Duration;
use std::usize; use std::usize;
use thread_pool::{self, Sender, Task, ThreadPool}; use thread_pool::{self, Sender, Task, ThreadPool};
use tokio_core::reactor; use tokio_core::reactor;
use tokio_service::{NewService, Service}; use tokio_service::{NewService, Service};
#[cfg(feature = "tls")]
use native_tls_inner::TlsAcceptor;
/// Additional options to configure how the server operates. /// Additional options to configure how the server operates.
#[derive(Debug)] #[derive(Debug)]
@@ -91,16 +91,19 @@ impl fmt::Debug for Handle {
const CORE: &'static &'static str = &"Core { .. }"; const CORE: &'static &'static str = &"Core { .. }";
const SERVER: &'static &'static str = &"Box<Future<Item = (), Error = ()>>"; const SERVER: &'static &'static str = &"Box<Future<Item = (), Error = ()>>";
f.debug_struct("Handle").field("reactor", CORE) f.debug_struct("Handle")
.field("handle", &self.handle) .field("reactor", CORE)
.field("server", SERVER) .field("handle", &self.handle)
.finish() .field("server", SERVER)
.finish()
} }
} }
#[doc(hidden)] #[doc(hidden)]
pub fn listen<S, Req, Resp, E>(new_service: S, addr: SocketAddr, options: Options) pub fn listen<S, Req, Resp, E>(new_service: S,
-> io::Result<Handle> addr: SocketAddr,
options: Options)
-> io::Result<Handle>
where S: NewService<Request = Result<Req, bincode::Error>, where S: NewService<Request = Result<Req, bincode::Error>,
Response = Response<Resp, E>, Response = Response<Resp, E>,
Error = io::Error> + 'static, Error = io::Error> + 'static,
@@ -117,27 +120,33 @@ pub fn listen<S, Req, Resp, E>(new_service: S, addr: SocketAddr, options: Option
future::server::listen(new_service, addr, &reactor.handle(), options.opts)?; future::server::listen(new_service, addr, &reactor.handle(), options.opts)?;
let server = Box::new(server); let server = Box::new(server);
Ok(Handle { Ok(Handle {
reactor: reactor, reactor: reactor,
handle: handle, handle: handle,
server: server, server: server,
}) })
} }
/// A service that uses a thread pool. /// A service that uses a thread pool.
struct NewThreadService<S> where S: NewService { struct NewThreadService<S>
where S: NewService
{
new_service: S, new_service: S,
sender: Sender<ServiceTask<<S::Instance as Service>::Future>>, sender: Sender<ServiceTask<<S::Instance as Service>::Future>>,
_pool: ThreadPool<ServiceTask<<S::Instance as Service>::Future>>, _pool: ThreadPool<ServiceTask<<S::Instance as Service>::Future>>,
} }
/// A service that runs by executing request handlers in a thread pool. /// A service that runs by executing request handlers in a thread pool.
struct ThreadService<S> where S: Service { struct ThreadService<S>
where S: Service
{
service: S, service: S,
sender: Sender<ServiceTask<S::Future>>, sender: Sender<ServiceTask<S::Future>>,
} }
/// A task that handles a single request. /// A task that handles a single request.
struct ServiceTask<F> where F: Future { struct ServiceTask<F>
where F: Future
{
future: F, future: F,
tx: oneshot::Sender<Result<F::Item, F::Error>>, tx: oneshot::Sender<Result<F::Item, F::Error>>,
} }
@@ -146,12 +155,16 @@ impl<S> NewThreadService<S>
where S: NewService, where S: NewService,
<S::Instance as Service>::Future: Send + 'static, <S::Instance as Service>::Future: Send + 'static,
S::Response: Send, S::Response: Send,
S::Error: Send, S::Error: Send
{ {
/// Create a NewThreadService by wrapping another service. /// Create a NewThreadService by wrapping another service.
fn new(new_service: S, pool: thread_pool::Builder) -> Self { fn new(new_service: S, pool: thread_pool::Builder) -> Self {
let (sender, _pool) = pool.build(); let (sender, _pool) = pool.build();
NewThreadService { new_service, sender, _pool } NewThreadService {
new_service,
sender,
_pool,
}
} }
} }
@@ -159,7 +172,7 @@ impl<S> NewService for NewThreadService<S>
where S: NewService, where S: NewService,
<S::Instance as Service>::Future: Send + 'static, <S::Instance as Service>::Future: Send + 'static,
S::Response: Send, S::Response: Send,
S::Error: Send, S::Error: Send
{ {
type Request = S::Request; type Request = S::Request;
type Response = S::Response; type Response = S::Response;
@@ -168,16 +181,16 @@ impl<S> NewService for NewThreadService<S>
fn new_service(&self) -> io::Result<Self::Instance> { fn new_service(&self) -> io::Result<Self::Instance> {
Ok(ThreadService { Ok(ThreadService {
service: self.new_service.new_service()?, service: self.new_service.new_service()?,
sender: self.sender.clone(), sender: self.sender.clone(),
}) })
} }
} }
impl<F> Task for ServiceTask<F> impl<F> Task for ServiceTask<F>
where F: Future + Send + 'static, where F: Future + Send + 'static,
F::Item: Send, F::Item: Send,
F::Error: Send, F::Error: Send
{ {
fn run(self) { fn run(self) {
// Don't care if sending fails. It just means the request is no longer // Don't care if sending fails. It just means the request is no longer
@@ -190,25 +203,26 @@ impl<S> Service for ThreadService<S>
where S: Service, where S: Service,
S::Future: Send + 'static, S::Future: Send + 'static,
S::Response: Send, S::Response: Send,
S::Error: Send, S::Error: Send
{ {
type Request = S::Request; type Request = S::Request;
type Response = S::Response; type Response = S::Response;
type Error = S::Error; type Error = S::Error;
type Future = type Future = futures::AndThen<futures::MapErr<oneshot::Receiver<Result<Self::Response,
futures::AndThen< Self::Error>>,
futures::MapErr< fn(oneshot::Canceled) -> Self::Error>,
oneshot::Receiver<Result<Self::Response, Self::Error>>, Result<Self::Response, Self::Error>,
fn(oneshot::Canceled) -> Self::Error>, fn(Result<Self::Response, Self::Error>)
Result<Self::Response, Self::Error>, -> Result<Self::Response, Self::Error>>;
fn(Result<Self::Response, Self::Error>) -> Result<Self::Response, Self::Error>>;
fn call(&self, request: Self::Request) -> Self::Future { fn call(&self, request: Self::Request) -> Self::Future {
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
self.sender.send(ServiceTask { self.sender
future: self.service.call(request), .send(ServiceTask {
tx: tx, future: self.service.call(request),
}).unwrap(); tx: tx,
})
.unwrap();
rx.map_err(unreachable as _).and_then(ident) rx.map_err(unreachable as _).and_then(ident)
} }
} }
@@ -222,4 +236,3 @@ fn unreachable<T, U>(t: T) -> U
fn ident<T>(t: T) -> T { fn ident<T>(t: T) -> T {
t t
} }

13
src/tls.rs
View File

@@ -19,9 +19,9 @@ pub mod client {
/// validation. /// validation.
pub fn new<S: Into<String>>(domain: S) -> Result<Self, Error> { pub fn new<S: Into<String>>(domain: S) -> Result<Self, Error> {
Ok(Context { Ok(Context {
domain: domain.into(), domain: domain.into(),
tls_connector: TlsConnector::builder()?.build()?, tls_connector: TlsConnector::builder()?.build()?,
}) })
} }
/// Construct a new `Context` using the provided domain and `TlsConnector` /// Construct a new `Context` using the provided domain and `TlsConnector`
@@ -41,11 +41,10 @@ pub mod client {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> { fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
const TLS_CONNECTOR: &'static &'static str = &"TlsConnector { .. }"; const TLS_CONNECTOR: &'static &'static str = &"TlsConnector { .. }";
f.debug_struct("Context") f.debug_struct("Context")
.field("domain", &self.domain) .field("domain", &self.domain)
.field("tls_connector", TLS_CONNECTOR) .field("tls_connector", TLS_CONNECTOR)
.finish() .finish()
} }
} }
} }

8
src/util.rs
View File

@@ -122,9 +122,7 @@ pub fn lazy<F, A, R>(f: F, args: A) -> Lazy<F, A, R>
where F: FnOnce(A) -> R, where F: FnOnce(A) -> R,
R: IntoFuture R: IntoFuture
{ {
Lazy { Lazy { inner: _Lazy::First(f, args) }
inner: _Lazy::First(f, args),
}
} }
/// A future which defers creation of the actual future until a callback is /// A future which defers creation of the actual future until a callback is
@@ -146,7 +144,7 @@ enum _Lazy<F, A, R> {
impl<F, A, R> Lazy<F, A, R> impl<F, A, R> Lazy<F, A, R>
where F: FnOnce(A) -> R, where F: FnOnce(A) -> R,
R: IntoFuture, R: IntoFuture
{ {
fn get(&mut self) -> &mut R::Future { fn get(&mut self) -> &mut R::Future {
match self.inner { match self.inner {
@@ -167,7 +165,7 @@ impl<F, A, R> Lazy<F, A, R>
impl<F, A, R> Future for Lazy<F, A, R> impl<F, A, R> Future for Lazy<F, A, R>
where F: FnOnce(A) -> R, where F: FnOnce(A) -> R,
R: IntoFuture, R: IntoFuture
{ {
type Item = R::Item; type Item = R::Item;
type Error = R::Error; type Error = R::Error;