rpc: Cleanup and separate namespaces into client and server.

bin/dao-cli/src/main.rs

@@ -3,7 +3,7 @@ use serde_json::{json, Value};
 use url::Url;
 
 use darkfi::{
-    rpc::{jsonrpc, rpcclient::RpcClient},
+    rpc::{client::RpcClient, jsonrpc::JsonRequest},
     Result,
 };
 
@@ -32,7 +32,7 @@ impl Rpc {
     // --> {"jsonrpc": "2.0", "method": "say_hello", "params": [], "id": 42}
     // <-- {"jsonrpc": "2.0", "result": "hello world", "id": 42}
     async fn say_hello(&self) -> Result<Value> {
-        let req = jsonrpc::request(json!("say_hello"), json!([]));
+        let req = JsonRequest::new("say_hello", json!([]));
         self.client.request(req).await
     }
 }

bin/daod/src/main.rs

@@ -8,7 +8,7 @@ use url::Url;
 
 use darkfi::{
     rpc::{
-        jsonrpc::{error as jsonerr, response as jsonresp, ErrorCode::*, JsonRequest, JsonResult},
+        jsonrpc::{ErrorCode::*, JsonError, JsonRequest, JsonResponse, JsonResult},
         server::{listen_and_serve, RequestHandler},
     },
     Result,
@@ -28,14 +28,14 @@ struct JsonRpcInterface {}
 impl RequestHandler for JsonRpcInterface {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if req.params.as_array().is_none() {
-            return JsonResult::Err(jsonerr(InvalidParams, None, req.id))
+            return JsonError::new(InvalidParams, None, req.id).into()
         }
 
         debug!(target: "RPC", "--> {}", serde_json::to_string(&req).unwrap());
 
         match req.method.as_str() {
             Some("say_hello") => return self.say_hello(req.id, req.params).await,
-            Some(_) | None => return JsonResult::Err(jsonerr(MethodNotFound, None, req.id)),
+            Some(_) | None => return JsonError::new(MethodNotFound, None, req.id).into(),
         }
     }
 }
@@ -44,7 +44,7 @@ impl JsonRpcInterface {
     // --> {"method": "say_hello", "params": []}
     // <-- {"result": "hello world"}
     async fn say_hello(&self, id: Value, _params: Value) -> JsonResult {
-        JsonResult::Resp(jsonresp(json!("hello world"), id))
+        JsonResponse::new(json!("hello world"), id).into()
     }
 }
 

bin/darkfid/src/error.rs

@@ -1,9 +1,6 @@
 use serde_json::Value;
 
-use darkfi::rpc::{
-    jsonrpc,
-    jsonrpc::{ErrorCode::ServerError, JsonResult},
-};
+use darkfi::rpc::jsonrpc::{ErrorCode::ServerError, JsonError, JsonResult};
 
 pub enum RpcError {
     Keygen = -32101,
@@ -18,6 +15,8 @@ pub enum RpcError {
     ParseError = -32110,
     TxBroadcastFail = -32111,
     NotYetSynced = -32112,
+    InvalidAddressParam = -32113,
+    InvalidAmountParam = -32114,
 }
 
 fn to_tuple(e: RpcError) -> (i64, String) {
@@ -34,6 +33,8 @@ fn to_tuple(e: RpcError) -> (i64, String) {
         RpcError::ParseError => "Parse error",
         RpcError::TxBroadcastFail => "Failed broadcasting transaction",
         RpcError::NotYetSynced => "Blockchain not yet synced",
+        RpcError::InvalidAddressParam => "Invalid address parameter",
+        RpcError::InvalidAmountParam => "invalid amount parameter",
     };
 
     (e as i64, msg.to_string())
@@ -41,5 +42,5 @@ fn to_tuple(e: RpcError) -> (i64, String) {
 
 pub fn server_error(e: RpcError, id: Value) -> JsonResult {
     let (code, msg) = to_tuple(e);
-    jsonrpc::error(ServerError(code), Some(msg), id).into()
+    JsonError::new(ServerError(code), Some(msg), id).into()
 }

bin/darkfid/src/main.rs

@@ -29,10 +29,9 @@ use darkfi::{
     net::P2pPtr,
     node::Client,
     rpc::{
-        jsonrpc,
         jsonrpc::{
             ErrorCode::{InvalidParams, MethodNotFound},
-            JsonRequest, JsonResult,
+            JsonError, JsonRequest, JsonResult,
         },
         server::{listen_and_serve, RequestHandler},
     },
@@ -155,7 +154,7 @@ mod rpc_wallet;
 impl RequestHandler for Darkfid {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if !req.params.is_array() {
-            return jsonrpc::error(InvalidParams, None, req.id).into()
+            return JsonError::new(InvalidParams, None, req.id).into()
         }
 
         let params = req.params.as_array().unwrap();
@@ -173,7 +172,7 @@ impl RequestHandler for Darkfid {
                 return self.set_default_address(req.id, params).await
             }
             Some("wallet.get_balances") => return self.get_balances(req.id, params).await,
-            Some(_) | None => return jsonrpc::error(MethodNotFound, None, req.id).into(),
+            Some(_) | None => return JsonError::new(MethodNotFound, None, req.id).into(),
         }
     }
 }

bin/darkfid/src/rpc_blockchain.rs

@@ -3,12 +3,9 @@ use serde_json::{json, Value};
 
 use darkfi::{
     crypto::merkle_node::MerkleNode,
-    rpc::{
-        jsonrpc,
-        jsonrpc::{
-            ErrorCode::{InternalError, InvalidParams},
-            JsonResult,
-        },
+    rpc::jsonrpc::{
+        ErrorCode::{InternalError, InvalidParams},
+        JsonError, JsonResponse, JsonResult,
     },
 };
 
@@ -23,7 +20,7 @@ impl Darkfid {
     // <-- {"jsonrpc": "2.0", "result": {...}, "id": 1}
     pub async fn get_slot(&self, id: Value, params: &[Value]) -> JsonResult {
         if params.len() != 1 || !params[0].is_u64() {
-            return jsonrpc::error(InvalidParams, None, id).into()
+            return JsonError::new(InvalidParams, None, id).into()
         }
 
         let blocks = match self
@@ -36,7 +33,7 @@ impl Darkfid {
             Ok(v) => v,
             Err(e) => {
                 error!("Failed fetching block by slot: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         };
 
@@ -46,7 +43,7 @@ impl Darkfid {
 
         // TODO: Return block as JSON
         debug!("{:#?}", blocks[0]);
-        jsonrpc::response(json!(true), id).into()
+        JsonResponse::new(json!(true), id).into()
     }
 
     // RPCAPI:
@@ -59,10 +56,10 @@ impl Darkfid {
                 Ok(v) => v,
                 Err(e) => {
                     error!("Failed getting merkle roots from rootstore: {}", e);
-                    return jsonrpc::error(InternalError, None, id).into()
+                    return JsonError::new(InternalError, None, id).into()
                 }
             };
 
-        jsonrpc::response(json!(roots), id).into()
+        JsonResponse::new(json!(roots), id).into()
     }
 }

bin/darkfid/src/rpc_misc.rs

@@ -1,6 +1,6 @@
 use serde_json::{json, Value};
 
-use darkfi::rpc::{jsonrpc, jsonrpc::JsonResult};
+use darkfi::rpc::jsonrpc::{JsonResponse, JsonResult};
 
 use super::Darkfid;
 
@@ -10,6 +10,6 @@ impl Darkfid {
     // --> {"jsonrpc": "2.0", "method": "ping", "params": [], "id": 1}
     // <-- {"jsonrpc": "2.0", "result": "pong", "id": 1}
     pub async fn pong(&self, id: Value, _params: &[Value]) -> JsonResult {
-        jsonrpc::response(json!("pong"), id).into()
+        JsonResponse::new(json!("pong"), id).into()
     }
 }

bin/darkfid/src/rpc_tx.rs

@@ -5,12 +5,9 @@ use serde_json::{json, Value};
 
 use darkfi::{
     crypto::{address::Address, keypair::PublicKey, token_id::generate_id},
-    rpc::{
-        jsonrpc,
-        jsonrpc::{
-            ErrorCode::{InternalError, InvalidAddressParam, InvalidAmountParam, InvalidParams},
-            JsonResult,
-        },
+    rpc::jsonrpc::{
+        ErrorCode::{InternalError, InvalidParams},
+        JsonError, JsonResponse, JsonResult,
     },
     util::{decode_base10, serial::serialize, NetworkName},
 };
@@ -31,7 +28,7 @@ impl Darkfid {
             !params[2].is_string() ||
             !params[3].is_f64()
         {
-            return jsonrpc::error(InvalidParams, None, id).into()
+            return JsonError::new(InvalidParams, None, id).into()
         }
 
         let network = params[0].as_str().unwrap();
@@ -48,7 +45,7 @@ impl Darkfid {
             Ok(v) => v,
             Err(e) => {
                 error!("transfer(): Failed parsing address from string: {}", e);
-                return jsonrpc::error(InvalidAddressParam, None, id).into()
+                return server_error(RpcError::InvalidAddressParam, id)
             }
         };
 
@@ -65,14 +62,14 @@ impl Darkfid {
             Ok(v) => v,
             Err(e) => {
                 error!("transfer(): Failed parsing amount from string: {}", e);
-                return jsonrpc::error(InvalidAmountParam, None, id).into()
+                return server_error(RpcError::InvalidAmountParam, id)
             }
         };
         let amount: u64 = match amount.try_into() {
             Ok(v) => v,
             Err(e) => {
                 error!("transfer(): Failed converting biguint to u64: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         };
 
@@ -92,7 +89,7 @@ impl Darkfid {
                     Ok(v) => v,
                     Err(e) => {
                         error!("transfer(): Failed generate_id(): {}", e);
-                        return jsonrpc::error(InternalError, None, id).into()
+                        return JsonError::new(InternalError, None, id).into()
                     }
                 }
             };
@@ -128,6 +125,6 @@ impl Darkfid {
         }
 
         let tx_hash = blake3::hash(&serialize(&tx)).to_hex().as_str().to_string();
-        jsonrpc::response(json!(tx_hash), id).into()
+        JsonResponse::new(json!(tx_hash), id).into()
     }
 }

bin/darkfid/src/rpc_wallet.rs

@@ -9,12 +9,9 @@ use darkfi::{
         address::Address,
         keypair::{Keypair, PublicKey, SecretKey},
     },
-    rpc::{
-        jsonrpc,
-        jsonrpc::{
-            ErrorCode::{InternalError, InvalidParams},
-            JsonResult,
-        },
+    rpc::jsonrpc::{
+        ErrorCode::{InternalError, InvalidParams},
+        JsonError, JsonResponse, JsonResult,
     },
     util::{decode_base10, encode_base10, NetworkName},
 };
@@ -29,7 +26,7 @@ impl Darkfid {
     // <-- {"jsonrpc": "2.0", "result": "1DarkFi...", "id": 1}
     pub async fn keygen(&self, id: Value, _params: &[Value]) -> JsonResult {
         match self.client.keygen().await {
-            Ok(a) => jsonrpc::response(json!(a.to_string()), id).into(),
+            Ok(a) => JsonResponse::new(json!(a.to_string()), id).into(),
             Err(e) => {
                 error!("Failed creating keypair: {}", e);
                 server_error(RpcError::Keygen, id)
@@ -44,7 +41,7 @@ impl Darkfid {
     // <-- {"jsonrpc": "2.0", "result": ["foo", "bar"], "id": 1}
     pub async fn get_key(&self, id: Value, params: &[Value]) -> JsonResult {
         if params.is_empty() {
-            return jsonrpc::error(InvalidParams, None, id).into()
+            return JsonError::new(InvalidParams, None, id).into()
         }
 
         let mut fetch_all = false;
@@ -88,7 +85,7 @@ impl Darkfid {
             }
         }
 
-        jsonrpc::response(json!(ret), id).into()
+        JsonResponse::new(json!(ret), id).into()
     }
 
     // RPCAPI:
@@ -98,7 +95,7 @@ impl Darkfid {
     // <-- {"jsonrpc": "2.0", "result": "foobar", "id": 1}
     pub async fn export_keypair(&self, id: Value, params: &[Value]) -> JsonResult {
         if params.len() != 1 || !params[0].is_u64() {
-            return jsonrpc::error(InvalidParams, None, id).into()
+            return JsonError::new(InvalidParams, None, id).into()
         }
 
         let keypairs = match self.client.get_keypairs().await {
@@ -110,7 +107,7 @@ impl Darkfid {
         };
 
         if let Some(kp) = keypairs.get(params[0].as_u64().unwrap() as usize) {
-            return jsonrpc::response(json!(kp.secret.to_bytes()), id).into()
+            return JsonResponse::new(json!(kp.secret.to_bytes()), id).into()
         }
 
         server_error(RpcError::KeypairNotFound, id)
@@ -123,7 +120,7 @@ impl Darkfid {
     // <-- {"jsonrpc": "2.0", "result": "pubfoobar", "id": 1}
     pub async fn import_keypair(&self, id: Value, params: &[Value]) -> JsonResult {
         if params.len() != 1 || !params[0].is_string() {
-            return jsonrpc::error(InvalidParams, None, id).into()
+            return JsonError::new(InvalidParams, None, id).into()
         }
 
         let bytes: [u8; 32] = match serde_json::from_str(params[0].as_str().unwrap()) {
@@ -150,11 +147,11 @@ impl Darkfid {
             Ok(()) => {}
             Err(e) => {
                 error!("Failed inserting keypair into wallet: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         };
 
-        jsonrpc::response(json!(address), id).into()
+        JsonResponse::new(json!(address), id).into()
     }
 
     // RPCAPI:
@@ -164,7 +161,7 @@ impl Darkfid {
     // <-- {"jsonrpc": "2.0", "result": true, "id": 1}
     pub async fn set_default_address(&self, id: Value, params: &[Value]) -> JsonResult {
         if params.len() != 1 || !params[0].is_u64() {
-            return jsonrpc::error(InvalidParams, None, id).into()
+            return JsonError::new(InvalidParams, None, id).into()
         }
 
         let idx = params[0].as_u64().unwrap();
@@ -186,11 +183,11 @@ impl Darkfid {
             Ok(()) => {}
             Err(e) => {
                 error!("Failed setting default keypair: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         };
 
-        jsonrpc::response(json!(true), id).into()
+        JsonResponse::new(json!(true), id).into()
     }
 
     // RPCAPI:
@@ -203,7 +200,7 @@ impl Darkfid {
             Ok(v) => v,
             Err(e) => {
                 error!("Failed fetching balances from wallet: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         };
 
@@ -242,7 +239,7 @@ impl Darkfid {
                     Ok(v) => v,
                     Err(e) => {
                         error!("Failed to decode_base10(): {}", e);
-                        return jsonrpc::error(InternalError, None, id).into()
+                        return JsonError::new(InternalError, None, id).into()
                     }
                 };
 
@@ -253,6 +250,6 @@ impl Darkfid {
             ret.insert(ticker, (amount, net_name.to_string(), net_addr, drk_addr));
         }
 
-        jsonrpc::response(json!(ret), id).into()
+        JsonResponse::new(json!(ret), id).into()
     }
 }

bin/dnetview/src/main.rs

@@ -17,7 +17,7 @@ use url::Url;
 
 use darkfi::{
     error::Result,
-    rpc::{jsonrpc, rpcclient::RpcClient},
+    rpc::{client::RpcClient, jsonrpc::JsonRequest},
     util::{
         async_util,
         cli::{log_config, spawn_config, Config},
@@ -50,14 +50,14 @@ impl DnetView {
     // --> {"jsonrpc": "2.0", "method": "ping", "params": [], "id": 42}
     // <-- {"jsonrpc": "2.0", "result": "pong", "id": 42}
     async fn _ping(&self) -> Result<Value> {
-        let req = jsonrpc::request(json!("ping"), json!([]));
+        let req = JsonRequest::new("ping", json!([]));
         self.rpc_client.request(req).await
     }
 
     //--> {"jsonrpc": "2.0", "method": "poll", "params": [], "id": 42}
     // <-- {"jsonrpc": "2.0", "result": {"nodeID": [], "nodeinfo" [], "id": 42}
     async fn get_info(&self) -> DnetViewResult<Value> {
-        let req = jsonrpc::request(json!("get_info"), json!([]));
+        let req = JsonRequest::new("get_info", json!([]));
         match self.rpc_client.request(req).await {
             Ok(req) => Ok(req),
             Err(e) => Err(DnetViewError::Darkfi(e)),

bin/drk/src/main.rs

@@ -9,7 +9,7 @@ use url::Url;
 use darkfi::{
     cli_desc,
     crypto::address::Address,
-    rpc::{jsonrpc, rpcclient::RpcClient},
+    rpc::{client::RpcClient, jsonrpc::JsonRequest},
     util::{cli::log_config, NetworkName},
     Result,
 };
@@ -99,7 +99,7 @@ impl Drk {
 
     async fn ping(&self) -> Result<()> {
         let start = Instant::now();
-        let req = jsonrpc::request(json!("ping"), json!([]));
+        let req = JsonRequest::new("ping", json!([]));
         let rep = self.rpc_client.request(req).await?;
         let latency = Instant::now() - start;
         println!("Got reply: {}", rep);
@@ -111,13 +111,13 @@ impl Drk {
         let addr = if address.is_some() {
             address.unwrap()
         } else {
-            let req = jsonrpc::request(json!("wallet.get_key"), json!([0_i64]));
+            let req = JsonRequest::new("wallet.get_key", json!([0_i64]));
             let rep = self.rpc_client.request(req).await?;
             Address::from_str(rep.as_array().unwrap()[0].as_str().unwrap())?
         };
 
         println!("Requesting airdrop for {}", addr);
-        let req = jsonrpc::request(json!("airdrop"), json!([json!(addr.to_string()), amount]));
+        let req = JsonRequest::new("airdrop", json!([json!(addr.to_string()), amount]));
         let rpc_client = RpcClient::new(endpoint).await?;
         let rep = rpc_client.request(req).await?;
         rpc_client.close().await?;
@@ -127,14 +127,14 @@ impl Drk {
     }
 
     async fn wallet_keygen(&self) -> Result<()> {
-        let req = jsonrpc::request(json!("wallet.keygen"), json!([]));
+        let req = JsonRequest::new("wallet.keygen", json!([]));
         let rep = self.rpc_client.request(req).await?;
         println!("New address: {}", rep);
         Ok(())
     }
 
     async fn wallet_balance(&self) -> Result<()> {
-        let req = jsonrpc::request(json!("wallet.get_balances"), json!([]));
+        let req = JsonRequest::new("wallet.get_balances", json!([]));
         let rep = self.rpc_client.request(req).await?;
         // TODO: Better representation
         println!("Balances:\n{:#?}", rep);
@@ -142,14 +142,14 @@ impl Drk {
     }
 
     async fn wallet_address(&self) -> Result<()> {
-        let req = jsonrpc::request(json!("wallet.get_key"), json!([0_i64]));
+        let req = JsonRequest::new("wallet.get_key", json!([0_i64]));
         let rep = self.rpc_client.request(req).await?;
         println!("Default wallet address: {}", rep);
         Ok(())
     }
 
     async fn wallet_all_addresses(&self) -> Result<()> {
-        let req = jsonrpc::request(json!("wallet.get_key"), json!([-1]));
+        let req = JsonRequest::new("wallet.get_key", json!([-1]));
         let rep = self.rpc_client.request(req).await?;
         println!("Wallet addresses:\n{:#?}", rep);
         Ok(())
@@ -164,8 +164,8 @@ impl Drk {
     ) -> Result<()> {
         println!("Attempting to transfer {} tokens to {}", amount, recipient);
 
-        let req = jsonrpc::request(
-            json!("tx.transfer"),
+        let req = JsonRequest::new(
+            "tx.transfer",
             json!([network.to_string(), token_id, recipient.to_string(), amount]),
         );
 

bin/faucetd/src/error.rs

@@ -1,9 +1,6 @@
 use serde_json::Value;
 
-use darkfi::rpc::{
-    jsonrpc,
-    jsonrpc::{ErrorCode::ServerError, JsonResult},
-};
+use darkfi::rpc::jsonrpc::{ErrorCode::ServerError, JsonError, JsonResult};
 
 pub enum RpcError {
     AmountExceedsLimit = -32107,
@@ -23,5 +20,5 @@ fn to_tuple(e: RpcError) -> (i64, String) {
 
 pub fn server_error(e: RpcError, id: Value) -> JsonResult {
     let (code, msg) = to_tuple(e);
-    jsonrpc::error(ServerError(code), Some(msg), id).into()
+    JsonError::new(ServerError(code), Some(msg), id).into()
 }

bin/faucetd/src/main.rs

@@ -28,10 +28,9 @@ use darkfi::{
     net::P2pPtr,
     node::Client,
     rpc::{
-        jsonrpc,
         jsonrpc::{
             ErrorCode::{InternalError, InvalidParams, MethodNotFound},
-            JsonRequest, JsonResult,
+            JsonError, JsonRequest, JsonResponse, JsonResult,
         },
         server::{listen_and_serve, RequestHandler},
     },
@@ -135,14 +134,14 @@ pub struct Faucetd {
 impl RequestHandler for Faucetd {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if !req.params.is_array() {
-            return jsonrpc::error(InvalidParams, None, req.id).into()
+            return JsonError::new(InvalidParams, None, req.id).into()
         }
 
         let params = req.params.as_array().unwrap();
 
         match req.method.as_str() {
             Some("airdrop") => return self.airdrop(req.id, params).await,
-            Some(_) | None => return jsonrpc::error(MethodNotFound, None, req.id).into(),
+            Some(_) | None => return JsonError::new(MethodNotFound, None, req.id).into(),
         }
     }
 }
@@ -174,12 +173,12 @@ impl Faucetd {
     // <-- {"jsonrpc": "2.0", "result": "txID", "id": 1}
     async fn airdrop(&self, id: Value, params: &[Value]) -> JsonResult {
         if params.len() != 2 || !params[0].is_string() || !params[1].is_f64() {
-            return jsonrpc::error(InvalidParams, None, id).into()
+            return JsonError::new(InvalidParams, None, id).into()
         }
 
         if !(*self.synced.lock().await) {
             error!("airdrop(): Blockchain is not yet synced");
-            return jsonrpc::error(InternalError, None, id).into()
+            return JsonError::new(InternalError, None, id).into()
         }
 
         let address = match Address::from_str(params[0].as_str().unwrap()) {
@@ -230,7 +229,7 @@ impl Faucetd {
             Ok(v) => v,
             Err(e) => {
                 error!("airdrop(): Failed converting biguint to u64: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         };
 
@@ -248,7 +247,7 @@ impl Faucetd {
             Ok(v) => v,
             Err(e) => {
                 error!("airdrop(): Failed building transaction: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         };
 
@@ -257,7 +256,7 @@ impl Faucetd {
             Ok(()) => {}
             Err(e) => {
                 error!("airdrop(): Failed broadcasting transaction: {}", e);
-                return jsonrpc::error(InternalError, None, id).into()
+                return JsonError::new(InternalError, None, id).into()
             }
         }
 
@@ -267,7 +266,7 @@ impl Faucetd {
         drop(map);
 
         let tx_hash = blake3::hash(&serialize(&tx)).to_hex().as_str().to_string();
-        jsonrpc::response(json!(tx_hash), id).into()
+        JsonResponse::new(json!(tx_hash), id).into()
     }
 }
 

bin/ircd/src/rpc.rs

@@ -6,8 +6,7 @@ use url::Url;
 use darkfi::{
     net,
     rpc::{
-        jsonrpc,
-        jsonrpc::{ErrorCode, JsonRequest, JsonResult},
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
         server::RequestHandler,
     },
 };
@@ -21,7 +20,7 @@ pub struct JsonRpcInterface {
 impl RequestHandler for JsonRpcInterface {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if req.params.as_array().is_none() {
-            return jsonrpc::error(ErrorCode::InvalidRequest, None, req.id).into()
+            return JsonError::new(ErrorCode::InvalidRequest, None, req.id).into()
         }
 
         debug!(target: "RPC", "--> {}", serde_json::to_string(&req).unwrap());
@@ -29,7 +28,7 @@ impl RequestHandler for JsonRpcInterface {
         match req.method.as_str() {
             Some("ping") => self.pong(req.id, req.params).await,
             Some("get_info") => self.get_info(req.id, req.params).await,
-            Some(_) | None => jsonrpc::error(ErrorCode::MethodNotFound, None, req.id).into(),
+            Some(_) | None => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
 }
@@ -40,7 +39,7 @@ impl JsonRpcInterface {
     // --> {"jsonrpc": "2.0", "method": "ping", "params": [], "id": 42}
     // <-- {"jsonrpc": "2.0", "result": "pong", "id": 42}
     async fn pong(&self, id: Value, _params: Value) -> JsonResult {
-        jsonrpc::response(json!("pong"), id).into()
+        JsonResponse::new(json!("pong"), id).into()
     }
 
     // RPCAPI:
@@ -49,6 +48,6 @@ impl JsonRpcInterface {
     // <-- {"jsonrpc": "2.0", result": {"nodeID": [], "nodeinfo": [], "id": 42}
     async fn get_info(&self, id: Value, _params: Value) -> JsonResult {
         let resp = self.p2p.get_info().await;
-        jsonrpc::response(resp, id).into()
+        JsonResponse::new(resp, id).into()
     }
 }

bin/tau/tau-cli/src/main.rs

@@ -5,7 +5,7 @@ use log::error;
 use simplelog::{ColorChoice, TermLogger, TerminalMode};
 use url::Url;
 
-use darkfi::{cli_desc, rpc::rpcclient::RpcClient, util::cli::log_config, Error, Result};
+use darkfi::{cli_desc, rpc::client::RpcClient, util::cli::log_config, Error, Result};
 
 mod filter;
 mod primitives;

bin/tau/tau-cli/src/rpc.rs

@@ -1,6 +1,6 @@
 use serde_json::json;
 
-use darkfi::{rpc::jsonrpc, Result};
+use darkfi::{rpc::jsonrpc::JsonRequest, Result};
 
 use crate::{
     primitives::{BaseTask, TaskInfo},
@@ -14,7 +14,7 @@ impl Tau {
 
     /// Add a new task.
     pub async fn add(&self, task: BaseTask) -> Result<()> {
-        let req = jsonrpc::request(json!("add"), json!([task]));
+        let req = JsonRequest::new("add", json!([task]));
         let rep = self.rpc_client.request(req).await?;
 
         println!("Got reply: {:?}", rep);
@@ -23,7 +23,7 @@ impl Tau {
 
     /// Get all task ids.
     pub async fn get_ids(&self) -> Result<Vec<u64>> {
-        let req = jsonrpc::request(json!("get_ids"), json!([]));
+        let req = JsonRequest::new("get_ids", json!([]));
         let rep = self.rpc_client.request(req).await?;
 
         let mut ret = vec![];
@@ -36,7 +36,7 @@ impl Tau {
 
     /// Update existing task given it's ID and some params.
     pub async fn update(&self, id: u64, task: BaseTask) -> Result<()> {
-        let req = jsonrpc::request(json!("update"), json!([id, task]));
+        let req = JsonRequest::new("update", json!([id, task]));
         let rep = self.rpc_client.request(req).await?;
 
         println!("Got reply: {:?}", rep);
@@ -45,7 +45,7 @@ impl Tau {
 
     /// Set the state for a task.
     pub async fn set_state(&self, id: u64, state: &str) -> Result<()> {
-        let req = jsonrpc::request(json!("set_state"), json!([id, state]));
+        let req = JsonRequest::new("set_state", json!([id, state]));
         let rep = self.rpc_client.request(req).await?;
 
         println!("Got reply: {:?}", rep);
@@ -54,7 +54,7 @@ impl Tau {
 
     /// Set a comment for a task.
     pub async fn set_comment(&self, id: u64, content: &str) -> Result<()> {
-        let req = jsonrpc::request(json!("set_comment"), json!([id, content]));
+        let req = JsonRequest::new("set_comment", json!([id, content]));
         let rep = self.rpc_client.request(req).await?;
 
         println!("Got reply: {:?}", rep);
@@ -63,7 +63,7 @@ impl Tau {
 
     /// Get task data by its ID.
     pub async fn get_task_by_id(&self, id: u64) -> Result<TaskInfo> {
-        let req = jsonrpc::request(json!("get_task_by_id"), json!([id]));
+        let req = JsonRequest::new("get_task_by_id", json!([id]));
         let rep = self.rpc_client.request(req).await?;
 
         Ok(serde_json::from_value(rep)?)

bin/tau/taud/src/error.rs

@@ -1,6 +1,6 @@
 use serde_json::Value;
 
-use darkfi::rpc::jsonrpc::{error as jsonerr, response as jsonresp, ErrorCode, JsonResult};
+use darkfi::rpc::jsonrpc::{ErrorCode, JsonError, JsonResponse, JsonResult};
 
 #[derive(Debug, thiserror::Error)]
 pub enum TaudError {
@@ -26,23 +26,19 @@ impl From<serde_json::Error> for TaudError {
 
 pub fn to_json_result(res: TaudResult<Value>, id: Value) -> JsonResult {
     match res {
-        Ok(v) => JsonResult::Resp(jsonresp(v, id)),
+        Ok(v) => JsonResponse::new(v, id).into(),
         Err(err) => match err {
-            TaudError::InvalidId => JsonResult::Err(jsonerr(
-                ErrorCode::InvalidParams,
-                Some("invalid task's id".into()),
-                id,
-            )),
+            TaudError::InvalidId => {
+                JsonError::new(ErrorCode::InvalidParams, Some("invalid task id".into()), id).into()
+            }
             TaudError::InvalidData(e) | TaudError::SerdeJsonError(e) => {
-                JsonResult::Err(jsonerr(ErrorCode::InvalidParams, Some(e), id))
+                JsonError::new(ErrorCode::InvalidParams, Some(e), id).into()
+            }
+            TaudError::InvalidDueTime => {
+                JsonError::new(ErrorCode::InvalidParams, Some("invalid due time".into()), id).into()
             }
-            TaudError::InvalidDueTime => JsonResult::Err(jsonerr(
-                ErrorCode::InvalidParams,
-                Some("invalid due time".into()),
-                id,
-            )),
             TaudError::Darkfi(e) => {
-                JsonResult::Err(jsonerr(ErrorCode::InternalError, Some(e.to_string()), id))
+                JsonError::new(ErrorCode::InternalError, Some(e.to_string()), id).into()
             }
         },
     }

bin/tau/taud/src/jsonrpc.rs

@@ -7,7 +7,7 @@ use serde_json::{json, Value};
 
 use darkfi::{
     rpc::{
-        jsonrpc::{error as jsonerr, ErrorCode, JsonRequest, JsonResult},
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult},
         server::RequestHandler,
     },
     util::Timestamp,
@@ -42,15 +42,13 @@ struct BaseTaskInfo {
 impl RequestHandler for JsonRpcInterface {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if req.params.as_array().is_none() {
-            return JsonResult::Err(jsonerr(ErrorCode::InvalidParams, None, req.id))
+            return JsonError::new(ErrorCode::InvalidParams, None, req.id).into()
         }
 
         if self.notify_queue_sender.send(None).await.is_err() {
-            return JsonResult::Err(jsonerr(ErrorCode::InternalError, None, req.id))
+            return JsonError::new(ErrorCode::InternalError, None, req.id).into()
         }
 
-        debug!(target: "RPC", "--> {}", serde_json::to_string(&req).unwrap());
-
         let rep = match req.method.as_str() {
             Some("add") => self.add(req.params).await,
             Some("get_ids") => self.get_ids(req.params).await,
@@ -58,9 +56,7 @@ impl RequestHandler for JsonRpcInterface {
             Some("set_state") => self.set_state(req.params).await,
             Some("set_comment") => self.set_comment(req.params).await,
             Some("get_task_by_id") => self.get_task_by_id(req.params).await,
-            Some(_) | None => {
-                return JsonResult::Err(jsonerr(ErrorCode::MethodNotFound, None, req.id))
-            }
+            Some(_) | None => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         };
 
         to_json_result(rep, req.id)

src/net/transport/tor.rs

@@ -156,8 +156,8 @@ impl TorTransport {
         }
     }
 
-    /// Query the environment for Tor variables, or fallback to defaults
-    pub fn get_env() -> Result<(Url, Url, String)> {
+    /// Query the environment for listener Tor variables, or fallback to defaults
+    pub fn get_listener_env() -> Result<(Url, Url, String)> {
         let socks5_url = Url::parse(
             &std::env::var("DARKFI_TOR_SOCKS5_URL")
                 .unwrap_or_else(|_| "socks5://127.0.0.1:9050".to_string()),
@@ -181,6 +181,14 @@ impl TorTransport {
         Ok((socks5_url, torc_url, auth_cookie.unwrap()))
     }
 
+    /// Query the environment for the dialer Tor variables, or fallback to defaults
+    pub fn get_dialer_env() -> Result<Url> {
+        Ok(Url::parse(
+            &std::env::var("DARKFI_TOR_SOCKS5_URL")
+                .unwrap_or_else(|_| "socks5://127.0.0.1:9050".to_string()),
+        )?)
+    }
+
     /// Creates an ephemeral hidden service pointing to local address, returns onion address
     /// when successful.
     ///

src/rpc/client.rs

@@ -0,0 +1,203 @@
+//! JSON-RPC client-side implementation.
+use std::time::Duration;
+
+use async_std::io::timeout;
+use futures::{select, AsyncReadExt, AsyncWriteExt, FutureExt};
+use log::{debug, error};
+use serde_json::{json, Value};
+use url::Url;
+
+use super::jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult};
+use crate::{
+    net::{
+        transport::Transport, TcpTransport, TorTransport, TransportName, TransportStream,
+        UnixTransport,
+    },
+    Error, Result,
+};
+
+/// JSON-RPC client implementation using asynchronous channels.
+pub struct RpcClient {
+    send: async_channel::Sender<Value>,
+    recv: async_channel::Receiver<JsonResult>,
+    stop_signal: async_channel::Sender<()>,
+    url: Url,
+}
+
+impl RpcClient {
+    /// Instantiate a new JSON-RPC client that will connect to the given URL.
+    pub async fn new(url: Url) -> Result<Self> {
+        let (send, recv, stop_signal) = Self::open_channels(&url).await?;
+        Ok(Self { send, recv, stop_signal, url })
+    }
+
+    /// Close the channels of an instantiated [`RpcClient`].
+    pub async fn close(&self) -> Result<()> {
+        self.stop_signal.send(()).await?;
+        Ok(())
+    }
+
+    /// Send a given JSON-RPC request over the instantiated client.
+    pub async fn request(&self, value: JsonRequest) -> Result<Value> {
+        let req_id = value.id.clone().as_u64().unwrap();
+
+        debug!(target: "jsonrpc-client", "--> {}", serde_json::to_string(&value)?);
+
+        // If the connection is closed, the sender will get an error for
+        // sending to a closed channel.
+        if let Err(e) = self.send.send(json!(value)).await {
+            error!("JSON-RPC client unable to send to {} (channels closed): {}", self.url, e);
+            return Err(Error::OperationFailed)
+        }
+
+        // If the connection is closed, the receiver will get an error for
+        // waiting on a closed channel.
+        let reply = self.recv.recv().await;
+        if reply.is_err() {
+            error!("JSON-RPC client unable to recv from {} (channels closed)", self.url);
+            return Err(Error::OperationFailed)
+        }
+
+        match reply? {
+            JsonResult::Response(r) => {
+                // Check if the IDs match
+                let resp_id = r.id.as_u64();
+                if resp_id.is_none() {
+                    let e = JsonError::new(ErrorCode::InvalidId, None, r.id);
+                    self.stop_signal.send(()).await?;
+                    return Err(Error::JsonRpcError(e.error.message.to_string()))
+                }
+
+                if resp_id.unwrap() != req_id {
+                    let e = JsonError::new(ErrorCode::InvalidId, None, r.id);
+                    self.stop_signal.send(()).await?;
+                    return Err(Error::JsonRpcError(e.error.message.to_string()))
+                }
+
+                debug!(target: "jsonrpc-client", "<-- {}", serde_json::to_string(&r)?);
+                Ok(r.result)
+            }
+            JsonResult::Error(e) => {
+                debug!(target: "jsonrpc-client", "<-- {}", serde_json::to_string(&e)?);
+                // Close the server connection
+                self.stop_signal.send(()).await?;
+                Err(Error::JsonRpcError(e.error.message.to_string()))
+            }
+            JsonResult::Notification(n) => {
+                debug!(target: "jsonrpc-client", "<-- {}", serde_json::to_string(&n)?);
+                // Close the server connection
+                self.stop_signal.send(()).await?;
+                Err(Error::JsonRpcError("Unexpected reply".to_string()))
+            }
+        }
+    }
+
+    /// Oneshot send a given JSON-RPC request over the instantiated client
+    /// and close the channels on reply.
+    pub async fn oneshot_request(&self, value: JsonRequest) -> Result<Value> {
+        let rep = self.request(value).await?;
+        self.stop_signal.send(()).await?;
+        Ok(rep)
+    }
+
+    /// Instantiate channels for a new [`RpcClient`].
+    async fn open_channels(
+        uri: &Url,
+    ) -> Result<(
+        async_channel::Sender<Value>,
+        async_channel::Receiver<JsonResult>,
+        async_channel::Sender<()>,
+    )> {
+        let (data_send, data_recv) = async_channel::unbounded();
+        let (result_send, result_recv) = async_channel::unbounded();
+        let (stop_send, stop_recv) = async_channel::unbounded();
+
+        let transport_name = TransportName::try_from(uri.clone())?;
+
+        macro_rules! reqrep {
+            ($stream:expr, $transport:expr, $upgrade:expr) => {{
+                if let Err(err) = $stream {
+                    error!("JSON-RPC client setup for {} failed: {}", uri, err);
+                    return Err(Error::ConnectFailed)
+                }
+
+                let stream = $stream?.await;
+                if let Err(err) = stream {
+                    error!("JSON-RPC client connection to {} failed: {}", uri, err);
+                    return Err(Error::ConnectFailed)
+                }
+
+                let stream = stream?;
+                match $upgrade {
+                    None => {
+                        smol::spawn(Self::reqrep_loop(stream, result_send, data_recv, stop_recv))
+                            .detach();
+                    }
+                    Some(u) if u == "tls" => {
+                        let stream = $transport.upgrade_dialer(stream)?.await?;
+                        smol::spawn(Self::reqrep_loop(stream, result_send, data_recv, stop_recv))
+                            .detach();
+                    }
+                    Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)),
+                }
+            }};
+        }
+
+        match transport_name {
+            TransportName::Tcp(upgrade) => {
+                let transport = TcpTransport::new(None, 1024);
+                let stream = transport.dial(uri.clone(), None);
+                reqrep!(stream, transport, upgrade);
+            }
+            TransportName::Tor(upgrade) => {
+                let socks5_url = TorTransport::get_dialer_env()?;
+                let transport = TorTransport::new(socks5_url, None)?;
+                let stream = transport.clone().dial(uri.clone(), None);
+                reqrep!(stream, transport, upgrade);
+            }
+            TransportName::Unix => {
+                let transport = UnixTransport::new();
+                let stream = transport.dial(uri.clone()).await;
+                if let Err(err) = stream {
+                    error!("JSON-RPC client connection to {} failed: {}", uri, err);
+                    return Err(Error::ConnectFailed)
+                }
+
+                smol::spawn(Self::reqrep_loop(stream?, result_send, data_recv, stop_recv)).detach();
+            }
+            _ => unimplemented!(),
+        }
+
+        Ok((data_send, result_recv, stop_send))
+    }
+
+    /// Internal function that loops on a given stream and multiplexes the data.
+    async fn reqrep_loop<T: TransportStream>(
+        mut stream: T,
+        result_send: async_channel::Sender<JsonResult>,
+        data_recv: async_channel::Receiver<Value>,
+        stop_recv: async_channel::Receiver<()>,
+    ) -> Result<()> {
+        // If we don't get a reply within 30 seconds, we'll fail.
+        let read_timeout = Duration::from_secs(30);
+
+        loop {
+            // Nasty size
+            let mut buf = vec![0; 2048 * 10];
+
+            select! {
+                data = data_recv.recv().fuse() => {
+                    let data_bytes = serde_json::to_vec(&data?)?;
+                    stream.write_all(&data_bytes).await?;
+                    let n = timeout(read_timeout, async { stream.read(&mut buf[..]).await }).await?;
+                    let reply: JsonResult = serde_json::from_slice(&buf[0..n])?;
+                    result_send.send(reply).await?;
+                }
+
+                _ = stop_recv.recv().fuse() => break
+            }
+        }
+
+        Ok(())
+    }
+}

src/rpc/jsonrpc.rs

@@ -1,18 +1,10 @@
-use std::{env, str, time::Duration};
-
-use async_std::io::timeout;
-use futures::{select, AsyncReadExt, AsyncWriteExt, FutureExt};
-use log::error;
+//! JSON-RPC 2.0 primitives
 use rand::Rng;
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
-use url::Url;
-
-use crate::{
-    net::{TcpTransport, TorTransport, Transport, TransportName, TransportStream, UnixTransport},
-    Error, Result,
-};
 
+/// JSON-RPC error codes.
+/// The error codes from and including -32768 to -32000 are reserved for pre-defined errors.
 #[derive(Debug, Clone)]
 pub enum ErrorCode {
     ParseError,
@@ -20,348 +12,153 @@ pub enum ErrorCode {
     MethodNotFound,
     InvalidParams,
     InternalError,
-    KeyGenError,
-    GetAddressesError,
-    ImportAndExportFile,
-    SetDefaultAddress,
-    InvalidAmountParam,
-    InvalidNetworkParam,
-    InvalidTokenIdParam,
-    InvalidAddressParam,
-    InvalidSymbolParam,
-    InvalidId,
     ServerError(i64),
+    InvalidId,
 }
 
 impl ErrorCode {
     pub fn code(&self) -> i64 {
         match *self {
-            ErrorCode::ParseError => -32700,
-            ErrorCode::InvalidRequest => -32600,
-            ErrorCode::MethodNotFound => -32601,
-            ErrorCode::InvalidParams => -32602,
-            ErrorCode::InternalError => -32603,
-            ErrorCode::KeyGenError => -32002,
-            ErrorCode::GetAddressesError => -32003,
-            ErrorCode::ImportAndExportFile => -32004,
-            ErrorCode::SetDefaultAddress => -32005,
-            ErrorCode::InvalidAmountParam => -32010,
-            ErrorCode::InvalidNetworkParam => -32011,
-            ErrorCode::InvalidTokenIdParam => -32012,
-            ErrorCode::InvalidAddressParam => -32013,
-            ErrorCode::InvalidSymbolParam => -32014,
-            ErrorCode::InvalidId => -32030,
-            ErrorCode::ServerError(c) => c,
+            Self::ParseError => -32700,
+            Self::InvalidRequest => -32600,
+            Self::MethodNotFound => -32601,
+            Self::InvalidParams => -32602,
+            Self::InternalError => -32603,
+            // -32000 to -32099
+            Self::ServerError(c) => c,
+            Self::InvalidId => -32001,
         }
     }
 
-    pub fn description(&self) -> String {
+    pub fn desc(&self) -> String {
         let desc = match *self {
-            ErrorCode::ParseError => "Parse error",
-            ErrorCode::InvalidRequest => "Invalid request",
-            ErrorCode::MethodNotFound => "Method not found",
-            ErrorCode::InvalidParams => "Invalid params",
-            ErrorCode::InternalError => "Internal error",
-            ErrorCode::KeyGenError => "Key gen error",
-            ErrorCode::GetAddressesError => "get addresses error",
-            ErrorCode::ImportAndExportFile => "error import/export a file",
-            ErrorCode::SetDefaultAddress => "error set default address",
-            ErrorCode::InvalidAmountParam => "Invalid amount param",
-            ErrorCode::InvalidNetworkParam => "Invalid network param",
-            ErrorCode::InvalidTokenIdParam => "Invalid token id param",
-            ErrorCode::InvalidAddressParam => "Invalid address param",
-            ErrorCode::InvalidSymbolParam => "Invalid symbol param",
-            ErrorCode::InvalidId => "Invalid Id",
-            ErrorCode::ServerError(_) => "Server error",
+            Self::ParseError => "Parse error",
+            Self::InvalidRequest => "Invalid request",
+            Self::MethodNotFound => "Method not found",
+            Self::InvalidParams => "Invalid params",
+            Self::InternalError => "Internal error",
+            Self::ServerError(_) => "",
+            Self::InvalidId => "Request ID mismatch",
         };
+
         desc.to_string()
     }
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug)]
+/// Wrapping enum around the possible JSON-RPC object types.
+#[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(untagged)]
 pub enum JsonResult {
-    Resp(JsonResponse),
-    Err(JsonError),
-    Notif(JsonNotification),
+    Response(JsonResponse),
+    Error(JsonError),
+    Notification(JsonNotification),
 }
 
 impl From<JsonResponse> for JsonResult {
     fn from(resp: JsonResponse) -> Self {
-        Self::Resp(resp)
+        Self::Response(resp)
     }
 }
 
 impl From<JsonError> for JsonResult {
     fn from(err: JsonError) -> Self {
-        Self::Err(err)
+        Self::Error(err)
     }
 }
 
 impl From<JsonNotification> for JsonResult {
     fn from(notif: JsonNotification) -> Self {
-        Self::Notif(notif)
+        Self::Notification(notif)
     }
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug)]
+/// A JSON-RPC request object.
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct JsonRequest {
+    /// JSON-RPC version
     pub jsonrpc: Value,
+    /// Request ID
+    pub id: Value,
+    /// Request method
     pub method: Value,
+    /// Request parameters
     pub params: Value,
-    pub id: Value,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct JsonErrorVal {
-    pub code: Value,
-    pub message: Value,
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct JsonError {
-    pub jsonrpc: Value,
-    pub error: JsonErrorVal,
-    pub id: Value,
-}
+impl JsonRequest {
+    pub fn new(method: &str, parameters: Value) -> Self {
+        let mut rng = rand::thread_rng();
 
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct JsonResponse {
-    pub jsonrpc: Value,
-    pub result: Value,
-    pub id: Value,
+        Self {
+            jsonrpc: json!("2.0"),
+            id: json!(rng.gen::<u64>()),
+            method: json!(method),
+            params: parameters,
+        }
+    }
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug)]
+/// A JSON-RPC notification object.
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct JsonNotification {
+    /// JSON-RPC version
     pub jsonrpc: Value,
+    /// Notification method
     pub method: Value,
+    /// Notification parameters
     pub params: Value,
 }
 
-pub fn request(m: Value, p: Value) -> JsonRequest {
-    let mut rng = rand::thread_rng();
-
-    JsonRequest { jsonrpc: json!("2.0"), method: m, params: p, id: json!(rng.gen::<u32>()) }
-}
-
-pub fn response(r: Value, i: Value) -> JsonResponse {
-    JsonResponse { jsonrpc: json!("2.0"), result: r, id: i }
-}
-
-pub fn error(c: ErrorCode, m: Option<String>, i: Value) -> JsonError {
-    let ev = JsonErrorVal {
-        code: json!(c.code()),
-        message: if m.is_none() { json!(c.description()) } else { json!(Some(m)) },
-    };
-
-    JsonError { jsonrpc: json!("2.0"), error: ev, id: i }
-}
-
-pub fn notification(m: Value, p: Value) -> JsonNotification {
-    JsonNotification { jsonrpc: json!("2.0"), method: m, params: p }
-}
-
-async fn reqrep_loop<T: TransportStream>(
-    mut stream: T,
-    result_sender: async_channel::Sender<JsonResult>,
-    data_receiver: async_channel::Receiver<Value>,
-    stop_receiver: async_channel::Receiver<()>,
-) -> Result<()> {
-    // If we don't get a reply after 30 seconds, we'll fail.
-    let read_timeout = Duration::from_secs(30);
-
-    loop {
-        let mut buf = [0; 8192];
-
-        select! {
-            data = data_receiver.recv().fuse()  => {
-                let data_str = serde_json::to_string(&data?)?;
-
-                stream.write_all(data_str.as_bytes()).await?;
-
-                let bytes_read = timeout(read_timeout, async { stream.read(&mut buf[..]).await }).await?;
-
-                let reply: JsonResult = serde_json::from_slice(&buf[0..bytes_read])?;
-
-                result_sender.send(reply).await?;
-            }
-
-            _ = stop_receiver.recv().fuse() => break
-        }
+impl JsonNotification {
+    pub fn new(method: &str, parameters: Value) -> Self {
+        Self { jsonrpc: json!("2.0"), method: json!(method), params: parameters }
     }
-
-    Ok(())
 }
 
-pub async fn open_channels(
-    uri: &Url,
-) -> Result<(
-    async_channel::Sender<Value>,
-    async_channel::Receiver<JsonResult>,
-    async_channel::Sender<()>,
-)> {
-    let (data_sender, data_receiver) = async_channel::unbounded();
-    let (result_sender, result_receiver) = async_channel::unbounded();
-    let (stop_sender, stop_receiver) = async_channel::unbounded();
-
-    let transport_name = TransportName::try_from(uri.clone())?;
-
-    macro_rules! reqrep {
-        ($stream:expr, $transport:expr, $upgrade:expr) => {{
-            if let Err(err) = $stream {
-                error!("RPC Setup for {} failed: {}", uri, err);
-                return Err(Error::ConnectFailed)
-            }
-
-            let stream = $stream?.await;
-
-            if let Err(err) = stream {
-                error!("RPC Connection to {} failed: {}", uri, err);
-                return Err(Error::ConnectFailed)
-            }
-
-            let stream = stream?;
-
-            match $upgrade {
-                None => {
-                    smol::spawn(reqrep_loop(stream, result_sender, data_receiver, stop_receiver))
-                        .detach();
-                }
-                Some(u) if u == "tls" => {
-                    let stream = $transport.upgrade_dialer(stream)?.await?;
-                    smol::spawn(reqrep_loop(stream, result_sender, data_receiver, stop_receiver))
-                        .detach();
-                }
-                Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)),
-            }
-        }};
-    }
-
-    match transport_name {
-        TransportName::Tcp(upgrade) => {
-            let transport = TcpTransport::new(None, 1024);
-            let stream = transport.dial(uri.clone(), None);
-
-            reqrep!(stream, transport, upgrade);
-        }
-        TransportName::Tor(upgrade) => {
-            let socks5_url = Url::parse(
-                &env::var("DARKFI_TOR_SOCKS5_URL")
-                    .unwrap_or_else(|_| "socks5://127.0.0.1:9050".to_string()),
-            )?;
-
-            let transport = TorTransport::new(socks5_url, None)?;
-
-            let stream = transport.clone().dial(uri.clone(), None);
-
-            reqrep!(stream, transport, upgrade);
-        }
-        TransportName::Unix => {
-            let transport = UnixTransport::new();
-
-            let stream = transport.dial(uri.clone()).await;
-
-            if let Err(err) = stream {
-                error!("RPC Connection to {}  failed: {}", uri, err);
-                return Err(Error::ConnectFailed)
-            }
-
-            smol::spawn(reqrep_loop(stream?, result_sender, data_receiver, stop_receiver)).detach();
-        }
-        _ => unimplemented!(),
-    }
-
-    Ok((data_sender, result_receiver, stop_sender))
+/// A JSON-RPC response object.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct JsonResponse {
+    /// JSON-RPC version
+    pub jsonrpc: Value,
+    /// Request ID
+    pub id: Value,
+    /// Response result
+    pub result: Value,
 }
 
-pub async fn send_request(uri: &Url, data: Value) -> Result<JsonResult> {
-    let data_str = serde_json::to_string(&data)?;
-
-    let transport_name = TransportName::try_from(uri.clone())?;
-
-    macro_rules! reply {
-        ($stream:expr, $transport:expr, $upgrade:expr) => {{
-            if let Err(err) = $stream {
-                error!("RPC Setup for {} failed: {}", uri, err);
-                return Err(Error::ConnectFailed)
-            }
-
-            let stream = $stream?.await;
-
-            if let Err(err) = stream {
-                error!("RPC Connection to {} failed: {}", uri, err);
-                return Err(Error::ConnectFailed)
-            }
-
-            let stream = stream?;
-
-            match $upgrade {
-                None => get_reply(stream, data_str).await,
-                Some(u) if u == "tls" => {
-                    let stream = $transport.upgrade_dialer(stream)?.await?;
-                    get_reply(stream, data_str).await
-                }
-                Some(u) => Err(Error::UnsupportedTransportUpgrade(u)),
-            }
-        }};
-    }
-
-    match transport_name {
-        TransportName::Tcp(upgrade) => {
-            let transport = TcpTransport::new(None, 1024);
-            let stream = transport.dial(uri.clone(), None);
-
-            reply!(stream, transport, upgrade)
-        }
-        TransportName::Tor(upgrade) => {
-            let socks5_url = Url::parse(
-                &env::var("DARKFI_TOR_SOCKS5_URL")
-                    .unwrap_or_else(|_| "socks5://127.0.0.1:9050".to_string()),
-            )?;
-
-            let transport = TorTransport::new(socks5_url, None)?;
-
-            let stream = transport.clone().dial(uri.clone(), None);
-
-            reply!(stream, transport, upgrade)
-        }
-        TransportName::Unix => {
-            let transport = UnixTransport::new();
-
-            let stream = transport.dial(uri.clone()).await;
-
-            if let Err(err) = stream {
-                error!("RPC Connection to {}  failed: {}", uri, err);
-                return Err(Error::ConnectFailed)
-            }
-
-            get_reply(stream?, data_str).await
-        }
-        _ => unimplemented!(),
+impl JsonResponse {
+    pub fn new(result: Value, id: Value) -> Self {
+        Self { jsonrpc: json!("2.0"), id, result }
     }
 }
 
-async fn get_reply<T: TransportStream>(mut stream: T, data_str: String) -> Result<JsonResult> {
-    // If we don't get a reply after 30 seconds, we'll fail.
-    let read_timeout = Duration::from_secs(30);
-
-    let mut buf = [0; 8192];
-
-    stream.write_all(data_str.as_bytes()).await?;
-
-    let bytes_read = timeout(read_timeout, async { stream.read(&mut buf[..]).await }).await?;
+/// A JSON-RPC error object.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct JsonError {
+    /// JSON-RPC version
+    pub jsonrpc: Value,
+    /// Request ID
+    pub id: Value,
+    /// JSON-RPC error (code and message)
+    pub error: JsonErrorVal,
+}
 
-    let reply: JsonResult = serde_json::from_slice(&buf[0..bytes_read])?;
-    Ok(reply)
+/// A JSON-RPC error value (code and message)
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct JsonErrorVal {
+    /// Error code
+    pub code: Value,
+    /// Error message
+    pub message: Value,
 }
 
-// Utils to quickly handle errors
-pub type ValueResult<Value> = std::result::Result<Value, ErrorCode>;
+impl JsonError {
+    pub fn new(c: ErrorCode, m: Option<String>, id: Value) -> Self {
+        let error = JsonErrorVal {
+            code: json!(c.code()),
+            message: if m.is_none() { json!(c.desc()) } else { json!(m.unwrap()) },
+        };
 
-pub fn from_result(res: ValueResult<Value>, id: Value) -> JsonResult {
-    match res {
-        Ok(v) => JsonResult::Resp(response(v, id)),
-        Err(e) => error(e, None, id).into(),
+        Self { jsonrpc: json!("2.0"), error, id }
     }
 }

src/rpc/mod.rs

@@ -1,4 +1,11 @@
+/// JSON-RPC primitives
 pub mod jsonrpc;
-pub mod rpcclient;
+
+/// Client-side JSON-RPC implementation
+pub mod client;
+
+/// Server-side JSON-RPC implementation
 pub mod server;
+
+/// Websockets client
 pub mod websockets;

src/rpc/rpcclient.rs

@@ -1,86 +0,0 @@
-use log::{debug, error};
-use serde_json::{json, Value};
-use url::Url;
-
-use crate::{Error, Result};
-
-use super::jsonrpc::{self, ErrorCode, JsonRequest, JsonResult};
-
-pub struct RpcClient {
-    sender: async_channel::Sender<Value>,
-    receiver: async_channel::Receiver<JsonResult>,
-    stop_signal: async_channel::Sender<()>,
-    url: Url,
-}
-
-impl RpcClient {
-    pub async fn new(url: Url) -> Result<Self> {
-        let (sender, receiver, stop_signal) = jsonrpc::open_channels(&url).await?;
-        Ok(Self { sender, receiver, stop_signal, url })
-    }
-
-    pub async fn close(&self) -> Result<()> {
-        self.stop_signal.send(()).await?;
-        Ok(())
-    }
-
-    pub async fn request(&self, value: JsonRequest) -> Result<Value> {
-        let req_id = value.id.clone().as_u64().unwrap_or(0);
-        let value = json!(value);
-
-        // if the connection is closed the sender will get an error
-        // for sending to closed channel
-        let result = self.sender.send(value).await;
-        if result.is_err() {
-            error!("Unable to send to the RPC server: {}", self.url);
-            return Err(Error::OperationFailed)
-        }
-
-        let reply = self.receiver.recv().await;
-
-        // if the connection is closed the receiver will get an error
-        // for waiting closed channel
-        if reply.is_err() {
-            error!("Unable to receive from the RPC server: {}", self.url);
-            return Err(Error::OperationFailed)
-        }
-
-        match reply? {
-            JsonResult::Resp(r) => {
-                // check if the ids match
-                let resp_id = r.id.as_u64();
-
-                if resp_id.is_none() {
-                    let error = jsonrpc::error(ErrorCode::InvalidId, None, r.id);
-                    self.stop_signal.send(()).await?;
-                    return Err(Error::JsonRpcError(error.error.message.to_string()))
-                }
-
-                if resp_id.unwrap() != req_id {
-                    let error = jsonrpc::error(
-                        ErrorCode::InvalidId,
-                        Some("Ids doesn't match".into()),
-                        r.id,
-                    );
-                    self.stop_signal.send(()).await?;
-                    return Err(Error::JsonRpcError(error.error.message.to_string()))
-                }
-
-                debug!(target: "RPC", "<-- {}", serde_json::to_string(&r)?);
-                Ok(r.result)
-            }
-
-            JsonResult::Err(e) => {
-                debug!(target: "RPC", "<-- {}", serde_json::to_string(&e)?);
-                // close the server connection
-                self.stop_signal.send(()).await?;
-                Err(Error::JsonRpcError(e.error.message.to_string()))
-            }
-
-            JsonResult::Notif(n) => {
-                debug!(target: "RPC", "<-- {}", serde_json::to_string(&n)?);
-                Err(Error::JsonRpcError("Unexpected reply".to_string()))
-            }
-        }
-    }
-}

src/rpc/server.rs

@@ -131,7 +131,7 @@ pub async fn listen_and_serve(
             accept!(listener, transport, upgrade);
         }
         TransportName::Tor(upgrade) => {
-            let (socks5_url, torc_url, auth_cookie) = TorTransport::get_env()?;
+            let (socks5_url, torc_url, auth_cookie) = TorTransport::get_listener_env()?;
             let auth_cookie = hex::encode(&std::fs::read(auth_cookie).unwrap());
             let transport = TorTransport::new(socks5_url, Some((torc_url, auth_cookie)))?;