X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-block-sync%2Fsrc%2Frpc.rs;h=1e0aa9d93fc45e9113c8a78ff2c545e551df50ba;hb=3cdbbf56e42daf67b954cd983052b09a9a18b8cd;hp=88199688aefd1a48fd128aac5c45f319f947c90e;hpb=0a0f87c00f3c5e30965751fb46502ebc1f110333;p=rust-lightning diff --git a/lightning-block-sync/src/rpc.rs b/lightning-block-sync/src/rpc.rs index 88199688..1e0aa9d9 100644 --- a/lightning-block-sync/src/rpc.rs +++ b/lightning-block-sync/src/rpc.rs @@ -8,6 +8,8 @@ use bitcoin::blockdata::block::Block; use bitcoin::hash_types::BlockHash; use bitcoin::hashes::hex::ToHex; +use futures::lock::Mutex; + use serde_json; use std::convert::TryFrom; @@ -18,7 +20,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; pub struct RpcClient { basic_auth: String, endpoint: HttpEndpoint, - client: HttpClient, + client: Mutex, id: AtomicUsize, } @@ -27,7 +29,7 @@ impl RpcClient { /// credentials should be a base64 encoding of a user name and password joined by a colon, as is /// required for HTTP basic access authentication. pub fn new(credentials: &str, endpoint: HttpEndpoint) -> std::io::Result { - let client = HttpClient::connect(&endpoint)?; + let client = Mutex::new(HttpClient::connect(&endpoint)?); Ok(Self { basic_auth: "Basic ".to_string() + credentials, endpoint, @@ -37,7 +39,7 @@ impl RpcClient { } /// Calls a method with the response encoded in JSON format and interpreted as type `T`. - pub async fn call_method(&mut self, method: &str, params: &[serde_json::Value]) -> std::io::Result + pub async fn call_method(&self, method: &str, params: &[serde_json::Value]) -> std::io::Result where JsonResponse: TryFrom, Error = std::io::Error> + TryInto { let host = format!("{}:{}", self.endpoint.host(), self.endpoint.port()); let uri = self.endpoint.path(); @@ -47,7 +49,7 @@ impl RpcClient { "id": &self.id.fetch_add(1, Ordering::AcqRel).to_string() }); - let mut response = match self.client.post::(&uri, &host, &self.basic_auth, content).await { + let mut response = match self.client.lock().await.post::(&uri, &host, &self.basic_auth, content).await { Ok(JsonResponse(response)) => response, Err(e) if e.kind() == std::io::ErrorKind::Other => { match e.get_ref().unwrap().downcast_ref::() { @@ -82,14 +84,14 @@ impl RpcClient { } impl BlockSource for RpcClient { - fn get_header<'a>(&'a mut self, header_hash: &'a BlockHash, _height: Option) -> AsyncBlockSourceResult<'a, BlockHeaderData> { + fn get_header<'a>(&'a self, header_hash: &'a BlockHash, _height: Option) -> AsyncBlockSourceResult<'a, BlockHeaderData> { Box::pin(async move { let header_hash = serde_json::json!(header_hash.to_hex()); Ok(self.call_method("getblockheader", &[header_hash]).await?) }) } - fn get_block<'a>(&'a mut self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, Block> { + fn get_block<'a>(&'a self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, Block> { Box::pin(async move { let header_hash = serde_json::json!(header_hash.to_hex()); let verbosity = serde_json::json!(0); @@ -97,7 +99,7 @@ impl BlockSource for RpcClient { }) } - fn get_best_block<'a>(&'a mut self) -> AsyncBlockSourceResult<'a, (BlockHash, Option)> { + fn get_best_block<'a>(&'a self) -> AsyncBlockSourceResult<'a, (BlockHash, Option)> { Box::pin(async move { Ok(self.call_method("getblockchaininfo", &[]).await?) }) @@ -127,7 +129,7 @@ mod tests { #[tokio::test] async fn call_method_returning_unknown_response() { let server = HttpServer::responding_with_not_found(); - let mut client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); + let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); match client.call_method::("getblockcount", &[]).await { Err(e) => assert_eq!(e.kind(), std::io::ErrorKind::Other), @@ -139,7 +141,7 @@ mod tests { async fn call_method_returning_malfomred_response() { let response = serde_json::json!("foo"); let server = HttpServer::responding_with_ok(MessageBody::Content(response)); - let mut client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); + let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); match client.call_method::("getblockcount", &[]).await { Err(e) => { @@ -156,7 +158,7 @@ mod tests { "error": { "code": -8, "message": "invalid parameter" }, }); let server = HttpServer::responding_with_server_error(response); - let mut client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); + let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); let invalid_block_hash = serde_json::json!("foo"); match client.call_method::("getblock", &[invalid_block_hash]).await { @@ -172,7 +174,7 @@ mod tests { async fn call_method_returning_missing_result() { let response = serde_json::json!({ "result": null }); let server = HttpServer::responding_with_ok(MessageBody::Content(response)); - let mut client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); + let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); match client.call_method::("getblockcount", &[]).await { Err(e) => { @@ -187,7 +189,7 @@ mod tests { async fn call_method_returning_malformed_result() { let response = serde_json::json!({ "result": "foo" }); let server = HttpServer::responding_with_ok(MessageBody::Content(response)); - let mut client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); + let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); match client.call_method::("getblockcount", &[]).await { Err(e) => { @@ -202,7 +204,7 @@ mod tests { async fn call_method_returning_valid_result() { let response = serde_json::json!({ "result": 654470 }); let server = HttpServer::responding_with_ok(MessageBody::Content(response)); - let mut client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); + let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap(); match client.call_method::("getblockcount", &[]).await { Err(e) => panic!("Unexpected error: {:?}", e),