converted to async/.await to support integration testing framework

(cherry picked from commit 26c129fa787ffec44f229ad8047d5c375fb4d257)

src/bin/mwixnet.rs

@@ -1,34 +1,28 @@
-use config::ServerConfig;
-use node::HttpGrinNode;
-use store::SwapStore;
-use wallet::HttpWallet;
+use mwixnet::config::{self, ServerConfig};
+use mwixnet::node::HttpGrinNode;
+use mwixnet::servers;
+use mwixnet::store::SwapStore;
+use mwixnet::tor;
+use mwixnet::wallet::HttpWallet;
 
-use crate::client::{MixClient, MixClientImpl};
-use crate::node::GrinNode;
-use crate::store::StoreError;
 use clap::App;
 use grin_core::global;
 use grin_core::global::ChainTypes;
 use grin_onion::crypto;
 use grin_onion::crypto::dalek::DalekPublicKey;
 use grin_util::{StopState, ZeroingString};
+use mwixnet::client::{MixClient, MixClientImpl};
+use mwixnet::node::GrinNode;
+use mwixnet::store::StoreError;
 use rpassword;
 use std::path::PathBuf;
 use std::sync::Arc;
-use tokio::runtime::Runtime;
+use std::thread::{sleep, spawn};
+use std::time::Duration;
 
 #[macro_use]
 extern crate clap;
 
-mod client;
-mod config;
-mod node;
-mod servers;
-mod store;
-mod tor;
-mod tx;
-mod wallet;
-
 const DEFAULT_INTERVAL: u32 = 12 * 60 * 60;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -37,7 +31,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 }
 
 fn real_main() -> Result<(), Box<dyn std::error::Error>> {
-	let yml = load_yaml!("../mwixnet.yml");
+	let yml = load_yaml!("mwixnet.yml");
 	let args = App::from_yaml(yml).get_matches();
 	let chain_type = if args.is_present("testnet") {
 		ChainTypes::Testnet
@@ -168,18 +162,22 @@ fn real_main() -> Result<(), Box<dyn std::error::Error>> {
 	);
 
 	// Node API health check
-	if let Err(e) = node.get_chain_height() {
+	let mut rt = tokio::runtime::Builder::new()
+		.threaded_scheduler()
+		.enable_all()
+		.build()?;
+	if let Err(e) = rt.block_on(node.async_get_chain_height()) {
 		eprintln!("Node communication failure. Is node listening?");
 		return Err(e.into());
 	};
 
 	// Open wallet
 	let wallet_pass = prompt_wallet_password(&args.value_of("wallet_pass"));
-	let wallet = HttpWallet::open_wallet(
+	let wallet = rt.block_on(HttpWallet::async_open_wallet(
 		&server_config.wallet_owner_url,
 		&server_config.wallet_owner_api_secret(),
 		&wallet_pass,
-	);
+	));
 	let wallet = match wallet {
 		Ok(w) => w,
 		Err(e) => {
@@ -188,12 +186,14 @@ fn real_main() -> Result<(), Box<dyn std::error::Error>> {
 		}
 	};
 
-	let mut tor_process = tor::init_tor_listener(&server_config)?;
+	let mut tor_process = tor::init_tor_listener(
+		&config::get_grin_path(&chain_type).to_str().unwrap(),
+		&server_config,
+	)?;
 
 	let stop_state = Arc::new(StopState::new());
 	let stop_state_clone = stop_state.clone();
 
-	let rt = Runtime::new()?;
 	rt.spawn(async move {
 		futures::executor::block_on(build_signals_fut());
 		let _ = tor_process.kill();
@@ -213,13 +213,26 @@ fn real_main() -> Result<(), Box<dyn std::error::Error>> {
 			server_config.server_pubkey().to_hex()
 		);
 
-		servers::mix_rpc::listen(
+		let (_, http_server) = servers::mix_rpc::listen(
+			rt.handle(),
 			server_config,
 			next_mixer,
 			Arc::new(wallet),
 			Arc::new(node),
-			stop_state,
-		)
+		)?;
+
+		let close_handle = http_server.close_handle();
+		let round_handle = spawn(move || loop {
+			if stop_state.is_stopped() {
+				close_handle.close();
+				break;
+			}
+
+			sleep(Duration::from_millis(100));
+		});
+
+		http_server.wait();
+		round_handle.join().unwrap();
 	} else {
 		println!(
 			"Starting SWAP server with public key {:?}",
@@ -237,15 +250,40 @@ fn real_main() -> Result<(), Box<dyn std::error::Error>> {
 		)?;
 
 		// Start the mwixnet JSON-RPC HTTP 'swap' server
-		servers::swap_rpc::listen(
-			server_config,
+		let (swap_server, http_server) = servers::swap_rpc::listen(
+			rt.handle(),
+			&server_config,
 			next_mixer,
 			Arc::new(wallet),
 			Arc::new(node),
 			store,
-			stop_state,
-		)
+		)?;
+
+		let close_handle = http_server.close_handle();
+		let round_handle = spawn(move || {
+			let mut secs = 0;
+			loop {
+				if stop_state.is_stopped() {
+					close_handle.close();
+					break;
+				}
+
+				sleep(Duration::from_secs(1));
+				secs = (secs + 1) % server_config.interval_s;
+
+				if secs == 0 {
+					let server = swap_server.clone();
+					rt.spawn(async move { server.lock().await.execute_round().await });
+					//let _ = swap_server.lock().unwrap().execute_round();
+				}
+			}
+		});
+
+		http_server.wait();
+		round_handle.join().unwrap();
 	}
+
+	Ok(())
 }
 
 #[cfg(unix)]

src/bin/mwixnet.yml

@@ -0,0 +1,55 @@
+name: mwixnet
+about: MWixnet CoinSwap Server
+author: scilio
+
+args:
+  - config_file:
+      help: Path to load/save the mwixnet-config.toml configuration file
+      short: c
+      long: config_file
+      takes_value: true
+  - testnet:
+      help: Run grin against the Testnet (as opposed to mainnet)
+      long: testnet
+      takes_value: false
+  - grin_node_url:
+      help: Api address of running GRIN node on which to check inputs and post transactions
+      short: n
+      long: grin_node_url
+      takes_value: true
+  - grin_node_secret_path:
+      help: Path to a file containing the secret for the GRIN node api
+      long: grin_node_secret_path
+      takes_value: true
+  - wallet_owner_url:
+      help: Api address of running wallet owner listener
+      short: l
+      long: wallet_owner_url
+      takes_value: true
+  - wallet_owner_secret_path:
+      help: Path to a file containing the secret for the wallet owner api
+      long: wallet_owner_secret_path
+      takes_value: true
+  - wallet_pass:
+      help: The wallet's password
+      long: wallet_pass
+      takes_value: true
+  - bind_addr:
+      help: Address to bind the rpc server to (e.g. 127.0.0.1:3000)
+      long: bind_addr
+      takes_value: true
+  - socks_addr:
+      help: Address to bind the SOCKS5 tor proxy to (e.g. 127.0.0.1:3001)
+      long: socks_addr
+      takes_value: true
+  - prev_server:
+      help: Hex public key of the previous swap/mix server
+      long: prev_server
+      takes_value: true
+  - next_server:
+      help: Hex public key of the next mix server
+      long: next_server
+      takes_value: true
+subcommands:
+  - init-config:
+      about: Writes a new configuration file

src/client.rs

@@ -1,17 +1,17 @@
 use crate::config::ServerConfig;
-use crate::crypto::dalek;
-use crate::servers::mix_rpc::MixReq;
-use crate::tx::TxComponents;
-use crate::{tor, DalekPublicKey};
+use crate::servers::mix_rpc::{MixReq, MixResp};
+use crate::tor;
+use grin_onion::crypto::dalek::{self, DalekPublicKey};
 use grin_onion::onion::Onion;
 
-use grin_api::client;
-use grin_api::json_rpc::build_request;
+use async_trait::async_trait;
+use grin_api::json_rpc::{build_request, Response};
 use grin_core::ser;
 use grin_core::ser::ProtocolVersion;
 use grin_wallet_util::OnionV3Address;
 use hyper::client::HttpConnector;
-use hyper_proxy::{Intercept, Proxy, ProxyConnector};
+use hyper::header::{ACCEPT, CONTENT_TYPE, USER_AGENT};
+use hyper_socks2::SocksConnector;
 use serde_json;
 use thiserror::Error;
 
@@ -24,16 +24,19 @@ pub enum ClientError {
 	API(grin_api::Error),
 	#[error("Dalek Error: {0:?}")]
 	Dalek(dalek::DalekError),
-	#[error("Error parsing response: {0:?}")]
-	ResponseParse(serde_json::Error),
+	#[error("Error decoding JSON response: {0:?}")]
+	DecodeResponseError(serde_json::Error),
+	#[error("Error in JSON-RPC response: {0:?}")]
+	ResponseError(grin_api::json_rpc::RpcError),
 	#[error("Custom client error: {0:?}")]
 	Custom(String),
 }
 
 /// A client for consuming a mix API
+#[async_trait]
 pub trait MixClient: Send + Sync {
 	/// Swaps the outputs provided and returns the final swapped outputs and kernels.
-	fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<(Vec<usize>, TxComponents), ClientError>;
+	async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError>;
 }
 
 pub struct MixClientImpl {
@@ -47,63 +50,104 @@ impl MixClientImpl {
 		MixClientImpl { config, addr }
 	}
 
-	fn send_json_request<D: serde::de::DeserializeOwned>(
+	async fn async_send_json_request<D: serde::de::DeserializeOwned>(
 		&self,
 		addr: &OnionV3Address,
 		method: &str,
 		params: &serde_json::Value,
 	) -> Result<D, ClientError> {
-		let _tor = tor::init_tor_sender(&self.config).map_err(ClientError::Tor)?;
-
 		let proxy = {
-			let proxy_uri = format!("http://{:?}", self.config.socks_proxy_addr)
-				.parse()
-				.unwrap();
-			let proxy = Proxy::new(Intercept::All, proxy_uri);
-			//proxy.set_authorization(Authorization::basic("John Doe", "Agent1234"));
-			let connector = HttpConnector::new();
-			let proxy_connector = ProxyConnector::from_proxy(connector, proxy).unwrap();
+			let proxy_uri = format!(
+				"socks5://{}:{}",
+				self.config.socks_proxy_addr.ip(),
+				self.config.socks_proxy_addr.port()
+			)
+			.parse()
+			.unwrap();
+			let mut connector = HttpConnector::new();
+			connector.enforce_http(false);
+			let proxy_connector = SocksConnector {
+				proxy_addr: proxy_uri,
+				auth: None,
+				connector,
+			};
 			proxy_connector
 		};
 
 		let url = format!("{}/v1", addr.to_http_str());
-		let mut req = client::create_post_request(&url, None, &build_request(method, params))
-			.map_err(ClientError::API)?;
 
-		let uri = url.parse().unwrap();
-		if let Some(headers) = proxy.http_headers(&uri) {
-			req.headers_mut().extend(headers.clone().into_iter());
+		let body =
+			hyper::body::Body::from(serde_json::to_string(&build_request(method, params)).unwrap());
+
+		let req = hyper::Request::builder()
+			.method(hyper::Method::POST)
+			.uri(url)
+			.header(USER_AGENT, "grin-client")
+			.header(ACCEPT, "application/json")
+			.header(CONTENT_TYPE, "application/json")
+			.body(body)
+			.map_err(|e| {
+				ClientError::API(grin_api::Error::RequestError(format!(
+					"Cannot make request: {}",
+					e
+				)))
+			})?;
+
+		let client = hyper::Client::builder().build::<_, hyper::Body>(proxy);
+		let res = client.request(req).await.unwrap();
+
+		let body_bytes = hyper::body::to_bytes(res.into_body()).await.unwrap();
+		let res = String::from_utf8(body_bytes.to_vec()).unwrap();
+
+		let response: Response =
+			serde_json::from_str(&res).map_err(ClientError::DecodeResponseError)?;
+
+		if let Some(ref e) = response.error {
+			return Err(ClientError::ResponseError(e.clone()));
 		}
 
-		let res = client::send_request(req).map_err(ClientError::API)?;
+		let result = match response.result.clone() {
+			Some(r) => serde_json::from_value(r).map_err(ClientError::DecodeResponseError),
+			None => serde_json::from_value(serde_json::Value::Null)
+				.map_err(ClientError::DecodeResponseError),
+		}?;
 
-		serde_json::from_str(&res).map_err(ClientError::ResponseParse)
+		Ok(result)
 	}
 }
 
+#[async_trait]
 impl MixClient for MixClientImpl {
-	fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<(Vec<usize>, TxComponents), ClientError> {
+	async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError> {
 		let serialized = ser::ser_vec(&onions, ProtocolVersion::local()).unwrap();
 		let sig =
 			dalek::sign(&self.config.key, serialized.as_slice()).map_err(ClientError::Dalek)?;
+		// println!(
+		// 	"Created sig ({:?}) with public key ({}) for server ({})",
+		// 	&sig,
+		// 	DalekPublicKey::from_secret(&self.config.key).to_hex(),
+		// 	self.config.next_server.as_ref().unwrap().to_hex()
+		// );
 		let mix = MixReq::new(onions.clone(), sig);
 
-		let params = serde_json::json!(mix);
+		let params = serde_json::json!([mix]);
 
-		self.send_json_request::<(Vec<usize>, TxComponents)>(&self.addr, "mix", &params)
+		self.async_send_json_request::<MixResp>(&self.addr, "mix", &params)
+			.await
 	}
 }
 
 #[cfg(test)]
 pub mod mock {
 	use super::{ClientError, MixClient};
-	use crate::tx::TxComponents;
 	use grin_onion::onion::Onion;
 
+	use crate::servers::mix_rpc::MixResp;
+	use async_trait::async_trait;
 	use std::collections::HashMap;
 
 	pub struct MockMixClient {
-		results: HashMap<Vec<Onion>, (Vec<usize>, TxComponents)>,
+		results: HashMap<Vec<Onion>, MixResp>,
 	}
 
 	impl MockMixClient {
@@ -113,16 +157,14 @@ pub mod mock {
 			}
 		}
 
-		pub fn set_response(&mut self, onions: &Vec<Onion>, r: (Vec<usize>, TxComponents)) {
+		pub fn set_response(&mut self, onions: &Vec<Onion>, r: MixResp) {
 			self.results.insert(onions.clone(), r);
 		}
 	}
 
+	#[async_trait]
 	impl MixClient for MockMixClient {
-		fn mix_outputs(
-			&self,
-			onions: &Vec<Onion>,
-		) -> Result<(Vec<usize>, TxComponents), ClientError> {
+		async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError> {
 			self.results
 				.get(onions)
 				.map(|r| Ok(r.clone()))
@@ -134,13 +176,13 @@ pub mod mock {
 #[cfg(test)]
 pub mod test_util {
 	use super::{ClientError, MixClient};
-	use crate::crypto::dalek;
-	use crate::crypto::secp::SecretKey;
 	use crate::servers::mix::MixServer;
-	use crate::tx::TxComponents;
-	use crate::DalekPublicKey;
+	use crate::servers::mix_rpc::MixResp;
+	use async_trait::async_trait;
 	use grin_core::ser;
 	use grin_core::ser::ProtocolVersion;
+	use grin_onion::crypto::dalek::{self, DalekPublicKey};
+	use grin_onion::crypto::secp::SecretKey;
 	use grin_onion::onion::Onion;
 	use std::sync::Arc;
 
@@ -152,11 +194,9 @@ pub mod test_util {
 		pub mix_server: Arc<dyn MixServer>,
 	}
 
+	#[async_trait]
 	impl MixClient for DirectMixClient {
-		fn mix_outputs(
-			&self,
-			onions: &Vec<Onion>,
-		) -> Result<(Vec<usize>, TxComponents), ClientError> {
+		async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError> {
 			let serialized = ser::ser_vec(&onions, ProtocolVersion::local()).unwrap();
 			let sig = dalek::sign(&self.key, serialized.as_slice()).map_err(ClientError::Dalek)?;
 
@@ -165,7 +205,7 @@ pub mod test_util {
 				serialized.as_slice(),
 			)
 			.unwrap();
-			Ok(self.mix_server.mix_outputs(&onions, &sig).unwrap())
+			Ok(self.mix_server.mix_outputs(&onions, &sig).await.unwrap())
 		}
 	}
 }

src/config.rs

@@ -1,5 +1,5 @@
-use crate::crypto::dalek::DalekPublicKey;
-use crate::crypto::secp::SecretKey;
+use grin_onion::crypto::dalek::DalekPublicKey;
+use grin_onion::crypto::secp::SecretKey;
 
 use core::num::NonZeroU32;
 use grin_core::global::ChainTypes;
@@ -39,10 +39,10 @@ pub struct ServerConfig {
 	/// path to file containing secret for the grin wallet's owner api
 	pub wallet_owner_secret_path: Option<String>,
 	/// public key of the previous mix/swap server (e.g. N_1 if this is N_2)
-	#[serde(with = "crate::crypto::dalek::option_dalek_pubkey_serde", default)]
+	#[serde(with = "grin_onion::crypto::dalek::option_dalek_pubkey_serde", default)]
 	pub prev_server: Option<DalekPublicKey>,
 	/// public key of the next mix server
-	#[serde(with = "crate::crypto::dalek::option_dalek_pubkey_serde", default)]
+	#[serde(with = "grin_onion::crypto::dalek::option_dalek_pubkey_serde", default)]
 	pub next_server: Option<DalekPublicKey>,
 }
 
@@ -186,9 +186,9 @@ struct RawConfig {
 	grin_node_secret_path: Option<String>,
 	wallet_owner_url: SocketAddr,
 	wallet_owner_secret_path: Option<String>,
-	#[serde(with = "crate::crypto::dalek::option_dalek_pubkey_serde", default)]
+	#[serde(with = "grin_onion::crypto::dalek::option_dalek_pubkey_serde", default)]
 	prev_server: Option<DalekPublicKey>,
-	#[serde(with = "crate::crypto::dalek::option_dalek_pubkey_serde", default)]
+	#[serde(with = "grin_onion::crypto::dalek::option_dalek_pubkey_serde", default)]
 	next_server: Option<DalekPublicKey>,
 }
 
@@ -289,7 +289,8 @@ pub fn wallet_owner_url(_chain_type: &ChainTypes) -> SocketAddr {
 
 #[cfg(test)]
 pub mod test_util {
-	use crate::{DalekPublicKey, ServerConfig};
+	use crate::config::ServerConfig;
+	use grin_onion::crypto::dalek::DalekPublicKey;
 	use secp256k1zkp::SecretKey;
 	use std::net::TcpListener;
 
@@ -317,7 +318,7 @@ pub mod test_util {
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use crate::crypto::secp;
+	use grin_onion::crypto::secp;
 
 	#[test]
 	fn server_key_encrypt() {

src/lib.rs

@@ -0,0 +1,21 @@
+#[macro_use]
+extern crate log;
+
+pub mod client;
+pub mod config;
+pub mod node;
+pub mod servers;
+pub mod store;
+pub mod tor;
+pub mod tx;
+pub mod wallet;
+
+pub use client::MixClient;
+pub use config::ServerConfig;
+pub use node::{GrinNode, HttpGrinNode, NodeError};
+pub use servers::mix::{MixError, MixServer};
+pub use servers::mix_rpc::listen as mix_listen;
+pub use servers::swap::{SwapError, SwapServer};
+pub use servers::swap_rpc::listen as swap_listen;
+pub use store::{StoreError, SwapStore};
+pub use wallet::{HttpWallet, Wallet, WalletError};

src/node.rs

@@ -1,26 +1,42 @@
-use crate::crypto::secp::Commitment;
+use grin_onion::crypto::secp::Commitment;
 
-use grin_api::client;
 use grin_api::json_rpc::{build_request, Request, Response};
+use grin_api::{client, LocatedTxKernel};
 use grin_api::{OutputPrintable, OutputType, Tip};
 use grin_core::consensus::COINBASE_MATURITY;
 use grin_core::core::{Input, OutputFeatures, Transaction};
 use grin_util::ToHex;
 
+use async_trait::async_trait;
 use serde_json::json;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use thiserror::Error;
 
+#[async_trait]
 pub trait GrinNode: Send + Sync {
 	/// Retrieves the unspent output with a matching commitment
-	fn get_utxo(&self, output_commit: &Commitment) -> Result<Option<OutputPrintable>, NodeError>;
+	async fn async_get_utxo(
+		&self,
+		output_commit: &Commitment,
+	) -> Result<Option<OutputPrintable>, NodeError>;
 
 	/// Gets the height of the chain tip
-	fn get_chain_height(&self) -> Result<u64, NodeError>;
+	async fn async_get_chain_height(&self) -> Result<u64, NodeError>;
 
 	/// Posts a transaction to the grin node
-	fn post_tx(&self, tx: &Transaction) -> Result<(), NodeError>;
+	async fn async_post_tx(&self, tx: &Transaction) -> Result<(), NodeError>;
+
+	/// Returns a LocatedTxKernel based on the kernel excess.
+	/// The min_height and max_height parameters are both optional.
+	/// If not supplied, min_height will be set to 0 and max_height will be set to the head of the chain.
+	/// The method will start at the block height max_height and traverse the kernel MMR backwards, until either the kernel is found or min_height is reached.
+	async fn async_get_kernel(
+		&self,
+		excess: &Commitment,
+		min_height: Option<u64>,
+		max_height: Option<u64>,
+	) -> Result<Option<LocatedTxKernel>, NodeError>;
 }
 
 /// Error types for interacting with nodes
@@ -35,18 +51,21 @@ pub enum NodeError {
 }
 
 /// Checks if a commitment is in the UTXO set
-pub fn is_unspent(node: &Arc<dyn GrinNode>, commit: &Commitment) -> Result<bool, NodeError> {
-	let utxo = node.get_utxo(&commit)?;
+pub async fn async_is_unspent(
+	node: &Arc<dyn GrinNode>,
+	commit: &Commitment,
+) -> Result<bool, NodeError> {
+	let utxo = node.async_get_utxo(&commit).await?;
 	Ok(utxo.is_some())
 }
 
 /// Checks whether a commitment is spendable at the block height provided
-pub fn is_spendable(
+pub async fn async_is_spendable(
 	node: &Arc<dyn GrinNode>,
 	commit: &Commitment,
 	next_block_height: u64,
 ) -> Result<bool, NodeError> {
-	let output = node.get_utxo(&commit)?;
+	let output = node.async_get_utxo(&commit).await?;
 	if let Some(out) = output {
 		let is_coinbase = match out.output_type {
 			OutputType::Coinbase => true,
@@ -70,11 +89,11 @@ pub fn is_spendable(
 }
 
 /// Builds an input for an unspent output commitment
-pub fn build_input(
+pub async fn async_build_input(
 	node: &Arc<dyn GrinNode>,
 	output_commit: &Commitment,
 ) -> Result<Option<Input>, NodeError> {
-	let output = node.get_utxo(&output_commit)?;
+	let output = node.async_get_utxo(&output_commit).await?;
 
 	if let Some(out) = output {
 		let features = match out.output_type {
@@ -106,16 +125,20 @@ impl HttpGrinNode {
 		}
 	}
 
-	fn send_json_request<D: serde::de::DeserializeOwned>(
+	async fn async_send_request<D: serde::de::DeserializeOwned>(
 		&self,
 		method: &str,
 		params: &serde_json::Value,
 	) -> Result<D, NodeError> {
 		let url = format!("http://{}{}", self.node_url, ENDPOINT);
 		let req = build_request(method, params);
-		let res =
-			client::post::<Request, Response>(url.as_str(), self.node_api_secret.clone(), &req)
-				.map_err(NodeError::ApiCommError)?;
+		let res = client::post_async::<Request, Response>(
+			url.as_str(),
+			&req,
+			self.node_api_secret.clone(),
+		)
+		.await
+		.map_err(NodeError::ApiCommError)?;
 		let parsed = res
 			.clone()
 			.into_result()
@@ -124,8 +147,12 @@ impl HttpGrinNode {
 	}
 }
 
+#[async_trait]
 impl GrinNode for HttpGrinNode {
-	fn get_utxo(&self, output_commit: &Commitment) -> Result<Option<OutputPrintable>, NodeError> {
+	async fn async_get_utxo(
+		&self,
+		output_commit: &Commitment,
+	) -> Result<Option<OutputPrintable>, NodeError> {
 		let commits: Vec<String> = vec![output_commit.to_hex()];
 		let start_height: Option<u64> = None;
 		let end_height: Option<u64> = None;
@@ -139,7 +166,9 @@ impl GrinNode for HttpGrinNode {
 			include_proof,
 			include_merkle_proof
 		]);
-		let outputs = self.send_json_request::<Vec<OutputPrintable>>("get_outputs", &params)?;
+		let outputs = self
+			.async_send_request::<Vec<OutputPrintable>>("get_outputs", &params)
+			.await?;
 		if outputs.is_empty() {
 			return Ok(None);
 		}
@@ -147,29 +176,54 @@ impl GrinNode for HttpGrinNode {
 		Ok(Some(outputs[0].clone()))
 	}
 
-	fn get_chain_height(&self) -> Result<u64, NodeError> {
+	async fn async_get_chain_height(&self) -> Result<u64, NodeError> {
 		let params = json!([]);
-		let tip_json = self.send_json_request::<serde_json::Value>("get_tip", &params)?;
+		let tip_json = self
+			.async_send_request::<serde_json::Value>("get_tip", &params)
+			.await?;
 		let tip =
 			serde_json::from_value::<Tip>(tip_json).map_err(NodeError::DecodeResponseError)?;
 
 		Ok(tip.height)
 	}
 
-	fn post_tx(&self, tx: &Transaction) -> Result<(), NodeError> {
+	async fn async_post_tx(&self, tx: &Transaction) -> Result<(), NodeError> {
 		let params = json!([tx, true]);
-		self.send_json_request::<serde_json::Value>("push_transaction", &params)?;
+		self.async_send_request::<serde_json::Value>("push_transaction", &params)
+			.await?;
 		Ok(())
 	}
+
+	async fn async_get_kernel(
+		&self,
+		excess: &Commitment,
+		min_height: Option<u64>,
+		max_height: Option<u64>,
+	) -> Result<Option<LocatedTxKernel>, NodeError> {
+		let params = json!([excess.0.as_ref().to_hex(), min_height, max_height]);
+		let value = self
+			.async_send_request::<serde_json::Value>("get_kernel", &params)
+			.await?;
+
+		let contents = format!("{:?}", value);
+		if contents.contains("NotFound") {
+			return Ok(None);
+		}
+
+		let located_kernel = serde_json::from_value::<LocatedTxKernel>(value)
+			.map_err(NodeError::DecodeResponseError)?;
+		Ok(Some(located_kernel))
+	}
 }
 
 #[cfg(test)]
 pub mod mock {
 	use super::{GrinNode, NodeError};
-	use crate::crypto::secp::Commitment;
 
-	use grin_api::{OutputPrintable, OutputType};
+	use async_trait::async_trait;
+	use grin_api::{LocatedTxKernel, OutputPrintable, OutputType};
 	use grin_core::core::Transaction;
+	use grin_onion::crypto::secp::Commitment;
 	use std::collections::HashMap;
 	use std::sync::RwLock;
 
@@ -178,6 +232,7 @@ pub mod mock {
 	pub struct MockGrinNode {
 		utxos: HashMap<Commitment, OutputPrintable>,
 		txns_posted: RwLock<Vec<Transaction>>,
+		kernels: HashMap<Commitment, LocatedTxKernel>,
 	}
 
 	impl MockGrinNode {
@@ -185,6 +240,7 @@ pub mod mock {
 			MockGrinNode {
 				utxos: HashMap::new(),
 				txns_posted: RwLock::new(Vec::new()),
+				kernels: HashMap::new(),
 			}
 		}
 
@@ -192,6 +248,7 @@ pub mod mock {
 			let mut node = MockGrinNode {
 				utxos: HashMap::new(),
 				txns_posted: RwLock::new(Vec::new()),
+				kernels: HashMap::new(),
 			};
 			for utxo in utxos {
 				node.add_default_utxo(utxo);
@@ -222,10 +279,16 @@ pub mod mock {
 			let read = self.txns_posted.read().unwrap();
 			read.clone()
 		}
+
+		pub fn add_kernel(&mut self, kernel: &LocatedTxKernel) {
+			self.kernels
+				.insert(kernel.tx_kernel.excess.clone(), kernel.clone());
+		}
 	}
 
+	#[async_trait]
 	impl GrinNode for MockGrinNode {
-		fn get_utxo(
+		async fn async_get_utxo(
 			&self,
 			output_commit: &Commitment,
 		) -> Result<Option<OutputPrintable>, NodeError> {
@@ -236,14 +299,27 @@ pub mod mock {
 			Ok(None)
 		}
 
-		fn get_chain_height(&self) -> Result<u64, NodeError> {
+		async fn async_get_chain_height(&self) -> Result<u64, NodeError> {
 			Ok(100)
 		}
 
-		fn post_tx(&self, tx: &Transaction) -> Result<(), NodeError> {
+		async fn async_post_tx(&self, tx: &Transaction) -> Result<(), NodeError> {
 			let mut write = self.txns_posted.write().unwrap();
 			write.push(tx.clone());
 			Ok(())
 		}
+
+		async fn async_get_kernel(
+			&self,
+			excess: &Commitment,
+			_min_height: Option<u64>,
+			_max_height: Option<u64>,
+		) -> Result<Option<LocatedTxKernel>, NodeError> {
+			if let Some(kernel) = self.kernels.get(&excess) {
+				return Ok(Some(kernel.clone()));
+			}
+
+			Ok(None)
+		}
 	}
 }

src/servers/mix.rs

@@ -1,9 +1,12 @@
 use crate::client::MixClient;
 use crate::config::ServerConfig;
-use crate::tx::TxComponents;
+use crate::node::{self, GrinNode};
+use crate::tx::{self, TxComponents};
 use crate::wallet::Wallet;
-use crate::{node, tx, GrinNode};
 
+use crate::servers::mix_rpc::MixResp;
+use async_trait::async_trait;
+use futures::stream::{self, StreamExt};
 use grin_core::core::{Output, OutputFeatures, TransactionBody};
 use grin_core::global::DEFAULT_ACCEPT_FEE_BASE;
 use grin_core::ser;
@@ -47,13 +50,14 @@ pub enum MixError {
 }
 
 /// An internal MWixnet server - a "Mixer"
+#[async_trait]
 pub trait MixServer: Send + Sync {
 	/// Swaps the outputs provided and returns the final swapped outputs and kernels.
-	fn mix_outputs(
+	async fn mix_outputs(
 		&self,
 		onions: &Vec<Onion>,
 		sig: &DalekSignature,
-	) -> Result<(Vec<usize>, TxComponents), MixError>;
+	) -> Result<MixResp, MixError>;
 }
 
 /// The standard MWixnet "Mixer" implementation
@@ -130,15 +134,19 @@ impl MixServerImpl {
 		Ok(peeled)
 	}
 
-	fn build_final_outputs(
+	async fn async_build_final_outputs(
 		&self,
 		peeled: &Vec<(usize, PeeledOnion)>,
-	) -> Result<(Vec<usize>, TxComponents), MixError> {
+	) -> Result<MixResp, MixError> {
 		// Filter out commitments that already exist in the UTXO set
-		let filtered: Vec<&(usize, PeeledOnion)> = peeled
-			.iter()
-			.filter(|(_, p)| !node::is_unspent(&self.node, &p.onion.commit).unwrap_or(true))
-			.collect();
+		let filtered: Vec<&(usize, PeeledOnion)> = stream::iter(peeled.iter())
+			.filter(|(_, p)| async {
+				!node::async_is_unspent(&self.node, &p.onion.commit)
+					.await
+					.unwrap_or(true)
+			})
+			.collect()
+			.await;
 
 		// Build plain outputs for each mix entry
 		let outputs: Vec<Output> = filtered
@@ -158,7 +166,7 @@ impl MixServerImpl {
 			.map(|(_, p)| p.payload.excess.clone())
 			.collect();
 
-		let components = tx::assemble_components(
+		let components = tx::async_assemble_components(
 			&self.wallet,
 			&TxComponents {
 				offset: ZERO_KEY,
@@ -169,17 +177,21 @@ impl MixServerImpl {
 			self.get_fee_base(),
 			fees_paid,
 		)
+		.await
 		.map_err(MixError::TxError)?;
 
 		let indices = filtered.iter().map(|(i, _)| *i).collect();
 
-		Ok((indices, components))
+		Ok(MixResp {
+			indices,
+			components,
+		})
 	}
 
-	fn call_next_mixer(
+	async fn call_next_mixer(
 		&self,
 		peeled: &Vec<(usize, PeeledOnion)>,
-	) -> Result<(Vec<usize>, TxComponents), MixError> {
+	) -> Result<MixResp, MixError> {
 		// Sort by commitment
 		let mut onions_with_index = peeled.clone();
 		onions_with_index
@@ -191,15 +203,16 @@ impl MixServerImpl {
 
 		// Call next server
 		let onions = peeled.iter().map(|(_, p)| p.onion.clone()).collect();
-		let (mixed_indices, mixed_components) = self
+		let mixed = self
 			.mix_client
 			.as_ref()
 			.unwrap()
 			.mix_outputs(&onions)
+			.await
 			.map_err(MixError::Client)?;
 
 		// Remove filtered entries
-		let kept_next_indices = HashSet::<_>::from_iter(mixed_indices.clone());
+		let kept_next_indices = HashSet::<_>::from_iter(mixed.indices.clone());
 		let filtered_onions: Vec<&(usize, PeeledOnion)> = onions_with_index
 			.iter()
 			.filter(|(i, _)| {
@@ -221,25 +234,30 @@ impl MixServerImpl {
 
 		let indices = kept_next_indices.into_iter().sorted().collect();
 
-		let components = tx::assemble_components(
+		let components = tx::async_assemble_components(
 			&self.wallet,
-			&mixed_components,
+			&mixed.components,
 			&excesses,
 			self.get_fee_base(),
 			fees_paid,
 		)
+		.await
 		.map_err(MixError::TxError)?;
 
-		Ok((indices, components))
+		Ok(MixResp {
+			indices,
+			components,
+		})
 	}
 }
 
+#[async_trait]
 impl MixServer for MixServerImpl {
-	fn mix_outputs(
+	async fn mix_outputs(
 		&self,
 		onions: &Vec<Onion>,
 		sig: &DalekSignature,
-	) -> Result<(Vec<usize>, TxComponents), MixError> {
+	) -> Result<MixResp, MixError> {
 		// Verify Signature
 		let serialized = ser::ser_vec(&onions, ProtocolVersion::local()).unwrap();
 		sig.verify(
@@ -252,10 +270,10 @@ impl MixServer for MixServerImpl {
 		let mut peeled: Vec<(usize, PeeledOnion)> = onions
 			.iter()
 			.enumerate()
-			.filter_map(|(i, o)| {
-				if let Some(p) = self.peel_onion(&o).ok() {
-					Some((i, p))
-				} else {
+			.filter_map(|(i, o)| match self.peel_onion(&o) {
+				Ok(p) => Some((i, p)),
+				Err(e) => {
+					println!("Error peeling onion: {:?}", e);
 					None
 				}
 			})
@@ -271,9 +289,9 @@ impl MixServer for MixServerImpl {
 		}
 
 		if self.server_config.next_server.is_some() {
-			self.call_next_mixer(&peeled)
+			self.call_next_mixer(&peeled).await
 		} else {
-			self.build_final_outputs(&peeled)
+			self.async_build_final_outputs(&peeled).await
 		}
 	}
 }
@@ -281,11 +299,13 @@ impl MixServer for MixServerImpl {
 #[cfg(test)]
 mod test_util {
 	use crate::client::test_util::DirectMixClient;
+	use crate::client::MixClient;
+	use crate::config;
 	use crate::node::mock::MockGrinNode;
-	use crate::wallet::mock::MockWallet;
-	use crate::{config, DalekPublicKey, MixClient};
-
 	use crate::servers::mix::MixServerImpl;
+	use crate::wallet::mock::MockWallet;
+
+	use grin_onion::crypto::dalek::DalekPublicKey;
 	use secp256k1zkp::SecretKey;
 	use std::sync::Arc;
 
@@ -320,10 +340,11 @@ mod test_util {
 
 #[cfg(test)]
 mod tests {
+	use crate::client::MixClient;
 	use crate::node::mock::MockGrinNode;
-	use crate::{DalekPublicKey, MixClient};
 
 	use ::function_name::named;
+	use grin_onion::crypto::dalek::DalekPublicKey;
 	use grin_onion::crypto::secp::{self, Commitment};
 	use grin_onion::test_util as onion_test_util;
 	use grin_onion::{create_onion, new_hop, Hop};
@@ -373,9 +394,9 @@ mod tests {
 	/// * Swap Server - Simulated by test
 	/// * Mixer 1 - Internal MixServerImpl directly called by test
 	/// * Mixer 2 - Final MixServerImpl called by Mixer 1
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn mix_lifecycle() -> Result<(), Box<dyn std::error::Error>> {
+	async fn mix_lifecycle() -> Result<(), Box<dyn std::error::Error>> {
 		init_test!();
 
 		// Setup Input(s)
@@ -426,13 +447,15 @@ mod tests {
 
 		// Simulate the swap server peeling the onion and then calling mix1
 		let mix1_onion = onion.peel_layer(&swap_vars.sk)?;
-		let (mixed_indices, mixed_components) =
-			mixer1_client.mix_outputs(&vec![mix1_onion.onion.clone()])?;
+		let mixed = mixer1_client
+			.mix_outputs(&vec![mix1_onion.onion.clone()])
+			.await?;
 
 		// Verify 3 outputs are returned: mixed output, mixer1's output, and mixer2's output
-		assert_eq!(mixed_indices, vec![0 as usize]);
-		assert_eq!(mixed_components.outputs.len(), 3);
-		let output_commits: HashSet<Commitment> = mixed_components
+		assert_eq!(mixed.indices, vec![0 as usize]);
+		assert_eq!(mixed.components.outputs.len(), 3);
+		let output_commits: HashSet<Commitment> = mixed
+			.components
 			.outputs
 			.iter()
 			.map(|o| o.identifier.commit.clone())

src/servers/mix_rpc.rs

@@ -1,19 +1,19 @@
 use crate::client::MixClient;
 use crate::config::ServerConfig;
-use crate::crypto::dalek::{self, DalekSignature};
 use crate::node::GrinNode;
 use crate::servers::mix::{MixError, MixServer, MixServerImpl};
 use crate::wallet::Wallet;
 
+use crate::tx::TxComponents;
+use futures::FutureExt;
+use grin_onion::crypto::dalek::{self, DalekSignature};
 use grin_onion::onion::Onion;
-use grin_util::StopState;
+use jsonrpc_core::BoxFuture;
 use jsonrpc_derive::rpc;
 use jsonrpc_http_server::jsonrpc_core::{self as jsonrpc, IoHandler};
 use jsonrpc_http_server::{DomainsValidation, ServerBuilder};
 use serde::{Deserialize, Serialize};
-use std::sync::{Arc, Mutex};
-use std::thread::{sleep, spawn};
-use std::time::Duration;
+use std::sync::Arc;
 
 #[derive(Serialize, Deserialize)]
 pub struct MixReq {
@@ -22,6 +22,12 @@ pub struct MixReq {
 	sig: DalekSignature,
 }
 
+#[derive(Clone, Serialize, Deserialize)]
+pub struct MixResp {
+	pub indices: Vec<usize>,
+	pub components: TxComponents,
+}
+
 impl MixReq {
 	pub fn new(onions: Vec<Onion>, sig: DalekSignature) -> Self {
 		MixReq { onions, sig }
@@ -31,22 +37,23 @@ impl MixReq {
 #[rpc(server)]
 pub trait MixAPI {
 	#[rpc(name = "mix")]
-	fn mix(&self, mix: MixReq) -> jsonrpc::Result<jsonrpc::Value>;
+	fn mix(&self, mix: MixReq) -> BoxFuture<jsonrpc::Result<MixResp>>;
 }
 
 #[derive(Clone)]
 struct RPCMixServer {
 	server_config: ServerConfig,
-	server: Arc<Mutex<dyn MixServer>>,
+	server: Arc<tokio::sync::Mutex<dyn MixServer>>,
 }
 
 impl RPCMixServer {
 	/// Spin up an instance of the JSON-RPC HTTP server.
-	fn start_http(&self) -> jsonrpc_http_server::Server {
+	fn start_http(&self, runtime_handle: tokio::runtime::Handle) -> jsonrpc_http_server::Server {
 		let mut io = IoHandler::new();
 		io.extend_with(RPCMixServer::to_delegate(self.clone()));
 
 		ServerBuilder::new(io)
+			.event_loop_executor(runtime_handle)
 			.cors(DomainsValidation::Disabled)
 			.request_middleware(|request: hyper::Request<hyper::Body>| {
 				if request.uri() == "/v1" {
@@ -67,50 +74,48 @@ impl From<MixError> for jsonrpc::Error {
 }
 
 impl MixAPI for RPCMixServer {
-	fn mix(&self, mix: MixReq) -> jsonrpc::Result<jsonrpc::Value> {
-		self.server
-			.lock()
-			.unwrap()
-			.mix_outputs(&mix.onions, &mix.sig)?;
-		Ok(jsonrpc::Value::String("success".into()))
+	fn mix(&self, mix: MixReq) -> BoxFuture<jsonrpc::Result<MixResp>> {
+		let server = self.server.clone();
+		async move {
+			let response = server
+				.lock()
+				.await
+				.mix_outputs(&mix.onions, &mix.sig)
+				.await?;
+			Ok(response)
+		}
+		.boxed()
 	}
 }
 
 /// Spin up the JSON-RPC web server
 pub fn listen(
+	rt_handle: &tokio::runtime::Handle,
 	server_config: ServerConfig,
 	next_server: Option<Arc<dyn MixClient>>,
 	wallet: Arc<dyn Wallet>,
 	node: Arc<dyn GrinNode>,
-	stop_state: Arc<StopState>,
-) -> Result<(), Box<dyn std::error::Error>> {
+) -> Result<
+	(
+		Arc<tokio::sync::Mutex<dyn MixServer>>,
+		jsonrpc_http_server::Server,
+	),
+	Box<dyn std::error::Error>,
+> {
 	let server = MixServerImpl::new(
 		server_config.clone(),
 		next_server,
 		wallet.clone(),
 		node.clone(),
 	);
-	let server = Arc::new(Mutex::new(server));
+	let server = Arc::new(tokio::sync::Mutex::new(server));
 
 	let rpc_server = RPCMixServer {
 		server_config: server_config.clone(),
 		server: server.clone(),
 	};
 
-	let http_server = rpc_server.start_http();
+	let http_server = rpc_server.start_http(rt_handle.clone());
 
-	let close_handle = http_server.close_handle();
-	let round_handle = spawn(move || loop {
-		if stop_state.is_stopped() {
-			close_handle.close();
-			break;
-		}
-
-		sleep(Duration::from_millis(100));
-	});
-
-	http_server.wait();
-	round_handle.join().unwrap();
-
-	Ok(())
+	Ok((server, http_server))
 }

src/servers/swap.rs

@@ -1,21 +1,21 @@
 use crate::client::MixClient;
 use crate::config::ServerConfig;
-use crate::crypto::comsig::ComSignature;
-use crate::crypto::secp::{Commitment, Secp256k1, SecretKey};
 use crate::node::{self, GrinNode};
 use crate::store::{StoreError, SwapData, SwapStatus, SwapStore};
 use crate::tx;
 use crate::wallet::Wallet;
 
-use grin_core::core::hash::Hashed;
+use async_trait::async_trait;
 use grin_core::core::{Input, Output, OutputFeatures, Transaction, TransactionBody};
 use grin_core::global::DEFAULT_ACCEPT_FEE_BASE;
+use grin_onion::crypto::comsig::ComSignature;
+use grin_onion::crypto::secp::{Commitment, Secp256k1, SecretKey};
 use grin_onion::onion::{Onion, OnionError};
 use itertools::Itertools;
 use secp256k1zkp::key::ZERO_KEY;
 use std::collections::HashSet;
 use std::result::Result;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use thiserror::Error;
 
 /// Swap error types
@@ -39,20 +39,43 @@ pub enum SwapError {
 	FeeTooLow { minimum_fee: u64, actual_fee: u64 },
 	#[error("Error saving swap to data store: {0}")]
 	StoreError(StoreError),
+	#[error("Error building transaction: {0}")]
+	TxError(String),
+	#[error("Node communication error: {0}")]
+	NodeError(String),
 	#[error("Client communication error: {0:?}")]
 	ClientError(String),
 	#[error("{0}")]
 	UnknownError(String),
 }
 
+impl From<StoreError> for SwapError {
+	fn from(e: StoreError) -> SwapError {
+		SwapError::StoreError(e)
+	}
+}
+
+impl From<tx::TxError> for SwapError {
+	fn from(e: tx::TxError) -> SwapError {
+		SwapError::TxError(e.to_string())
+	}
+}
+
+impl From<node::NodeError> for SwapError {
+	fn from(e: node::NodeError) -> SwapError {
+		SwapError::NodeError(e.to_string())
+	}
+}
+
 /// A public MWixnet server - the "Swap Server"
+#[async_trait]
 pub trait SwapServer: Send + Sync {
 	/// Submit a new output to be swapped.
-	fn swap(&self, onion: &Onion, comsig: &ComSignature) -> Result<(), SwapError>;
+	async fn swap(&self, onion: &Onion, comsig: &ComSignature) -> Result<(), SwapError>;
 
 	/// Iterate through all saved submissions, filter out any inputs that are no longer spendable,
 	/// and assemble the coinswap transaction, posting the transaction to the configured node.
-	fn execute_round(&self) -> Result<Option<Transaction>, Box<dyn std::error::Error>>;
+	async fn execute_round(&self) -> Result<Option<Arc<Transaction>>, SwapError>;
 }
 
 /// The standard MWixnet server implementation
@@ -62,7 +85,7 @@ pub struct SwapServerImpl {
 	next_server: Option<Arc<dyn MixClient>>,
 	wallet: Arc<dyn Wallet>,
 	node: Arc<dyn GrinNode>,
-	store: Arc<Mutex<SwapStore>>,
+	store: Arc<tokio::sync::Mutex<SwapStore>>,
 }
 
 impl SwapServerImpl {
@@ -79,7 +102,7 @@ impl SwapServerImpl {
 			next_server,
 			wallet,
 			node,
-			store: Arc::new(Mutex::new(store)),
+			store: Arc::new(tokio::sync::Mutex::new(store)),
 		}
 	}
 
@@ -93,10 +116,29 @@ impl SwapServerImpl {
 	fn get_minimum_swap_fee(&self) -> u64 {
 		TransactionBody::weight_by_iok(1, 1, 1) * self.get_fee_base()
 	}
+
+	async fn async_is_spendable(&self, next_block_height: u64, swap: &SwapData) -> bool {
+		if let SwapStatus::Unprocessed = swap.status {
+			if node::async_is_spendable(&self.node, &swap.input.commit, next_block_height)
+				.await
+				.unwrap_or(false)
+			{
+				if !node::async_is_unspent(&self.node, &swap.output_commit)
+					.await
+					.unwrap_or(true)
+				{
+					return true;
+				}
+			}
+		}
+
+		false
+	}
 }
 
+#[async_trait]
 impl SwapServer for SwapServerImpl {
-	fn swap(&self, onion: &Onion, comsig: &ComSignature) -> Result<(), SwapError> {
+	async fn swap(&self, onion: &Onion, comsig: &ComSignature) -> Result<(), SwapError> {
 		// Verify that more than 1 payload exists when there's a next server,
 		// or that exactly 1 payload exists when this is the final server
 		if self.server_config.next_server.is_some() && onion.enc_payloads.len() <= 1
@@ -114,7 +156,8 @@ impl SwapServer for SwapServerImpl {
 			.map_err(|_| SwapError::InvalidComSignature)?;
 
 		// Verify that commitment is unspent
-		let input = node::build_input(&self.node, &onion.commit)
+		let input = node::async_build_input(&self.node, &onion.commit)
+			.await
 			.map_err(|e| SwapError::UnknownError(e.to_string()))?;
 		let input = input.ok_or(SwapError::CoinNotFound {
 			commit: onion.commit.clone(),
@@ -144,7 +187,7 @@ impl SwapServer for SwapServerImpl {
 			return Err(SwapError::MissingRangeproof);
 		}
 
-		let locked = self.store.lock().unwrap();
+		let locked = self.store.lock().await;
 
 		locked
 			.save_swap(
@@ -168,23 +211,22 @@ impl SwapServer for SwapServerImpl {
 		Ok(())
 	}
 
-	fn execute_round(&self) -> Result<Option<Transaction>, Box<dyn std::error::Error>> {
-		let locked_store = self.store.lock().unwrap();
-		let next_block_height = self.node.get_chain_height()? + 1;
+	async fn execute_round(&self) -> Result<Option<Arc<Transaction>>, SwapError> {
+		let next_block_height = self.node.async_get_chain_height().await? + 1;
 
-		let spendable: Vec<SwapData> = locked_store
+		let locked_store = self.store.lock().await;
+		let swaps: Vec<SwapData> = locked_store
 			.swaps_iter()?
 			.unique_by(|s| s.output_commit)
-			.filter(|s| match s.status {
-				SwapStatus::Unprocessed => true,
-				_ => false,
-			})
-			.filter(|s| {
-				node::is_spendable(&self.node, &s.input.commit, next_block_height).unwrap_or(false)
-			})
-			.filter(|s| !node::is_unspent(&self.node, &s.output_commit).unwrap_or(true))
-			.sorted_by(|a, b| a.output_commit.partial_cmp(&b.output_commit).unwrap())
 			.collect();
+		let mut spendable: Vec<SwapData> = vec![];
+		for swap in &swaps {
+			if self.async_is_spendable(next_block_height, &swap).await {
+				spendable.push(swap.clone());
+			}
+		}
+
+		spendable.sort_by(|a, b| a.output_commit.partial_cmp(&b.output_commit).unwrap());
 
 		if spendable.len() == 0 {
 			return Ok(None);
@@ -193,12 +235,13 @@ impl SwapServer for SwapServerImpl {
 		let (filtered, failed, offset, outputs, kernels) = if let Some(client) = &self.next_server {
 			// Call next mix server
 			let onions = spendable.iter().map(|s| s.onion.clone()).collect();
-			let (indices, mixed) = client
+			let mixed = client
 				.mix_outputs(&onions)
+				.await
 				.map_err(|e| SwapError::ClientError(e.to_string()))?;
 
 			// Filter out failed entries
-			let kept_indices = HashSet::<_>::from_iter(indices.clone());
+			let kept_indices = HashSet::<_>::from_iter(mixed.indices.clone());
 			let filtered = spendable
 				.iter()
 				.enumerate()
@@ -213,7 +256,13 @@ impl SwapServer for SwapServerImpl {
 				.map(|(_, j)| j.clone())
 				.collect();
 
-			(filtered, failed, mixed.offset, mixed.outputs, mixed.kernels)
+			(
+				filtered,
+				failed,
+				mixed.components.offset,
+				mixed.components.outputs,
+				mixed.components.kernels,
+			)
 		} else {
 			// Build plain outputs for each swap entry
 			let outputs: Vec<Output> = spendable
@@ -234,7 +283,7 @@ impl SwapServer for SwapServerImpl {
 		let inputs: Vec<Input> = filtered.iter().map(|s| s.input).collect();
 		let output_excesses: Vec<SecretKey> = filtered.iter().map(|s| s.excess.clone()).collect();
 
-		let tx = tx::assemble_tx(
+		let tx = tx::async_assemble_tx(
 			&self.wallet,
 			&inputs,
 			&outputs,
@@ -243,14 +292,15 @@ impl SwapServer for SwapServerImpl {
 			fees_paid,
 			&offset,
 			&output_excesses,
-		)?;
+		)
+		.await?;
 
-		self.node.post_tx(&tx)?;
+		self.node.async_post_tx(&tx).await?;
 
 		// Update status to in process
-		let kernel_hash = tx.kernels().first().unwrap().hash();
+		let kernel_commit = tx.kernels().first().unwrap().excess;
 		for mut swap in filtered {
-			swap.status = SwapStatus::InProcess { kernel_hash };
+			swap.status = SwapStatus::InProcess { kernel_commit };
 			locked_store.save_swap(&swap, true)?;
 		}
 
@@ -260,16 +310,17 @@ impl SwapServer for SwapServerImpl {
 			locked_store.save_swap(&swap, true)?;
 		}
 
-		Ok(Some(tx))
+		Ok(Some(Arc::new(tx)))
 	}
 }
 
 #[cfg(test)]
 pub mod mock {
 	use super::{SwapError, SwapServer};
-	use crate::crypto::comsig::ComSignature;
 
+	use async_trait::async_trait;
 	use grin_core::core::Transaction;
+	use grin_onion::crypto::comsig::ComSignature;
 	use grin_onion::onion::Onion;
 	use std::collections::HashMap;
 
@@ -289,8 +340,9 @@ pub mod mock {
 		}
 	}
 
+	#[async_trait]
 	impl SwapServer for MockSwapServer {
-		fn swap(&self, onion: &Onion, _comsig: &ComSignature) -> Result<(), SwapError> {
+		async fn swap(&self, onion: &Onion, _comsig: &ComSignature) -> Result<(), SwapError> {
 			if let Some(e) = self.errors.get(&onion) {
 				return Err(e.clone());
 			}
@@ -298,7 +350,7 @@ pub mod mock {
 			Ok(())
 		}
 
-		fn execute_round(&self) -> Result<Option<Transaction>, Box<dyn std::error::Error>> {
+		async fn execute_round(&self) -> Result<Option<std::sync::Arc<Transaction>>, SwapError> {
 			Ok(None)
 		}
 	}
@@ -306,11 +358,15 @@ pub mod mock {
 
 #[cfg(test)]
 pub mod test_util {
-	use crate::crypto::dalek::DalekPublicKey;
-	use crate::crypto::secp::SecretKey;
+	use crate::client::MixClient;
+	use crate::config;
+	use crate::node::GrinNode;
 	use crate::servers::swap::SwapServerImpl;
+	use crate::store::SwapStore;
 	use crate::wallet::mock::MockWallet;
-	use crate::{config, GrinNode, MixClient, SwapStore};
+
+	use grin_onion::crypto::dalek::DalekPublicKey;
+	use grin_onion::crypto::secp::SecretKey;
 	use std::sync::Arc;
 
 	pub fn new_swapper(
@@ -339,14 +395,15 @@ pub mod test_util {
 
 #[cfg(test)]
 mod tests {
+	use crate::client::{self, MixClient};
 	use crate::node::mock::MockGrinNode;
+	use crate::servers::mix_rpc::MixResp;
 	use crate::servers::swap::{SwapError, SwapServer};
 	use crate::store::{SwapData, SwapStatus};
+	use crate::tx;
 	use crate::tx::TxComponents;
-	use crate::{client, tx, MixClient};
 
 	use ::function_name::named;
-	use grin_core::core::hash::Hashed;
 	use grin_core::core::{Committed, Input, Output, OutputFeatures, Transaction, Weighting};
 	use grin_onion::crypto::comsig::ComSignature;
 	use grin_onion::crypto::secp;
@@ -380,9 +437,9 @@ mod tests {
 	}
 
 	/// Standalone swap server to demonstrate request validation and onion unwrapping.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_standalone() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_standalone() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -400,7 +457,7 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		server.swap(&onion, &comsig)?;
+		server.swap(&onion, &comsig).await?;
 
 		// Make sure entry is added to server.
 		let expected = SwapData {
@@ -418,21 +475,21 @@ mod tests {
 		};
 
 		{
-			let store = server.store.lock().unwrap();
+			let store = server.store.lock().await;
 			assert_eq!(1, store.swaps_iter().unwrap().count());
 			assert!(store.swap_exists(&input_commit).unwrap());
 			assert_eq!(expected, store.get_swap(&input_commit).unwrap());
 		}
 
-		let tx = server.execute_round()?;
+		let tx = server.execute_round().await?;
 		assert!(tx.is_some());
 
 		{
 			// check that status was updated
-			let store = server.store.lock().unwrap();
+			let store = server.store.lock().await;
 			assert!(match store.get_swap(&input_commit)?.status {
-				SwapStatus::InProcess { kernel_hash } =>
-					kernel_hash == tx.unwrap().kernels().first().unwrap().hash(),
+				SwapStatus::InProcess { kernel_commit } =>
+					kernel_commit == tx.unwrap().kernels().first().unwrap().excess,
 				_ => false,
 			});
 		}
@@ -451,9 +508,9 @@ mod tests {
 	}
 
 	/// Multi-server test to verify proper MixClient communication.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_multiserver() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_multiserver() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 		let test_dir = init_test!();
 
 		// Setup input
@@ -498,7 +555,10 @@ mod tests {
 		};
 		mock_mixer.set_response(
 			&vec![mixer_onion.clone()],
-			(vec![0 as usize], mixer_response),
+			MixResp {
+				indices: vec![0 as usize],
+				components: mixer_response,
+			},
 		);
 
 		let mixer: Arc<dyn MixClient> = Arc::new(mock_mixer);
@@ -508,9 +568,9 @@ mod tests {
 			Some((&mixer_pk, &mixer)),
 			node.clone(),
 		);
-		swapper.swap(&onion, &comsig)?;
+		swapper.swap(&onion, &comsig).await?;
 
-		let tx = swapper.execute_round()?;
+		let tx = swapper.execute_round().await?;
 		assert!(tx.is_some());
 
 		// check that the transaction was posted
@@ -527,9 +587,9 @@ mod tests {
 	}
 
 	/// Returns InvalidPayloadLength when too many payloads are provided.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_too_many_payloads() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_too_many_payloads() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -549,22 +609,19 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 		assert_eq!(Err(SwapError::InvalidPayloadLength), result);
 
 		// Make sure no entry is added to the store
-		assert_eq!(
-			0,
-			server.store.lock().unwrap().swaps_iter().unwrap().count()
-		);
+		assert_eq!(0, server.store.lock().await.swaps_iter().unwrap().count());
 
 		Ok(())
 	}
 
 	/// Returns InvalidComSignature when ComSignature fails to verify.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_invalid_com_signature() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_invalid_com_signature() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -585,22 +642,19 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 		assert_eq!(Err(SwapError::InvalidComSignature), result);
 
 		// Make sure no entry is added to the store
-		assert_eq!(
-			0,
-			server.store.lock().unwrap().swaps_iter().unwrap().count()
-		);
+		assert_eq!(0, server.store.lock().await.swaps_iter().unwrap().count());
 
 		Ok(())
 	}
 
 	/// Returns InvalidRangeProof when the rangeproof fails to verify for the commitment.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_invalid_rangeproof() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_invalid_rangeproof() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -620,22 +674,19 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 		assert_eq!(Err(SwapError::InvalidRangeproof), result);
 
 		// Make sure no entry is added to the store
-		assert_eq!(
-			0,
-			server.store.lock().unwrap().swaps_iter().unwrap().count()
-		);
+		assert_eq!(0, server.store.lock().await.swaps_iter().unwrap().count());
 
 		Ok(())
 	}
 
 	/// Returns MissingRangeproof when no rangeproof is provided.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_missing_rangeproof() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_missing_rangeproof() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -652,22 +703,19 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 		assert_eq!(Err(SwapError::MissingRangeproof), result);
 
 		// Make sure no entry is added to the store
-		assert_eq!(
-			0,
-			server.store.lock().unwrap().swaps_iter().unwrap().count()
-		);
+		assert_eq!(0, server.store.lock().await.swaps_iter().unwrap().count());
 
 		Ok(())
 	}
 
 	/// Returns CoinNotFound when there's no matching output in the UTXO set.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_utxo_missing() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_utxo_missing() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -686,7 +734,7 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new());
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 		assert_eq!(
 			Err(SwapError::CoinNotFound {
 				commit: input_commit.clone()
@@ -695,18 +743,15 @@ mod tests {
 		);
 
 		// Make sure no entry is added to the store
-		assert_eq!(
-			0,
-			server.store.lock().unwrap().swaps_iter().unwrap().count()
-		);
+		assert_eq!(0, server.store.lock().await.swaps_iter().unwrap().count());
 
 		Ok(())
 	}
 
 	/// Returns AlreadySwapped when trying to swap the same commitment multiple times.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_already_swapped() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_already_swapped() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -725,10 +770,10 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		server.swap(&onion, &comsig)?;
+		server.swap(&onion, &comsig).await?;
 
 		// Call swap a second time
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 		assert_eq!(
 			Err(SwapError::AlreadySwapped {
 				commit: input_commit.clone()
@@ -740,9 +785,9 @@ mod tests {
 	}
 
 	/// Returns PeelOnionFailure when a failure occurs trying to decrypt the onion payload.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_peel_onion_failure() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_peel_onion_failure() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -763,7 +808,7 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 
 		assert!(result.is_err());
 		assert_error_type!(result, SwapError::PeelOnionFailure(_));
@@ -772,9 +817,9 @@ mod tests {
 	}
 
 	/// Returns FeeTooLow when the minimum fee is not met.
-	#[test]
+	#[tokio::test]
 	#[named]
-	fn swap_fee_too_low() -> Result<(), Box<dyn std::error::Error>> {
+	async fn swap_fee_too_low() -> Result<(), Box<dyn std::error::Error>> {
 		let test_dir = init_test!();
 
 		let value: u64 = 200_000_000;
@@ -793,7 +838,7 @@ mod tests {
 
 		let node: Arc<MockGrinNode> = Arc::new(MockGrinNode::new_with_utxos(&vec![&input_commit]));
 		let (server, _) = super::test_util::new_swapper(&test_dir, &server_key, None, node.clone());
-		let result = server.swap(&onion, &comsig);
+		let result = server.swap(&onion, &comsig).await;
 		assert_eq!(
 			Err(SwapError::FeeTooLow {
 				minimum_fee: 12_500_000,

src/servers/swap_rpc.rs

@@ -5,17 +5,15 @@ use crate::servers::swap::{SwapError, SwapServer, SwapServerImpl};
 use crate::store::SwapStore;
 use crate::wallet::Wallet;
 
+use futures::FutureExt;
 use grin_onion::crypto::comsig::{self, ComSignature};
 use grin_onion::onion::Onion;
-use grin_util::StopState;
 use jsonrpc_core::Value;
 use jsonrpc_derive::rpc;
 use jsonrpc_http_server::jsonrpc_core::*;
-use jsonrpc_http_server::*;
+use jsonrpc_http_server::{DomainsValidation, ServerBuilder};
 use serde::{Deserialize, Serialize};
-use std::sync::{Arc, Mutex};
-use std::thread::{sleep, spawn};
-use std::time::Duration;
+use std::sync::Arc;
 
 #[derive(Serialize, Deserialize)]
 pub struct SwapReq {
@@ -27,22 +25,23 @@ pub struct SwapReq {
 #[rpc(server)]
 pub trait SwapAPI {
 	#[rpc(name = "swap")]
-	fn swap(&self, swap: SwapReq) -> jsonrpc_core::Result<Value>;
+	fn swap(&self, swap: SwapReq) -> BoxFuture<jsonrpc_core::Result<Value>>;
 }
 
 #[derive(Clone)]
 struct RPCSwapServer {
 	server_config: ServerConfig,
-	server: Arc<Mutex<dyn SwapServer>>,
+	server: Arc<tokio::sync::Mutex<dyn SwapServer>>,
 }
 
 impl RPCSwapServer {
 	/// Spin up an instance of the JSON-RPC HTTP server.
-	fn start_http(&self) -> jsonrpc_http_server::Server {
+	fn start_http(&self, runtime_handle: tokio::runtime::Handle) -> jsonrpc_http_server::Server {
 		let mut io = IoHandler::new();
 		io.extend_with(RPCSwapServer::to_delegate(self.clone()));
 
 		ServerBuilder::new(io)
+			.event_loop_executor(runtime_handle)
 			.cors(DomainsValidation::Disabled)
 			.request_middleware(|request: hyper::Request<hyper::Body>| {
 				if request.uri() == "/v1" {
@@ -70,24 +69,31 @@ impl From<SwapError> for Error {
 }
 
 impl SwapAPI for RPCSwapServer {
-	fn swap(&self, swap: SwapReq) -> jsonrpc_core::Result<Value> {
-		self.server
-			.lock()
-			.unwrap()
-			.swap(&swap.onion, &swap.comsig)?;
-		Ok(Value::String("success".into()))
+	fn swap(&self, swap: SwapReq) -> BoxFuture<jsonrpc_core::Result<Value>> {
+		let server = self.server.clone();
+		async move {
+			server.lock().await.swap(&swap.onion, &swap.comsig).await?;
+			Ok(Value::String("success".into()))
+		}
+		.boxed()
 	}
 }
 
 /// Spin up the JSON-RPC web server
 pub fn listen(
-	server_config: ServerConfig,
+	rt_handle: &tokio::runtime::Handle,
+	server_config: &ServerConfig,
 	next_server: Option<Arc<dyn MixClient>>,
 	wallet: Arc<dyn Wallet>,
 	node: Arc<dyn GrinNode>,
 	store: SwapStore,
-	stop_state: Arc<StopState>,
-) -> std::result::Result<(), Box<dyn std::error::Error>> {
+) -> std::result::Result<
+	(
+		Arc<tokio::sync::Mutex<dyn SwapServer>>,
+		jsonrpc_http_server::Server,
+	),
+	Box<dyn std::error::Error>,
+> {
 	let server = SwapServerImpl::new(
 		server_config.clone(),
 		next_server,
@@ -95,54 +101,33 @@ pub fn listen(
 		node.clone(),
 		store,
 	);
-	let server = Arc::new(Mutex::new(server));
+	let server = Arc::new(tokio::sync::Mutex::new(server));
 
 	let rpc_server = RPCSwapServer {
 		server_config: server_config.clone(),
 		server: server.clone(),
 	};
 
-	let http_server = rpc_server.start_http();
+	let http_server = rpc_server.start_http(rt_handle.clone());
 
-	let close_handle = http_server.close_handle();
-	let round_handle = spawn(move || {
-		let mut secs = 0;
-		loop {
-			if stop_state.is_stopped() {
-				close_handle.close();
-				break;
-			}
-
-			sleep(Duration::from_secs(1));
-			secs = (secs + 1) % server_config.interval_s;
-
-			if secs == 0 {
-				let _ = server.lock().unwrap().execute_round();
-			}
-		}
-	});
-
-	http_server.wait();
-	round_handle.join().unwrap();
-
-	Ok(())
+	Ok((server, http_server))
 }
 
 #[cfg(test)]
 mod tests {
 	use crate::config::ServerConfig;
-	use crate::crypto::comsig::ComSignature;
-	use crate::crypto::secp;
 	use crate::servers::swap::mock::MockSwapServer;
 	use crate::servers::swap::{SwapError, SwapServer};
 	use crate::servers::swap_rpc::{RPCSwapServer, SwapReq};
 
 	use grin_onion::create_onion;
+	use grin_onion::crypto::comsig::ComSignature;
+	use grin_onion::crypto::secp;
 	use std::net::TcpListener;
-	use std::sync::{Arc, Mutex};
+	use std::sync::Arc;
 
 	use hyper::{Body, Client, Request, Response};
-	use tokio::runtime::Runtime;
+	use tokio::sync::Mutex;
 
 	async fn body_to_string(req: Response<Body>) -> String {
 		let body_bytes = hyper::body::to_bytes(req.into_body()).await.unwrap();
@@ -150,10 +135,11 @@ mod tests {
 	}
 
 	/// Spin up a temporary web service, query the API, then cleanup and return response
-	fn make_request(
-		server: Arc<Mutex<dyn SwapServer>>,
+	async fn async_make_request(
+		server: Arc<tokio::sync::Mutex<dyn SwapServer>>,
 		req: String,
-	) -> Result<String, Box<dyn std::error::Error>> {
+		runtime_handle: &tokio::runtime::Handle,
+	) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
 		let server_config = ServerConfig {
 			key: secp::random_secret(),
 			interval_s: 1,
@@ -173,28 +159,21 @@ mod tests {
 		};
 
 		// Start the JSON-RPC server
-		let http_server = rpc_server.start_http();
+		let http_server = rpc_server.start_http(runtime_handle.clone());
 
 		let uri = format!("http://{}/v1", server_config.addr);
 
-		let threaded_rt = Runtime::new()?;
-		let do_request = async move {
-			let request = Request::post(uri)
-				.header("Content-Type", "application/json")
-				.body(Body::from(req))
-				.unwrap();
+		let request = Request::post(uri)
+			.header("Content-Type", "application/json")
+			.body(Body::from(req))
+			.unwrap();
 
-			Client::new().request(request).await
-		};
+		let response = Client::new().request(request).await?;
 
-		let response = threaded_rt.block_on(do_request)?;
-		let response_str: String = threaded_rt.block_on(body_to_string(response));
-
-		// Wait for shutdown
-		threaded_rt.shutdown_background();
+		let response_str: String = body_to_string(response).await;
 
 		// Execute one round
-		server.lock().unwrap().execute_round()?;
+		server.lock().await.execute_round().await?;
 
 		// Stop the server
 		http_server.close();
@@ -206,7 +185,11 @@ mod tests {
 
 	/// Demonstrates a successful swap response
 	#[test]
-	fn swap_success() -> Result<(), Box<dyn std::error::Error>> {
+	fn swap_success() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+		let mut rt = tokio::runtime::Builder::new()
+			.threaded_scheduler()
+			.enable_all()
+			.build()?;
 		let commitment = secp::commit(1234, &secp::random_secret())?;
 		let onion = create_onion(&commitment, &vec![])?;
 		let comsig = ComSignature::sign(1234, &secp::random_secret(), &onion.serialize()?)?;
@@ -221,7 +204,8 @@ mod tests {
 			"{{\"jsonrpc\": \"2.0\", \"method\": \"swap\", \"params\": [{}], \"id\": \"1\"}}",
 			serde_json::json!(swap)
 		);
-		let response = make_request(server, req)?;
+		let rt_handle = rt.handle().clone();
+		let response = rt.block_on(async_make_request(server, req, &rt_handle))?;
 		let expected = "{\"jsonrpc\":\"2.0\",\"result\":\"success\",\"id\":\"1\"}\n";
 		assert_eq!(response, expected);
 
@@ -229,7 +213,11 @@ mod tests {
 	}
 
 	#[test]
-	fn swap_bad_request() -> Result<(), Box<dyn std::error::Error>> {
+	fn swap_bad_request() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+		let mut rt = tokio::runtime::Builder::new()
+			.threaded_scheduler()
+			.enable_all()
+			.build()?;
 		let server: Arc<Mutex<dyn SwapServer>> = Arc::new(Mutex::new(MockSwapServer::new()));
 
 		let params = "{ \"param\": \"Not a valid Swap request\" }";
@@ -237,7 +225,8 @@ mod tests {
 			"{{\"jsonrpc\": \"2.0\", \"method\": \"swap\", \"params\": [{}], \"id\": \"1\"}}",
 			params
 		);
-		let response = make_request(server, req)?;
+		let rt_handle = rt.handle().clone();
+		let response = rt.block_on(async_make_request(server, req, &rt_handle))?;
 		let expected = "{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32602,\"message\":\"Invalid params: missing field `onion`.\"},\"id\":\"1\"}\n";
 		assert_eq!(response, expected);
 		Ok(())
@@ -245,7 +234,12 @@ mod tests {
 
 	/// Returns "Commitment not found" when there's no matching output in the UTXO set.
 	#[test]
-	fn swap_utxo_missing() -> Result<(), Box<dyn std::error::Error>> {
+	fn swap_utxo_missing() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+		let mut rt = tokio::runtime::Builder::new()
+			.threaded_scheduler()
+			.enable_all()
+			.build()?;
+
 		let commitment = secp::commit(1234, &secp::random_secret())?;
 		let onion = create_onion(&commitment, &vec![])?;
 		let comsig = ComSignature::sign(1234, &secp::random_secret(), &onion.serialize()?)?;
@@ -267,7 +261,8 @@ mod tests {
 			"{{\"jsonrpc\": \"2.0\", \"method\": \"swap\", \"params\": [{}], \"id\": \"1\"}}",
 			serde_json::json!(swap)
 		);
-		let response = make_request(server, req)?;
+		let rt_handle = rt.handle().clone();
+		let response = rt.block_on(async_make_request(server, req, &rt_handle))?;
 		let expected = format!(
             "{{\"jsonrpc\":\"2.0\",\"error\":{{\"code\":-32602,\"message\":\"Output {:?} does not exist, or is already spent.\"}},\"id\":\"1\"}}\n",
             commitment

src/store.rs

@@ -21,8 +21,13 @@ const SWAP_PREFIX: u8 = b'S';
 #[derive(Clone, Debug, PartialEq)]
 pub enum SwapStatus {
 	Unprocessed,
-	InProcess { kernel_hash: Hash },
-	Completed { kernel_hash: Hash, block_hash: Hash },
+	InProcess {
+		kernel_commit: Commitment,
+	},
+	Completed {
+		kernel_commit: Commitment,
+		block_hash: Hash,
+	},
 	Failed,
 }
 
@@ -32,16 +37,16 @@ impl Writeable for SwapStatus {
 			SwapStatus::Unprocessed => {
 				writer.write_u8(0)?;
 			}
-			SwapStatus::InProcess { kernel_hash } => {
+			SwapStatus::InProcess { kernel_commit } => {
 				writer.write_u8(1)?;
-				kernel_hash.write(writer)?;
+				kernel_commit.write(writer)?;
 			}
 			SwapStatus::Completed {
-				kernel_hash,
+				kernel_commit,
 				block_hash,
 			} => {
 				writer.write_u8(2)?;
-				kernel_hash.write(writer)?;
+				kernel_commit.write(writer)?;
 				block_hash.write(writer)?;
 			}
 			SwapStatus::Failed => {
@@ -58,14 +63,14 @@ impl Readable for SwapStatus {
 		let status = match reader.read_u8()? {
 			0 => SwapStatus::Unprocessed,
 			1 => {
-				let kernel_hash = Hash::read(reader)?;
-				SwapStatus::InProcess { kernel_hash }
+				let kernel_commit = Commitment::read(reader)?;
+				SwapStatus::InProcess { kernel_commit }
 			}
 			2 => {
-				let kernel_hash = Hash::read(reader)?;
+				let kernel_commit = Commitment::read(reader)?;
 				let block_hash = Hash::read(reader)?;
 				SwapStatus::Completed {
-					kernel_hash,
+					kernel_commit,
 					block_hash,
 				}
 			}
@@ -244,8 +249,7 @@ impl SwapStore {
 
 #[cfg(test)]
 mod tests {
-	use crate::store::{SwapData, SwapStatus, SwapStore};
-	use crate::StoreError;
+	use crate::store::{StoreError, SwapData, SwapStatus, SwapStore};
 	use grin_core::core::{Input, OutputFeatures};
 	use grin_core::global::{self, ChainTypes};
 	use grin_onion::crypto::secp;
@@ -278,11 +282,11 @@ mod tests {
 			SwapStatus::Unprocessed
 		} else if s == 1 {
 			SwapStatus::InProcess {
-				kernel_hash: onion_test_util::rand_hash(),
+				kernel_commit: onion_test_util::rand_commit(),
 			}
 		} else {
 			SwapStatus::Completed {
-				kernel_hash: onion_test_util::rand_hash(),
+				kernel_commit: onion_test_util::rand_commit(),
 				block_hash: onion_test_util::rand_hash(),
 			}
 		};
@@ -330,7 +334,7 @@ mod tests {
 		assert!(store.swap_exists(&swap.input.commit)?);
 
 		swap.status = SwapStatus::InProcess {
-			kernel_hash: onion_test_util::rand_hash(),
+			kernel_commit: onion_test_util::rand_commit(),
 		};
 		let result = store.save_swap(&swap, false);
 		assert_eq!(

src/tor.rs

@@ -1,6 +1,5 @@
-use crate::config::{self, ServerConfig};
+use crate::config::ServerConfig;
 
-use grin_core::global;
 use grin_wallet_impls::tor::config as tor_config;
 use grin_wallet_impls::tor::process::TorProcess;
 use std::collections::HashMap;
@@ -10,72 +9,96 @@ use thiserror::Error;
 #[derive(Error, Debug)]
 pub enum TorError {
 	#[error("Error generating config: {0:?}")]
-	ConfigError(String),
+	ConfigError(grin_wallet_impls::Error),
 	#[error("Error starting process: {0:?}")]
 	ProcessError(grin_wallet_impls::tor::process::Error),
 }
 
-pub fn init_tor_listener(server_config: &ServerConfig) -> Result<TorProcess, TorError> {
-	println!("Initializing tor listener");
+pub fn init_tor_listener(
+	data_dir: &str,
+	server_config: &ServerConfig,
+) -> Result<TorProcess, TorError> {
+	warn!("Initializing tor listener");
 
-	let mut tor_dir = config::get_grin_path(&global::get_chain_type());
-	tor_dir.push("tor/listener");
+	let tor_dir = format!("{}/tor/listener", &data_dir);
+	trace!(
+		"Dir: {}, Proxy: {}",
+		&tor_dir,
+		server_config.socks_proxy_addr.to_string()
+	);
 
-	let mut torrc_dir = tor_dir.clone();
-	torrc_dir.push("torrc");
+	// create data directory if it doesn't exist
+	std::fs::create_dir_all(&format!("{}/data", tor_dir)).unwrap();
 
-	tor_config::output_tor_listener_config(
-		tor_dir.to_str().unwrap(),
+	let service_dir = tor_config::output_onion_service_config(tor_dir.as_str(), &server_config.key)
+		.map_err(|e| TorError::ConfigError(e))?;
+	let service_dirs = vec![service_dir.to_string()];
+
+	tor_config::output_torrc(
+		tor_dir.as_str(),
 		server_config.addr.to_string().as_str(),
-		&vec![server_config.key.clone()],
+		&server_config.socks_proxy_addr.to_string(),
+		&service_dirs,
 		HashMap::new(),
 		HashMap::new(),
 	)
-	.map_err(|e| TorError::ConfigError(e.to_string()))?;
+	.map_err(|e| TorError::ConfigError(e))?;
 
 	// Start TOR process
 	let mut process = TorProcess::new();
 	process
-		.torrc_path(torrc_dir.to_str().unwrap())
-		.working_dir(tor_dir.to_str().unwrap())
-		.timeout(20)
-		.completion_percent(100)
-		.launch()
-		.map_err(TorError::ProcessError)?;
+		.torrc_path("./torrc")
+		.working_dir(tor_dir.as_str())
+		.timeout(30)
+		.completion_percent(100);
 
-	println!(
+	let mut attempts = 0;
+	let max_attempts = 3;
+	let mut result;
+
+	loop {
+		attempts += 1;
+		info!("Launching TorProcess... Attempt {}", attempts);
+		result = process.launch();
+
+		if result.is_ok() || attempts >= max_attempts {
+			break;
+		}
+	}
+
+	result.map_err(TorError::ProcessError)?;
+
+	warn!(
 		"Server listening at http://{}.onion",
 		server_config.onion_address().to_ov3_str()
 	);
 	Ok(process)
 }
 
-pub fn init_tor_sender(server_config: &ServerConfig) -> Result<TorProcess, TorError> {
-	println!(
+pub fn init_tor_sender(
+	data_dir: &str,
+	server_config: &ServerConfig,
+) -> Result<TorProcess, TorError> {
+	warn!(
 		"Starting TOR Process for send at {:?}",
 		server_config.socks_proxy_addr
 	);
 
-	let mut tor_dir = config::get_grin_path(&global::get_chain_type());
-	tor_dir.push("tor/sender");
-
-	let mut torrc_dir = tor_dir.clone();
-	torrc_dir.push("torrc");
-
+	let tor_dir = format!("{}/tor/sender", data_dir);
 	tor_config::output_tor_sender_config(
-		tor_dir.to_str().unwrap(),
+		tor_dir.as_str(),
 		&server_config.socks_proxy_addr.to_string(),
 		HashMap::new(),
 		HashMap::new(),
 	)
-	.map_err(|e| TorError::ConfigError(e.to_string()))?;
+	.map_err(|e| TorError::ConfigError(e))?;
 
 	// Start TOR process
 	let mut tor_process = TorProcess::new();
 	tor_process
-		.torrc_path(torrc_dir.to_str().unwrap())
-		.working_dir(tor_dir.to_str().unwrap())
-		.timeout(20)
+		.torrc_path("./torrc")
+		.working_dir(tor_dir.as_str())
+		.timeout(40)
 		.completion_percent(100)
 		.launch()
 		.map_err(TorError::ProcessError)?;

src/tx.rs

@@ -1,10 +1,10 @@
-use crate::crypto::secp;
 use crate::wallet::Wallet;
 
 use grin_core::core::{
 	FeeFields, Input, Inputs, KernelFeatures, Output, Transaction, TransactionBody, TxKernel,
 };
 use grin_keychain::BlindingFactor;
+use grin_onion::crypto::secp;
 use secp256k1zkp::{ContextFlag, Secp256k1, SecretKey};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -43,7 +43,7 @@ pub struct TxComponents {
 }
 
 /// Builds and verifies the finalized swap 'Transaction' using the provided components.
-pub fn assemble_tx(
+pub async fn async_assemble_tx(
 	wallet: &Arc<dyn Wallet>,
 	inputs: &Vec<Input>,
 	outputs: &Vec<Output>,
@@ -57,7 +57,7 @@ pub fn assemble_tx(
 	let min_kernel_fee =
 		TransactionBody::weight_by_iok(inputs.len() as u64, outputs.len() as u64, 1) * fee_base;
 
-	let components = add_kernel_and_collect_fees(
+	let components = async_add_kernel_and_collect_fees(
 		&wallet,
 		&outputs,
 		&kernels,
@@ -66,7 +66,8 @@ pub fn assemble_tx(
 		fees_paid,
 		&prev_offset,
 		&output_excesses,
-	)?;
+	)
+	.await?;
 
 	// assemble the transaction
 	let tx = Transaction::new(
@@ -79,7 +80,7 @@ pub fn assemble_tx(
 }
 
 /// Adds a kernel and output to a collection of transaction components to consume fees and offset excesses.
-pub fn assemble_components(
+pub async fn async_assemble_components(
 	wallet: &Arc<dyn Wallet>,
 	components: &TxComponents,
 	output_excesses: &Vec<SecretKey>,
@@ -89,7 +90,7 @@ pub fn assemble_components(
 	// calculate minimum fee required for the kernel
 	let min_kernel_fee = TransactionBody::weight_by_iok(0, 0, 1) * fee_base;
 
-	add_kernel_and_collect_fees(
+	async_add_kernel_and_collect_fees(
 		&wallet,
 		&components.outputs,
 		&components.kernels,
@@ -99,9 +100,10 @@ pub fn assemble_components(
 		&components.offset,
 		&output_excesses,
 	)
+	.await
 }
 
-fn add_kernel_and_collect_fees(
+async fn async_add_kernel_and_collect_fees(
 	wallet: &Arc<dyn Wallet>,
 	outputs: &Vec<Output>,
 	kernels: &Vec<TxKernel>,
@@ -128,7 +130,10 @@ fn add_kernel_and_collect_fees(
 		let amount = fees_paid - (min_kernel_fee + fee_to_collect);
 		kernel_fee -= amount;
 
-		let wallet_output = wallet.build_output(amount).map_err(TxError::WalletError)?;
+		let wallet_output = wallet
+			.async_build_output(amount)
+			.await
+			.map_err(TxError::WalletError)?;
 		txn_outputs.push(wallet_output.1);
 
 		let output_excess = SecretKey::from_slice(&secp, &wallet_output.0.as_ref())
@@ -202,11 +207,12 @@ fn add_kernel_and_collect_fees(
 ///
 /// ```rust
 /// use secp256k1zkp::key::SecretKey;
-/// use crate::crypto::secp;
+/// use secp256k1zkp::rand::thread_rng;
+/// use grin_onion::crypto::secp;
 ///
-/// let secret_key = SecretKey::new(&mut secp::rand::thread_rng());
+/// let secret_key = secp::random_secret();
 /// let fee = 10; // 10 nanogrin
-/// let kernel = build_kernel(&secret_key, fee);
+/// let kernel = mwixnet::tx::build_kernel(&secret_key, fee);
 /// ```
 pub fn build_kernel(excess: &SecretKey, fee: u64) -> Result<TxKernel, TxError> {
 	let mut kernel = TxKernel::with_features(KernelFeatures::Plain {

src/wallet.rs

@@ -1,10 +1,10 @@
-use crate::crypto::secp;
-
+use async_trait::async_trait;
 use grin_api::client;
-use grin_api::json_rpc::{build_request, Request, Response};
+use grin_api::json_rpc::{build_request, Request, Response, RpcError};
 use grin_core::core::Output;
 use grin_core::libtx::secp_ser;
 use grin_keychain::BlindingFactor;
+use grin_onion::crypto::secp;
 use grin_util::{ToHex, ZeroingString};
 use grin_wallet_api::{EncryptedRequest, EncryptedResponse, JsonId, Token};
 use secp256k1zkp::{PublicKey, Secp256k1, SecretKey};
@@ -13,9 +13,13 @@ use serde_json::json;
 use std::net::SocketAddr;
 use thiserror::Error;
 
+#[async_trait]
 pub trait Wallet: Send + Sync {
 	/// Builds an output for the wallet with the provided amount.
-	fn build_output(&self, amount: u64) -> Result<(BlindingFactor, Output), WalletError>;
+	async fn async_build_output(
+		&self,
+		amount: u64,
+	) -> Result<(BlindingFactor, Output), WalletError>;
 }
 
 /// Error types for interacting with wallets
@@ -31,6 +35,8 @@ pub enum WalletError {
 	ApiCommError(grin_api::Error),
 	#[error("Error decoding JSON-RPC response: {0:?}")]
 	ResponseParseError(grin_api::json_rpc::Error),
+	#[error("Unsucessful response returned: {0:?}")]
+	ResponseRpcError(Option<RpcError>),
 }
 
 /// HTTP (JSONRPC) implementation of the 'Wallet' trait.
@@ -55,26 +61,27 @@ pub struct ECDHPubkey {
 
 impl HttpWallet {
 	/// Calls the 'open_wallet' using the RPC API.
-	pub fn open_wallet(
+	pub async fn async_open_wallet(
 		wallet_owner_url: &SocketAddr,
 		wallet_owner_secret: &Option<String>,
 		wallet_pass: &ZeroingString,
 	) -> Result<HttpWallet, WalletError> {
-		println!("Opening wallet at {}", wallet_owner_url);
-		let shared_key = HttpWallet::init_secure_api(&wallet_owner_url, &wallet_owner_secret)?;
-
+		info!("Opening wallet at {}", wallet_owner_url);
+		let shared_key =
+			HttpWallet::async_init_secure_api(&wallet_owner_url, &wallet_owner_secret).await?;
 		let open_wallet_params = json!({
 			"name": null,
 			"password": wallet_pass.to_string()
 		});
-		let token: Token = HttpWallet::send_enc_request(
+		let token: Token = HttpWallet::async_send_enc_request(
 			&wallet_owner_url,
 			&wallet_owner_secret,
 			"open_wallet",
 			&open_wallet_params,
 			&shared_key,
-		)?;
-		println!("Connected to wallet");
+		)
+		.await?;
+		info!("Connected to wallet");
 
 		Ok(HttpWallet {
 			wallet_owner_url: wallet_owner_url.clone(),
@@ -84,7 +91,7 @@ impl HttpWallet {
 		})
 	}
 
-	fn init_secure_api(
+	async fn async_init_secure_api(
 		wallet_owner_url: &SocketAddr,
 		wallet_owner_secret: &Option<String>,
 	) -> Result<SecretKey, WalletError> {
@@ -95,12 +102,13 @@ impl HttpWallet {
 			"ecdh_pubkey": ephemeral_pk.serialize_vec(&secp, true).to_hex()
 		});
 
-		let response_pk: ECDHPubkey = HttpWallet::send_json_request(
+		let response_pk: ECDHPubkey = HttpWallet::async_send_json_request(
 			&wallet_owner_url,
 			&wallet_owner_secret,
 			"init_secure_api",
 			&init_params,
-		)?;
+		)
+		.await?;
 
 		let shared_key = {
 			let mut shared_pubkey = response_pk.ecdh_pubkey.clone();
@@ -113,7 +121,22 @@ impl HttpWallet {
 		Ok(shared_key)
 	}
 
-	fn send_enc_request<D: serde::de::DeserializeOwned>(
+	pub async fn async_perform_request<D: serde::de::DeserializeOwned>(
+		&self,
+		method: &str,
+		params: &serde_json::Value,
+	) -> Result<D, WalletError> {
+		HttpWallet::async_send_enc_request(
+			&self.wallet_owner_url,
+			&self.wallet_owner_secret,
+			method,
+			params,
+			&self.shared_key,
+		)
+		.await
+	}
+
+	async fn async_send_enc_request<D: serde::de::DeserializeOwned>(
 		wallet_owner_url: &SocketAddr,
 		wallet_owner_secret: &Option<String>,
 		method: &str,
@@ -129,23 +152,30 @@ impl HttpWallet {
 		});
 		let enc_req = EncryptedRequest::from_json(&JsonId::IntId(1), &req, &shared_key)
 			.map_err(WalletError::EncryptRequestError)?;
-		let res = client::post::<EncryptedRequest, EncryptedResponse>(
+		let res = client::post_async::<EncryptedRequest, EncryptedResponse>(
 			url.as_str(),
-			wallet_owner_secret.clone(),
 			&enc_req,
+			wallet_owner_secret.clone(),
 		)
+		.await
 		.map_err(WalletError::ApiCommError)?;
 		let decrypted = res
 			.decrypt(&shared_key)
 			.map_err(WalletError::DecryptResponseError)?;
 		let response: Response =
 			serde_json::from_value(decrypted).map_err(WalletError::DecodeResponseError)?;
-		let ok = response.result.unwrap().get("Ok").unwrap().clone();
-		let parsed = serde_json::from_value(ok).map_err(WalletError::DecodeResponseError)?;
+		let result = response
+			.result
+			.ok_or(WalletError::ResponseRpcError(response.error.clone()))?;
+		let ok = result
+			.get("Ok")
+			.ok_or(WalletError::ResponseRpcError(response.error.clone()))?;
+		let parsed =
+			serde_json::from_value(ok.clone()).map_err(WalletError::DecodeResponseError)?;
 		Ok(parsed)
 	}
 
-	fn send_json_request<D: serde::de::DeserializeOwned>(
+	async fn async_send_json_request<D: serde::de::DeserializeOwned>(
 		wallet_owner_url: &SocketAddr,
 		wallet_owner_secret: &Option<String>,
 		method: &str,
@@ -153,15 +183,23 @@ impl HttpWallet {
 	) -> Result<D, WalletError> {
 		let url = format!("http://{}{}", wallet_owner_url, ENDPOINT);
 		let req = build_request(method, params);
-		let res =
-			client::post::<Request, Response>(url.as_str(), wallet_owner_secret.clone(), &req)
-				.map_err(WalletError::ApiCommError)?;
+		let res = client::post_async::<Request, Response>(
+			url.as_str(),
+			&req,
+			wallet_owner_secret.clone(),
+		)
+		.await
+		.map_err(WalletError::ApiCommError)?;
 		let parsed = res
 			.clone()
 			.into_result()
 			.map_err(WalletError::ResponseParseError)?;
 		Ok(parsed)
 	}
+
+	pub fn get_token(&self) -> Token {
+		self.token.clone()
+	}
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -174,21 +212,26 @@ pub struct OutputWithBlind {
 	output: Output,
 }
 
+#[async_trait]
 impl Wallet for HttpWallet {
 	/// Builds an 'Output' for the wallet using the 'build_output' RPC API.
-	fn build_output(&self, amount: u64) -> Result<(BlindingFactor, Output), WalletError> {
+	async fn async_build_output(
+		&self,
+		amount: u64,
+	) -> Result<(BlindingFactor, Output), WalletError> {
 		let req_json = json!({
-			"token": self.token.keychain_mask.clone().unwrap().0,
+			"token": self.token,
 			"features": "Plain",
 			"amount":  amount
 		});
-		let output: OutputWithBlind = HttpWallet::send_enc_request(
+		let output: OutputWithBlind = HttpWallet::async_send_enc_request(
 			&self.wallet_owner_url,
 			&self.wallet_owner_secret,
 			"build_output",
 			&req_json,
 			&self.shared_key,
-		)?;
+		)
+		.await?;
 		Ok((output.blind, output.output))
 	}
 }
@@ -196,11 +239,12 @@ impl Wallet for HttpWallet {
 #[cfg(test)]
 pub mod mock {
 	use super::{Wallet, WalletError};
-	use crate::crypto::secp;
 	use std::borrow::BorrowMut;
 
+	use async_trait::async_trait;
 	use grin_core::core::{Output, OutputFeatures};
 	use grin_keychain::BlindingFactor;
+	use grin_onion::crypto::secp;
 	use secp256k1zkp::pedersen::Commitment;
 	use secp256k1zkp::Secp256k1;
 	use std::sync::{Arc, Mutex};
@@ -225,9 +269,13 @@ pub mod mock {
 		}
 	}
 
+	#[async_trait]
 	impl Wallet for MockWallet {
 		/// Builds an 'Output' for the wallet using the 'build_output' RPC API.
-		fn build_output(&self, amount: u64) -> Result<(BlindingFactor, Output), WalletError> {
+		async fn async_build_output(
+			&self,
+			amount: u64,
+		) -> Result<(BlindingFactor, Output), WalletError> {
 			let secp = Secp256k1::new();
 			let blind = secp::random_secret();
 			let commit = secp::commit(amount, &blind).unwrap();