Add server::Handle::shutdown

Cargo.toml

@@ -15,7 +15,7 @@ description = "An RPC framework for Rust with a focus on ease of use."
 travis-ci = { repository = "google/tarpc" }
 
 [dependencies]
-bincode = "1.0.0-alpha2"
+bincode = { git = "https://github.com/tikue/bincode", branch = "faster-byte-buf-deserialization" }
 byteorder = "1.0"
 cfg-if = "0.1.0"
 futures = "0.1.7"

benches/latency.rs

@@ -40,12 +40,13 @@ impl FutureService for Server {
 fn latency(bencher: &mut Bencher) {
     let _ = env_logger::init();
     let mut reactor = reactor::Core::new().unwrap();
-    let (addr, server) = Server.listen("localhost:0".first_socket_addr(),
+    let (handle, server) = Server.listen("localhost:0".first_socket_addr(),
                 &reactor.handle(),
                 server::Options::default())
         .unwrap();
     reactor.handle().spawn(server);
-    let client = FutureClient::connect(addr, client::Options::default().handle(reactor.handle()));
+    let client = FutureClient::connect(handle.addr(),
+                                       client::Options::default().handle(reactor.handle()));
     let client = reactor.run(client).unwrap();
 
     bencher.iter(|| reactor.run(client.ack()).unwrap());

examples/concurrency.rs

@@ -167,20 +167,20 @@ fn main() {
         .unwrap_or(4);
 
     let mut reactor = reactor::Core::new().unwrap();
-    let (addr, server) = Server::new()
+    let (handle, server) = Server::new()
         .listen("localhost:0".first_socket_addr(),
                 &reactor.handle(),
                 server::Options::default())
         .unwrap();
     reactor.handle().spawn(server);
-    info!("Server listening on {}.", addr);
+    info!("Server listening on {}.", handle.addr());
 
     let clients = (0..num_clients)
         // Spin up a couple threads to drive the clients.
         .map(|i| (i, spawn_core()))
         .map(|(i, remote)| {
             info!("Client {} connecting...", i);
-            FutureClient::connect(addr, client::Options::default().remote(remote))
+            FutureClient::connect(handle.addr(), client::Options::default().remote(remote))
                 .map_err(|e| panic!(e))
         });
 

examples/pubsub.rs

@@ -58,12 +58,15 @@ impl subscriber::FutureService for Subscriber {
 }
 
 impl Subscriber {
-    fn listen(id: u32, handle: &reactor::Handle, options: server::Options) -> SocketAddr {
-        let (addr, server) = Subscriber { id: id }
+    fn listen(id: u32,
+              handle: &reactor::Handle,
+              options: server::Options)
+              -> server::future::Handle {
+        let (server_handle, server) = Subscriber { id: id }
             .listen("localhost:0".first_socket_addr(), handle, options)
             .unwrap();
         handle.spawn(server);
-        addr
+        server_handle
     }
 }
 
@@ -120,7 +123,7 @@ impl publisher::FutureService for Publisher {
 fn main() {
     let _ = env_logger::init();
     let mut reactor = reactor::Core::new().unwrap();
-    let (publisher_addr, server) = Publisher::new()
+    let (publisher_handle, server) = Publisher::new()
         .listen("localhost:0".first_socket_addr(),
                 &reactor.handle(),
                 server::Options::default())
@@ -131,10 +134,11 @@ fn main() {
     let subscriber2 = Subscriber::listen(1, &reactor.handle(), server::Options::default());
 
     let publisher =
-        reactor.run(publisher::FutureClient::connect(publisher_addr, client::Options::default()))
+        reactor.run(publisher::FutureClient::connect(publisher_handle.addr(),
+                                                  client::Options::default()))
             .unwrap();
-    reactor.run(publisher.subscribe(0, subscriber1)
-            .and_then(|_| publisher.subscribe(1, subscriber2))
+    reactor.run(publisher.subscribe(0, subscriber1.addr())
+            .and_then(|_| publisher.subscribe(1, subscriber2.addr()))
             .map_err(|e| panic!(e))
             .and_then(|_| {
                 println!("Broadcasting...");

examples/readme_errors.rs

@@ -55,8 +55,7 @@ impl SyncService for HelloServer {
 fn main() {
     let (tx, rx) = mpsc::channel();
     thread::spawn(move || {
-        let mut handle = HelloServer.listen("localhost:10000", server::Options::default())
-            .unwrap();
+        let handle = HelloServer.listen("localhost:10000", server::Options::default()).unwrap();
         tx.send(handle.addr()).unwrap();
         handle.run();
     });

examples/readme_futures.rs

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

examples/readme_sync.rs

@@ -34,8 +34,7 @@ impl SyncService for HelloServer {
 fn main() {
     let (tx, rx) = mpsc::channel();
     thread::spawn(move || {
-        let mut handle = HelloServer.listen("localhost:0", server::Options::default())
-            .unwrap();
+        let handle = HelloServer.listen("localhost:0", server::Options::default()).unwrap();
         tx.send(handle.addr()).unwrap();
         handle.run();
     });

examples/server_calling_server.rs

@@ -72,24 +72,24 @@ impl DoubleFutureService for DoubleServer {
 fn main() {
     let _ = env_logger::init();
     let mut reactor = reactor::Core::new().unwrap();
-    let (add_addr, server) = AddServer.listen("localhost:0".first_socket_addr(),
+    let (add, server) = AddServer.listen("localhost:0".first_socket_addr(),
                 &reactor.handle(),
                 server::Options::default())
         .unwrap();
     reactor.handle().spawn(server);
 
     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_addr, server) = DoubleServer::new(add_client)
+    let (double, server) = DoubleServer::new(add_client)
         .listen("localhost:0".first_socket_addr(),
                 &reactor.handle(),
                 server::Options::default())
         .unwrap();
     reactor.handle().spawn(server);
 
     let double_client =
-        reactor.run(double::FutureClient::connect(double_addr, client::Options::default()))
+        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))

examples/throughput.rs

@@ -66,7 +66,8 @@ fn bench_tarpc(target: u64) {
         tx.send(addr).unwrap();
         reactor.run(server).unwrap();
     });
-    let mut client = SyncClient::connect(rx.recv().unwrap(), client::Options::default()).unwrap();
+    let mut client = SyncClient::connect(rx.recv().unwrap().addr(), client::Options::default())
+        .unwrap();
     let start = time::Instant::now();
     let mut nread = 0;
     while nread < target {

examples/two_clients.rs

@@ -66,30 +66,30 @@ fn main() {
         let (tx, rx) = mpsc::channel();
         thread::spawn(move || {
             let mut reactor = reactor::Core::new().unwrap();
-            let (addr, server) = Bar.listen("localhost:0".first_socket_addr(),
+            let (handle, server) = Bar.listen("localhost:0".first_socket_addr(),
                         &reactor.handle(),
                         server::Options::default())
                 .unwrap();
-            tx.send(addr).unwrap();
+            tx.send(handle).unwrap();
             reactor.run(server).unwrap();
         });
-        let addr = rx.recv().unwrap();
-        bar::SyncClient::connect(addr, client::Options::default()).unwrap()
+        let handle = rx.recv().unwrap();
+        bar::SyncClient::connect(handle.addr(), client::Options::default()).unwrap()
     };
 
     let mut baz_client = {
         let (tx, rx) = mpsc::channel();
         thread::spawn(move || {
             let mut reactor = reactor::Core::new().unwrap();
-            let (addr, server) = Baz.listen("localhost:0".first_socket_addr(),
+            let (handle, server) = Baz.listen("localhost:0".first_socket_addr(),
                         &reactor.handle(),
                         server::Options::default())
                 .unwrap();
-            tx.send(addr).unwrap();
+            tx.send(handle).unwrap();
             reactor.run(server).unwrap();
         });
-        let addr = rx.recv().unwrap();
-        baz::SyncClient::connect(addr, client::Options::default()).unwrap()
+        let handle = rx.recv().unwrap();
+        baz::SyncClient::connect(handle.addr(), client::Options::default()).unwrap()
     };
 
 

src/lib.rs

@@ -114,7 +114,7 @@
 //! ```
 //!
 #![deny(missing_docs)]
-#![feature(never_type, plugin, struct_field_attributes, fn_traits, unboxed_closures)]
+#![feature(fn_traits, move_cell, never_type, plugin, struct_field_attributes, unboxed_closures)]
 #![plugin(tarpc_plugins)]
 
 extern crate byteorder;