Remove the Send bound from FutureService (#96)

* Make a reactor handle mandatory for server.

This removes the Send bound from FutureService. The Send bound
is still required for SyncService, since clones are sent to
new threads for each request. (This is more fodder for the argument
that there should be a distinct Options struct for each combination of
async/sync and client/server.)

This commit also makes FutureService::listen return an io::Result
rather than a Future; the future was never really necessary and
had the unintended consequence of making SyncService::listen
deadlock when the options specified a handle (because that means
the reactor driving the service lives on the same thread that
SyncService is waiting on).

`SyncClient` is no longer `Clone` because it needs to create
a new `reactor::Core` when cloning. Tokio Clients are `Clone` but
they don't allow moving the cloned client onto a new reactor.

* Change pubsub to use Rc<Refcell<>> instead of Arc<Mutex<>>.

This is possible since services no longer need to be Send.

* Remove some unnecessary unstable features.

There 3 remaining unstable features. The hardest to remove is plugin, because
we rely on compiler plugins to rewrite types from snake case to camel. It's
possible this can be removed before the proc macros rewrite lands if
impl Trait is extended to work with traits.

* Clean up example

* Sync servers now spawn a reactor on a thread. It's decided that
   sync users should not have to know about tokio at all.

* Don't allow specifying a reactor::Core on client options.

* Fail fast in server::listen if local_addr() returns Err.

examples/concurrency.rs

@@ -25,7 +25,7 @@ use std::sync::{Arc, mpsc};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Duration, Instant};
 use tarpc::{client, server};
-use tarpc::client::future::Connect;
+use tarpc::client::future::ClientExt;
 use tarpc::util::{FirstSocketAddr, Never};
 use tokio_core::reactor;
 
@@ -167,10 +167,11 @@ fn main() {
         .map(Result::unwrap)
         .unwrap_or(4);
 
+    let mut reactor = reactor::Core::new().unwrap();
     let addr = Server::new()
         .listen("localhost:0".first_socket_addr(),
+                &reactor.handle(),
                 server::Options::default())
-        .wait()
         .unwrap();
     info!("Server listening on {}.", addr);
 
@@ -190,8 +191,5 @@ fn main() {
 
     info!("Starting...");
 
-    // The driver of the main future.
-    let mut core = reactor::Core::new().unwrap();
-
-    core.run(run).unwrap();
+    reactor.run(run).unwrap();
 }

examples/pubsub.rs

@@ -10,20 +10,21 @@ extern crate env_logger;
 extern crate futures;
 #[macro_use]
 extern crate tarpc;
-extern crate tokio_proto as tokio;
+extern crate tokio_core;
 
-use futures::{BoxFuture, Future};
+use futures::{Future, future};
 use publisher::FutureServiceExt as PublisherExt;
+use std::cell::RefCell;
 use std::collections::HashMap;
 use std::net::SocketAddr;
-use std::sync::{Arc, Mutex};
+use std::rc::Rc;
 use std::thread;
 use std::time::Duration;
 use subscriber::FutureServiceExt as SubscriberExt;
 use tarpc::{client, server};
-use tarpc::client::future::Connect as Fc;
-use tarpc::client::sync::Connect as Sc;
+use tarpc::client::future::ClientExt;
 use tarpc::util::{FirstSocketAddr, Message, Never};
+use tokio_core::reactor;
 
 pub mod subscriber {
     service! {
@@ -57,39 +58,39 @@ impl subscriber::FutureService for Subscriber {
 }
 
 impl Subscriber {
-    fn listen(id: u32) -> SocketAddr {
+    fn listen(id: u32, handle: &reactor::Handle, options: server::Options) -> SocketAddr {
         Subscriber { id: id }
-            .listen("localhost:0".first_socket_addr(),
-                    server::Options::default())
-            .wait()
+            .listen("localhost:0".first_socket_addr(), handle, options)
             .unwrap()
     }
 }
 
 #[derive(Clone, Debug)]
 struct Publisher {
-    clients: Arc<Mutex<HashMap<u32, subscriber::FutureClient>>>,
+    clients: Rc<RefCell<HashMap<u32, subscriber::FutureClient>>>,
 }
 
 impl Publisher {
     fn new() -> Publisher {
-        Publisher { clients: Arc::new(Mutex::new(HashMap::new())) }
+        Publisher { clients: Rc::new(RefCell::new(HashMap::new())) }
     }
 }
 
 impl publisher::FutureService for Publisher {
-    type BroadcastFut = BoxFuture<(), Never>;
+    type BroadcastFut = Box<Future<Item = (), Error = Never>>;
 
     fn broadcast(&self, message: String) -> Self::BroadcastFut {
-        futures::collect(self.clients
-                             .lock()
-                             .unwrap()
-                             .values_mut()
-                             // Ignore failing subscribers.
-                             .map(move |client| client.receive(message.clone()).then(|_| Ok(())))
-                             .collect::<Vec<_>>())
-            .map(|_| ())
-            .boxed()
+        let acks = self.clients
+                       .borrow()
+                       .values()
+                       .map(move |client| client.receive(message.clone())
+                           // Ignore failing subscribers. In a real pubsub,
+                           // you'd want to continually retry until subscribers
+                           // ack.
+                           .then(|_| Ok(())))
+                       // Collect to a vec to end the borrow on `self.clients`.
+                       .collect::<Vec<_>>();
+        Box::new(future::join_all(acks).map(|_| ()))
     }
 
     type SubscribeFut = Box<Future<Item = (), Error = Message>>;
@@ -99,42 +100,45 @@ impl publisher::FutureService for Publisher {
         Box::new(subscriber::FutureClient::connect(address, client::Options::default())
             .map(move |subscriber| {
                 println!("Subscribing {}.", id);
-                clients.lock().unwrap().insert(id, subscriber);
+                clients.borrow_mut().insert(id, subscriber);
                 ()
             })
             .map_err(|e| e.to_string().into()))
     }
 
-    type UnsubscribeFut = BoxFuture<(), Never>;
+    type UnsubscribeFut = Box<Future<Item = (), Error = Never>>;
 
     fn unsubscribe(&self, id: u32) -> Self::UnsubscribeFut {
         println!("Unsubscribing {}", id);
-        self.clients.lock().unwrap().remove(&id).unwrap();
+        self.clients.borrow_mut().remove(&id).unwrap();
         futures::finished(()).boxed()
     }
 }
 
 fn main() {
     let _ = env_logger::init();
+    let mut reactor = reactor::Core::new().unwrap();
     let publisher_addr = Publisher::new()
         .listen("localhost:0".first_socket_addr(),
+                &reactor.handle(),
                 server::Options::default())
-        .wait()
         .unwrap();
 
-    let mut publisher_client =
-        publisher::SyncClient::connect(publisher_addr, client::Options::default()).unwrap();
+    let subscriber1 = Subscriber::listen(0, &reactor.handle(), server::Options::default());
+    let subscriber2 = Subscriber::listen(1, &reactor.handle(), server::Options::default());
 
-    let subscriber1 = Subscriber::listen(0);
-    publisher_client.subscribe(0, subscriber1).unwrap();
-
-    let subscriber2 = Subscriber::listen(1);
-    publisher_client.subscribe(1, subscriber2).unwrap();
-
-
-    println!("Broadcasting...");
-    publisher_client.broadcast("hello to all".to_string()).unwrap();
-    publisher_client.unsubscribe(1).unwrap();
-    publisher_client.broadcast("hello again".to_string()).unwrap();
+    let publisher =
+        reactor.run(publisher::FutureClient::connect(publisher_addr, client::Options::default()))
+            .unwrap();
+    reactor.run(publisher.subscribe(0, subscriber1)
+            .and_then(|_| publisher.subscribe(1, subscriber2))
+            .map_err(|e| panic!(e))
+            .and_then(|_| {
+                println!("Broadcasting...");
+                publisher.broadcast("hello to all".to_string())
+            })
+            .and_then(|_| publisher.unsubscribe(1))
+            .and_then(|_| publisher.broadcast("hi again".to_string())))
+        .unwrap();
     thread::sleep(Duration::from_millis(300));
 }

examples/readme_errors.rs

@@ -11,11 +11,13 @@ extern crate futures;
 extern crate tarpc;
 #[macro_use]
 extern crate serde_derive;
+extern crate tokio_core;
 
 use std::error::Error;
 use std::fmt;
 use tarpc::{client, server};
-use tarpc::client::sync::Connect;
+use tarpc::client::sync::ClientExt;
+use tarpc::util::FirstSocketAddr;
 
 service! {
     rpc hello(name: String) -> String | NoNameGiven;
@@ -50,7 +52,9 @@ impl SyncService for HelloServer {
 }
 
 fn main() {
-    let addr = HelloServer.listen("localhost:10000", server::Options::default()).unwrap();
+    let addr = HelloServer.listen("localhost:10000".first_socket_addr(),
+                server::Options::default())
+        .unwrap();
     let mut client = SyncClient::connect(addr, client::Options::default()).unwrap();
     println!("{}", client.hello("Mom".to_string()).unwrap());
     println!("{}", client.hello("".to_string()).unwrap_err());

examples/readme_futures.rs

@@ -13,7 +13,7 @@ extern crate tokio_core;
 
 use futures::Future;
 use tarpc::{client, server};
-use tarpc::client::future::Connect;
+use tarpc::client::future::ClientExt;
 use tarpc::util::{FirstSocketAddr, Never};
 use tokio_core::reactor;
 
@@ -33,11 +33,13 @@ impl FutureService for HelloServer {
 }
 
 fn main() {
-    let addr = "localhost:10000".first_socket_addr();
-    let mut core = reactor::Core::new().unwrap();
-    HelloServer.listen(addr, server::Options::default().handle(core.handle())).wait().unwrap();
-    let options = client::Options::default().handle(core.handle());
-    core.run(FutureClient::connect(addr, options)
+    let mut reactor = reactor::Core::new().unwrap();
+    let addr = HelloServer.listen("localhost:10000".first_socket_addr(),
+                &reactor.handle(),
+                server::Options::default())
+        .unwrap();
+    let options = client::Options::default().handle(reactor.handle());
+    reactor.run(FutureClient::connect(addr, options)
             .map_err(tarpc::Error::from)
             .and_then(|client| client.hello("Mom".to_string()))
             .map(|resp| println!("{}", resp)))

examples/readme_sync.rs

@@ -10,10 +10,11 @@
 extern crate futures;
 #[macro_use]
 extern crate tarpc;
+extern crate tokio_core;
 
 use tarpc::{client, server};
-use tarpc::client::sync::Connect;
-use tarpc::util::Never;
+use tarpc::client::sync::ClientExt;
+use tarpc::util::{FirstSocketAddr, Never};
 
 service! {
     rpc hello(name: String) -> String;
@@ -29,8 +30,9 @@ impl SyncService for HelloServer {
 }
 
 fn main() {
-    let addr = "localhost:10000";
-    HelloServer.listen(addr, server::Options::default()).unwrap();
+    let addr = HelloServer.listen("localhost:0".first_socket_addr(),
+                server::Options::default())
+        .unwrap();
     let mut client = SyncClient::connect(addr, client::Options::default()).unwrap();
     println!("{}", client.hello("Mom".to_string()).unwrap());
 }

examples/server_calling_server.rs

@@ -10,15 +10,15 @@ extern crate env_logger;
 #[macro_use]
 extern crate tarpc;
 extern crate futures;
+extern crate tokio_core;
 
 use add::{FutureService as AddFutureService, FutureServiceExt as AddExt};
 use double::{FutureService as DoubleFutureService, FutureServiceExt as DoubleExt};
-use futures::{BoxFuture, Future};
-use std::sync::{Arc, Mutex};
+use futures::{BoxFuture, Future, Stream};
 use tarpc::{client, server};
-use tarpc::client::future::Connect as Fc;
-use tarpc::client::sync::Connect as Sc;
+use tarpc::client::future::ClientExt as Fc;
 use tarpc::util::{FirstSocketAddr, Message, Never};
+use tokio_core::reactor;
 
 pub mod add {
     service! {
@@ -49,12 +49,12 @@ impl AddFutureService for AddServer {
 
 #[derive(Clone)]
 struct DoubleServer {
-    client: Arc<Mutex<add::FutureClient>>,
+    client: add::FutureClient,
 }
 
 impl DoubleServer {
     fn new(client: add::FutureClient) -> Self {
-        DoubleServer { client: Arc::new(Mutex::new(client)) }
+        DoubleServer { client: client }
     }
 }
 
@@ -63,8 +63,6 @@ impl DoubleFutureService for DoubleServer {
 
     fn double(&self, x: i32) -> Self::DoubleFut {
         self.client
-            .lock()
-            .unwrap()
             .add(x, x)
             .map_err(|e| e.to_string().into())
             .boxed()
@@ -73,22 +71,29 @@ impl DoubleFutureService for DoubleServer {
 
 fn main() {
     let _ = env_logger::init();
+    let mut reactor = reactor::Core::new().unwrap();
     let add_addr = AddServer.listen("localhost:0".first_socket_addr(),
+                &reactor.handle(),
                 server::Options::default())
-        .wait()
-        .unwrap();
-    let add_client =
-        add::FutureClient::connect(add_addr, client::Options::default()).wait().unwrap();
-
-    let double = DoubleServer::new(add_client);
-    let double_addr = double.listen("localhost:0".first_socket_addr(),
-                server::Options::default())
-        .wait()
         .unwrap();
 
-    let mut double_client = double::SyncClient::connect(double_addr, client::Options::default())
+    let options = client::Options::default().handle(reactor.handle());
+    let add_client = reactor.run(add::FutureClient::connect(add_addr, options)).unwrap();
+
+    let double_addr = DoubleServer::new(add_client)
+        .listen("localhost:0".first_socket_addr(),
+                &reactor.handle(),
+                server::Options::default())
+        .unwrap();
+
+    let double_client =
+        reactor.run(double::FutureClient::connect(double_addr, client::Options::default()))
+            .unwrap();
+    reactor.run(futures::stream::futures_unordered((0..5).map(|i| double_client.double(i)))
+            .map_err(|e| println!("{}", e))
+            .for_each(|i| {
+                println!("{:?}", i);
+                Ok(())
+            }))
         .unwrap();
-    for i in 0..5 {
-        println!("{:?}", double_client.double(i).unwrap());
-    }
 }

examples/throughput.rs

@@ -12,16 +12,17 @@ extern crate lazy_static;
 extern crate tarpc;
 extern crate env_logger;
 extern crate futures;
+extern crate tokio_core;
 
-use futures::Future;
 use std::io::{Read, Write, stdout};
 use std::net;
 use std::sync::Arc;
 use std::thread;
 use std::time;
 use tarpc::{client, server};
-use tarpc::client::sync::Connect;
+use tarpc::client::sync::ClientExt;
 use tarpc::util::{FirstSocketAddr, Never};
+use tokio_core::reactor;
 
 lazy_static! {
     static ref BUF: Arc<Vec<u8>> = Arc::new(gen_vec(CHUNK_SIZE as usize));
@@ -53,9 +54,10 @@ impl FutureService for Server {
 const CHUNK_SIZE: u32 = 1 << 19;
 
 fn bench_tarpc(target: u64) {
+    let reactor = reactor::Core::new().unwrap();
     let addr = Server.listen("localhost:0".first_socket_addr(),
+                &reactor.handle(),
                 server::Options::default())
-        .wait()
         .unwrap();
     let mut client = SyncClient::connect(addr, client::Options::default()).unwrap();
     let start = time::Instant::now();

examples/two_clients.rs

@@ -13,13 +13,14 @@ extern crate tarpc;
 extern crate bincode;
 extern crate env_logger;
 extern crate futures;
+extern crate tokio_core;
 
 use bar::FutureServiceExt as BarExt;
 use baz::FutureServiceExt as BazExt;
-use futures::Future;
 use tarpc::{client, server};
-use tarpc::client::sync::Connect;
+use tarpc::client::sync::ClientExt;
 use tarpc::util::{FirstSocketAddr, Never};
+use tokio_core::reactor;
 
 mod bar {
     service! {
@@ -59,17 +60,27 @@ macro_rules! pos {
 
 fn main() {
     let _ = env_logger::init();
-    let bar_addr = Bar.listen("localhost:0".first_socket_addr(),
-                server::Options::default())
-        .wait()
-        .unwrap();
-    let baz_addr = Baz.listen("localhost:0".first_socket_addr(),
-                server::Options::default())
-        .wait()
-        .unwrap();
+    let mut bar_client = {
+        let reactor = reactor::Core::new().unwrap();
+        let addr = Bar.listen("localhost:0".first_socket_addr(),
+                    &reactor.handle(),
+                    server::Options::default())
+            .unwrap();
+        // TODO: Need to set up each client with its own reactor. Should it be shareable across
+        // multiple clients? e.g. Rc<RefCell<Core>> or something similar?
+        bar::SyncClient::connect(addr, client::Options::default()).unwrap()
+    };
+
+    let mut baz_client = {
+        // Need to set up each client with its own reactor.
+        let reactor = reactor::Core::new().unwrap();
+        let addr = Baz.listen("localhost:0".first_socket_addr(),
+                    &reactor.handle(),
+                    server::Options::default())
+            .unwrap();
+        baz::SyncClient::connect(addr, client::Options::default()).unwrap()
+    };
 
-    let mut bar_client = bar::SyncClient::connect(bar_addr, client::Options::default()).unwrap();
-    let mut baz_client = baz::SyncClient::connect(baz_addr, client::Options::default()).unwrap();
 
     info!("Result: {:?}", bar_client.bar(17));