Kill Handler for the api http rpc server

commerceblock · Dec 11, 2019 · f591541 · f591541
1 parent 98faa22
commit f591541
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 27 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -16,7 +16,7 @@ config = "0.9"
 serde = { version = "1.0", features = ["derive"] }
 serde_json="1.0"
 mongodb = "0.3.11"
-jsonrpc-http-server = "11.0"
+jsonrpc-http-server = "14.0.5"
 rust-ocean = { git = "https://github.com/commerceblock/rust-ocean"}
 ocean-rpc = { git = "https://github.com/commerceblock/rust-ocean-rpc"}
 bitcoin = { version = "0.20", features = [ "use-serde" ] }
diff --git a/src/api.rs b/src/api.rs
@@ -11,7 +11,9 @@ use base64::decode as b64decode;
 use bitcoin::hashes::sha256d;
 use hyper::{Body, Request, StatusCode};
 use jsonrpc_http_server::jsonrpc_core::{Error, ErrorCode, IoHandler, Params, Value};
-use jsonrpc_http_server::{hyper::header, AccessControlAllowOrigin, DomainsValidation, Response, ServerBuilder};
+use jsonrpc_http_server::{
+    hyper::header, AccessControlAllowOrigin, CloseHandle, DomainsValidation, Response, ServerBuilder,
+};
 use serde::{Deserialize, Serialize};
 
 use crate::config::ApiConfig;
@@ -148,10 +150,7 @@ fn authorize(our_auth: &str, request: &Request<Body>) -> bool {
 /// Run Api RPC server for external requests that require information from the
 /// coordinator. Data returned to the caller are drawn from the storage
 /// interface which is shared with the main coordinator process
-pub fn run_api_server<D: Storage + Send + Sync + 'static>(
-    config: &ApiConfig,
-    storage: Arc<D>,
-) -> thread::JoinHandle<()> {
+pub fn run_api_server<D: Storage + Send + Sync + 'static>(config: &ApiConfig, storage: Arc<D>) -> CloseHandle {
     let mut io = IoHandler::default();
     let storage_ref = storage.clone();
     io.add_method("getrequestresponse", move |params: Params| {
@@ -189,7 +188,9 @@ pub fn run_api_server<D: Storage + Send + Sync + 'static>(
         .start_http(&addr[0])
         .expect("api error");
 
-    thread::spawn(move || server.wait())
+    let close_handle = server.close_handle();
+    let _ = thread::spawn(move || server.wait());
+    close_handle // handler to stop the server from the main thread
 }
 
 #[cfg(test)]

diff --git a/src/coordinator.rs b/src/coordinator.rs
@@ -24,22 +24,30 @@ pub fn run(config: Config) -> Result<()> {
     let storage = Arc::new(MongoStorage::new(config.storage.clone())?);
     let genesis_hash = sha256d::Hash::from_hex(&config.clientchain.genesis_hash)?;
 
-    let _ = ::api::run_api_server(&config.api, storage.clone());
+    let api_handler = ::api::run_api_server(&config.api, storage.clone());
     let (req_send, req_recv): (Sender<sha256d::Hash>, Receiver<sha256d::Hash>) = channel();
     let _ = ::payments::run_payments(config.clientchain.clone(), storage.clone(), req_recv)?;
 
     // This loop runs continuously fetching and running challenge requests,
     // generating challenge responses and fails on any errors that occur
     loop {
-        if let Some(request_id) = run_request(&config, &service, &clientchain, storage.clone(), genesis_hash)? {
-            // if challenge request succeeds print responses
-            req_send.send(request_id).unwrap();
-            info! {"***** Response *****"}
-            let resp = storage.get_response(request_id)?.unwrap();
-            info! {"{}", serde_json::to_string_pretty(&resp).unwrap()};
+        match run_request(&config, &service, &clientchain, storage.clone(), genesis_hash) {
+            Ok(res) => {
+                if let Some(request_id) = res {
+                    // if challenge request succeeds print responses
+                    req_send.send(request_id).unwrap();
+                    info! {"***** Response *****"}
+                    let resp = storage.get_response(request_id)?.unwrap();
+                    info! {"{}", serde_json::to_string_pretty(&resp).unwrap()};
+                }
+                info! {"Sleeping for {} sec...", config.block_time}
+                thread::sleep(time::Duration::from_secs(config.block_time))
+            }
+            Err(err) => {
+                api_handler.close(); // try closing the api rpc server
+                return Err(err);
+            }
         }
-        info! {"Sleeping for {} sec...", config.block_time}
-        thread::sleep(time::Duration::from_secs(config.block_time))
     }
 }