Merge branch 'improve-error-handling'

Author: faern

http/src/lib.rs

@@ -229,10 +229,12 @@ impl HttpTransport {
                 Ok((mut core, request_tx, future)) => {
                     tx.send(Ok(HttpTransport::new_internal(request_tx)))
                         .unwrap();
-                    let _ = core.run(future);
+                    if let Err(_) = core.run(future) {
+                        error!("JSON-RPC processing thread had an error");
+                    }
+                    debug!("Standalone HttpTransport thread exiting");
                 }
             }
-            debug!("Standalone HttpTransport thread exiting");
         });
 
         rx.recv().unwrap()
@@ -291,6 +293,7 @@ fn create_request_processing_future<CC: hyper::client::Connect>(
     client: Client<CC, hyper::Body>,
 ) -> Box<Future<Item = (), Error = ()>> {
     let f = request_rx.for_each(move |(request, response_tx)| {
+        trace!("Sending request to {}", request.uri());
         client
             .request(request)
             .from_err()
@@ -306,10 +309,10 @@ fn create_request_processing_future<CC: hyper::client::Connect>(
             })
             .map(|response_chunk| response_chunk.to_vec())
             .then(move |response_result| {
-                response_tx.send(response_result).map_err(|_| {
+                if let Err(_) = response_tx.send(response_result) {
                     warn!("Unable to send response back to caller");
-                    ()
-                })
+                }
+                Ok(())
             })
     });
     Box::new(f) as Box<Future<Item = (), Error = ()>>

http/tests/localhost.rs

@@ -20,45 +20,49 @@ extern crate jsonrpc_http_server;
 extern crate jsonrpc_macros;
 
 use futures::Future;
-
+use futures::future::Either;
 use jsonrpc_client_http::HttpTransport;
 use jsonrpc_core::{Error, IoHandler};
 use jsonrpc_http_server::ServerBuilder;
+use std::time::Duration;
+use tokio_core::reactor::{Core, Timeout};
 
 
 // Generate server API trait. Actual implementation at bottom of file.
 build_rpc_trait! {
     pub trait ServerApi {
         #[rpc(name = "to_upper")]
         fn to_upper(&self, String) -> Result<String, Error>;
+
+        #[rpc(name = "sleep")]
+        fn sleep(&self, u64) -> Result<(), Error>;
     }
 }
 
 // Generate client struct with same API as server.
-jsonrpc_client!(pub struct ToUpperClient {
+jsonrpc_client!(pub struct TestClient {
     pub fn to_upper(&mut self, string: &str) -> RpcRequest<String>;
+    pub fn sleep(&mut self, time: u64) -> RpcRequest<()>;
 });
 
 
 #[test]
 fn localhost_ping_pong() {
-    env_logger::init().unwrap();
+    let _ = env_logger::init();
 
     // Spawn a server hosting the `ServerApi` API.
-    let server = spawn_server();
-
-    let uri = format!("http://{}", server.address());
+    let (_server, uri) = spawn_server();
     println!("Testing towards server at {}", uri);
 
     // Create the Tokio Core event loop that will drive the RPC client and the async requests.
-    let mut core = tokio_core::reactor::Core::new().unwrap();
+    let mut core = Core::new().unwrap();
 
     // Create the HTTP transport handle and create a RPC client with that handle.
     let transport = HttpTransport::shared(&core.handle())
         .unwrap()
         .handle(&uri)
         .unwrap();
-    let mut client = ToUpperClient::new(transport);
+    let mut client = TestClient::new(transport);
 
     // Just calling the method gives back a `RpcRequest`, which is a future
     // that can be used to execute the actual RPC call.
@@ -73,6 +77,34 @@ fn localhost_ping_pong() {
     assert_eq!("FOOBAR", result2);
 }
 
+#[test]
+fn dropped_rpc_request_should_not_crash_transport() {
+    let _ = env_logger::init();
+
+    let (_server, uri) = spawn_server();
+
+    let mut core = Core::new().unwrap();
+    let transport = HttpTransport::shared(&core.handle())
+        .unwrap()
+        .handle(&uri)
+        .unwrap();
+    let mut client = TestClient::new(transport);
+
+    let rpc = client.sleep(5).map_err(|e| e.to_string());
+    let timeout = Timeout::new(Duration::from_millis(100), &core.handle()).unwrap();
+    match core.run(rpc.select2(timeout.map_err(|e| e.to_string()))) {
+        Ok(Either::B(((), _rpc))) => (),
+        _ => panic!("The timeout did not finish first"),
+    }
+
+    // Now, sending a second request should still work. This is a regression test catching a
+    // previous error where a dropped `RpcRequest` would crash the future running on the event loop.
+    match core.run(client.sleep(0)) {
+        Ok(()) => (),
+        _ => panic!("Sleep did not return as it should"),
+    }
+}
+
 
 /// Simple struct that will implement the RPC API defined at the top of this file.
 struct Server;
@@ -81,14 +113,22 @@ impl ServerApi for Server {
     fn to_upper(&self, s: String) -> Result<String, Error> {
         Ok(s.to_uppercase())
     }
+
+    fn sleep(&self, time: u64) -> Result<(), Error> {
+        println!("Sleeping on server");
+        ::std::thread::sleep(Duration::from_secs(time));
+        Ok(())
+    }
 }
 
-fn spawn_server() -> jsonrpc_http_server::Server {
+fn spawn_server() -> (jsonrpc_http_server::Server, String) {
     let server = Server;
     let mut io = IoHandler::new();
     io.extend_with(server.to_delegate());
 
-    ServerBuilder::new(io)
+    let server = ServerBuilder::new(io)
         .start_http(&"127.0.0.1:0".parse().unwrap())
-        .unwrap()
+        .unwrap();
+    let uri = format!("http://{}", server.address());
+    (server, uri)
 }