[package]
name = "magical-bitcoin-wallet"
version = "0.1.0"
+edition = "2018"
authors = ["Riccardo Casatta <riccardo@casatta.it>", "Alekos Filini <alekos.filini@gmail.com>"]
[dependencies]
serde = { version = "^1.0", features = ["derive"] }
serde_json = { version = "^1.0" }
base64 = "^0.11"
+async-trait = "0.1"
# Optional dependencies
sled = { version = "0.31.0", optional = true }
-electrum-client = { version = "0.1.0-beta.5", optional = true }
-reqwest = { version = "0.10", optional = true, features = ["blocking", "json"] }
+electrum-client = { git = "https://github.com/MagicalBitcoin/rust-electrum-client.git", optional = true }
+reqwest = { version = "0.10", optional = true, features = ["json"] }
+futures = { version = "0.3", optional = true }
[features]
minimal = []
compiler = ["miniscript/compiler"]
default = ["key-value-db", "electrum"]
electrum = ["electrum-client"]
-esplora = ["reqwest"]
+esplora = ["reqwest", "futures"]
key-value-db = ["sled"]
[dev-dependencies]
+tokio = { version = "0.2", features = ["macros"] }
lazy_static = "1.4"
rustyline = "5.0" # newer version requires 2018 edition
clap = "2.33"
use std::fs;
use std::path::PathBuf;
use std::str::FromStr;
+use std::sync::Arc;
use clap::{App, AppSettings, Arg, ArgMatches, SubCommand};
parse_outpoint(&s).map(|_| ())
}
-fn main() {
+#[tokio::main]
+async fn main() {
env_logger::init();
let app = App::new("Magical Bitcoin Wallet")
.unwrap();
debug!("database opened successfully");
- let client = Client::new(matches.value_of("server").unwrap()).unwrap();
+ let client = Client::new(matches.value_of("server").unwrap())
+ .await
+ .unwrap();
let wallet = Wallet::new(
descriptor,
change_descriptor,
tree,
ElectrumBlockchain::from(client),
)
+ .await
.unwrap();
+ let wallet = Arc::new(wallet);
// TODO: print errors in a nice way
- let handle_matches = |matches: ArgMatches<'_>| {
+ async fn handle_matches<C, D>(wallet: Arc<Wallet<C, D>>, matches: ArgMatches<'_>)
+ where
+ C: magical_bitcoin_wallet::blockchain::OnlineBlockchain,
+ D: magical_bitcoin_wallet::database::BatchDatabase,
+ {
if let Some(_sub_matches) = matches.subcommand_matches("get_new_address") {
println!("{}", wallet.get_new_address().unwrap().to_string());
} else if let Some(_sub_matches) = matches.subcommand_matches("sync") {
- wallet.sync(None, None).unwrap();
+ wallet.sync(None, None).await.unwrap();
} else if let Some(_sub_matches) = matches.subcommand_matches("list_unspent") {
for utxo in wallet.list_unspent().unwrap() {
println!("{} value {} SAT", utxo.outpoint, utxo.txout.value);
} else if let Some(sub_matches) = matches.subcommand_matches("broadcast") {
let psbt = base64::decode(sub_matches.value_of("psbt").unwrap()).unwrap();
let psbt: PartiallySignedTransaction = deserialize(&psbt).unwrap();
- let (txid, _) = wallet.broadcast(psbt).unwrap();
+ let (txid, _) = wallet.broadcast(psbt).await.unwrap();
println!("TXID: {}", txid);
}
continue;
}
- handle_matches(matches.unwrap());
+ handle_matches(Arc::clone(&wallet), matches.unwrap()).await;
}
Err(ReadlineError::Interrupted) => continue,
Err(ReadlineError::Eof) => break,
// rl.save_history("history.txt").unwrap();
} else {
- handle_matches(matches);
+ handle_matches(wallet, matches).await;
}
}
use std::collections::HashSet;
-use std::io::{Read, Write};
#[allow(unused_imports)]
use log::{debug, error, info, trace};
use bitcoin::{Script, Transaction, Txid};
+use electrum_client::tokio::io::{AsyncRead, AsyncWrite};
use electrum_client::Client;
use self::utils::{ELSGetHistoryRes, ELSListUnspentRes, ElectrumLikeSync};
use crate::database::{BatchDatabase, DatabaseUtils};
use crate::error::Error;
-pub struct ElectrumBlockchain<T: Read + Write>(Option<Client<T>>);
+pub struct ElectrumBlockchain<T: AsyncRead + AsyncWrite + Send>(Option<Client<T>>);
-impl<T: Read + Write> std::convert::From<Client<T>> for ElectrumBlockchain<T> {
+impl<T: AsyncRead + AsyncWrite + Send> std::convert::From<Client<T>> for ElectrumBlockchain<T> {
fn from(client: Client<T>) -> Self {
ElectrumBlockchain(Some(client))
}
}
-impl<T: Read + Write> Blockchain for ElectrumBlockchain<T> {
+impl<T: AsyncRead + AsyncWrite + Send> Blockchain for ElectrumBlockchain<T> {
fn offline() -> Self {
ElectrumBlockchain(None)
}
}
}
-impl<T: Read + Write> OnlineBlockchain for ElectrumBlockchain<T> {
- fn get_capabilities(&self) -> HashSet<Capability> {
+#[async_trait(?Send)]
+impl<T: AsyncRead + AsyncWrite + Send> OnlineBlockchain for ElectrumBlockchain<T> {
+ async fn get_capabilities(&self) -> HashSet<Capability> {
vec![Capability::FullHistory, Capability::GetAnyTx]
.into_iter()
.collect()
}
- fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+ async fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
&mut self,
stop_gap: Option<usize>,
database: &mut D,
.as_mut()
.ok_or(Error::OfflineClient)?
.electrum_like_setup(stop_gap, database, progress_update)
+ .await
}
- fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
+ async fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
Ok(self
.0
.as_mut()
.ok_or(Error::OfflineClient)?
.transaction_get(txid)
+ .await
.map(Option::Some)?)
}
- fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
+ async fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
Ok(self
.0
.as_mut()
.ok_or(Error::OfflineClient)?
.transaction_broadcast(tx)
+ .await
.map(|_| ())?)
}
- fn get_height(&mut self) -> Result<usize, Error> {
+ async fn get_height(&mut self) -> Result<usize, Error> {
// TODO: unsubscribe when added to the client, or is there a better call to use here?
Ok(self
.as_mut()
.ok_or(Error::OfflineClient)?
.block_headers_subscribe()
+ .await
.map(|data| data.height)?)
}
}
-impl<T: Read + Write> ElectrumLikeSync for Client<T> {
- fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
+#[async_trait(?Send)]
+impl<T: AsyncRead + AsyncWrite + Send> ElectrumLikeSync for Client<T> {
+ async fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error> {
self.batch_script_get_history(scripts)
+ .await
.map(|v| {
v.into_iter()
.map(|v| {
.map_err(Error::Electrum)
}
- fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
+ async fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSListUnspentRes>>, Error> {
self.batch_script_list_unspent(scripts)
+ .await
.map(|v| {
v.into_iter()
.map(|v| {
.map_err(Error::Electrum)
}
- fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
- self.transaction_get(txid).map_err(Error::Electrum)
+ async fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
+ self.transaction_get(txid).await.map_err(Error::Electrum)
}
}
use std::collections::HashSet;
+use futures::stream::{self, StreamExt, TryStreamExt};
+
#[allow(unused_imports)]
use log::{debug, error, info, trace};
use serde::Deserialize;
-use reqwest::blocking::Client;
+use reqwest::Client;
use reqwest::StatusCode;
use bitcoin::consensus::{deserialize, serialize};
}
}
+#[async_trait(?Send)]
impl OnlineBlockchain for EsploraBlockchain {
- fn get_capabilities(&self) -> HashSet<Capability> {
+ async fn get_capabilities(&self) -> HashSet<Capability> {
vec![Capability::FullHistory, Capability::GetAnyTx]
.into_iter()
.collect()
}
- fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+ async fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
&mut self,
stop_gap: Option<usize>,
database: &mut D,
.as_mut()
.ok_or(Error::OfflineClient)?
.electrum_like_setup(stop_gap, database, progress_update)
+ .await
}
- fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
- Ok(self.0.as_mut().ok_or(Error::OfflineClient)?._get_tx(txid)?)
+ async fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
+ Ok(self
+ .0
+ .as_mut()
+ .ok_or(Error::OfflineClient)?
+ ._get_tx(txid)
+ .await?)
}
- fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
+ async fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
Ok(self
.0
.as_mut()
.ok_or(Error::OfflineClient)?
- ._broadcast(tx)?)
+ ._broadcast(tx)
+ .await?)
}
- fn get_height(&mut self) -> Result<usize, Error> {
- Ok(self.0.as_mut().ok_or(Error::OfflineClient)?._get_height()?)
+ async fn get_height(&mut self) -> Result<usize, Error> {
+ Ok(self
+ .0
+ .as_mut()
+ .ok_or(Error::OfflineClient)?
+ ._get_height()
+ .await?)
}
}
sha256::Hash::hash(script.as_bytes()).into_inner().to_hex()
}
- fn _get_tx(&self, txid: &Txid) -> Result<Option<Transaction>, EsploraError> {
+ async fn _get_tx(&self, txid: &Txid) -> Result<Option<Transaction>, EsploraError> {
let resp = self
.client
.get(&format!("{}/api/tx/{}/raw", self.url, txid))
- .send()?;
+ .send()
+ .await?;
if let StatusCode::NOT_FOUND = resp.status() {
return Ok(None);
}
- Ok(Some(deserialize(&resp.error_for_status()?.bytes()?)?))
+ Ok(Some(deserialize(&resp.error_for_status()?.bytes().await?)?))
}
- fn _broadcast(&self, transaction: &Transaction) -> Result<(), EsploraError> {
+ async fn _broadcast(&self, transaction: &Transaction) -> Result<(), EsploraError> {
self.client
.post(&format!("{}/api/tx", self.url))
.body(serialize(transaction).to_hex())
- .send()?
+ .send()
+ .await?
.error_for_status()?;
Ok(())
}
- fn _get_height(&self) -> Result<usize, EsploraError> {
+ async fn _get_height(&self) -> Result<usize, EsploraError> {
Ok(self
.client
.get(&format!("{}/api/blocks/tip/height", self.url))
- .send()?
+ .send()
+ .await?
.error_for_status()?
- .text()?
+ .text()
+ .await?
.parse()?)
}
- fn _script_get_history(&self, script: &Script) -> Result<Vec<ELSGetHistoryRes>, EsploraError> {
+ async fn _script_get_history(
+ &self,
+ script: &Script,
+ ) -> Result<Vec<ELSGetHistoryRes>, EsploraError> {
let mut result = Vec::new();
let scripthash = Self::script_to_scripthash(script);
"{}/api/scripthash/{}/txs/mempool",
self.url, scripthash
))
- .send()?
+ .send()
+ .await?
.error_for_status()?
- .json::<Vec<EsploraGetHistory>>()?
+ .json::<Vec<EsploraGetHistory>>()
+ .await?
.into_iter()
.map(|x| ELSGetHistoryRes {
tx_hash: x.txid,
"{}/api/scripthash/{}/txs/chain/{}",
self.url, scripthash, last_txid
))
- .send()?
+ .send()
+ .await?
.error_for_status()?
- .json::<Vec<EsploraGetHistory>>()?;
+ .json::<Vec<EsploraGetHistory>>()
+ .await?;
let len = response.len();
if let Some(elem) = response.last() {
last_txid = elem.txid.to_hex();
Ok(result)
}
- fn _script_list_unspent(
+ async fn _script_list_unspent(
&self,
script: &Script,
) -> Result<Vec<ELSListUnspentRes>, EsploraError> {
self.url,
Self::script_to_scripthash(script)
))
- .send()?
+ .send()
+ .await?
.error_for_status()?
- .json::<Vec<EsploraListUnspent>>()?
+ .json::<Vec<EsploraListUnspent>>()
+ .await?
.into_iter()
.map(|x| ELSListUnspentRes {
tx_hash: x.txid,
}
}
+#[async_trait(?Send)]
impl ElectrumLikeSync for UrlClient {
- fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
+ async fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error> {
- Ok(scripts
- .into_iter()
- .map(|script| self._script_get_history(script))
- .collect::<Result<Vec<_>, _>>()?)
+ Ok(stream::iter(scripts)
+ .then(|script| self._script_get_history(&script))
+ .try_collect()
+ .await?)
}
- fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
+ async fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSListUnspentRes>>, Error> {
- Ok(scripts
- .into_iter()
- .map(|script| self._script_list_unspent(script))
- .collect::<Result<Vec<_>, _>>()?)
+ Ok(stream::iter(scripts)
+ .then(|script| self._script_list_unspent(&script))
+ .try_collect()
+ .await?)
}
- fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
+ async fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
Ok(self
- ._get_tx(txid)?
+ ._get_tx(txid)
+ .await?
.ok_or_else(|| EsploraError::TransactionNotFound(*txid))?)
}
}
}
}
+#[async_trait(?Send)]
pub trait OnlineBlockchain: Blockchain {
- fn get_capabilities(&self) -> HashSet<Capability>;
+ async fn get_capabilities(&self) -> HashSet<Capability>;
- fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+ async fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
&mut self,
stop_gap: Option<usize>,
database: &mut D,
progress_update: P,
) -> Result<(), Error>;
- fn sync<D: BatchDatabase + DatabaseUtils, P: Progress>(
+ async fn sync<D: BatchDatabase + DatabaseUtils, P: Progress>(
&mut self,
stop_gap: Option<usize>,
database: &mut D,
progress_update: P,
) -> Result<(), Error> {
- self.setup(stop_gap, database, progress_update)
+ self.setup(stop_gap, database, progress_update).await
}
- fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error>;
- fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error>;
+ async fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error>;
+ async fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error>;
- fn get_height(&mut self) -> Result<usize, Error>;
+ async fn get_height(&mut self) -> Result<usize, Error>;
}
pub type ProgressData = (f32, Option<String>);
}
/// Implements the synchronization logic for an Electrum-like client.
+#[async_trait(?Send)]
pub trait ElectrumLikeSync {
- fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
+ async fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error>;
- fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
+ async fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSListUnspentRes>>, Error>;
- fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error>;
+ async fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error>;
// Provided methods down here...
- fn electrum_like_setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+ async fn electrum_like_setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
&mut self,
stop_gap: Option<usize>,
database: &mut D,
let until = cmp::min(to_check_later.len(), batch_query_size);
let chunk: Vec<Script> = to_check_later.drain(..until).collect();
- let call_result = self.els_batch_script_get_history(chunk.iter())?;
+ let call_result = self.els_batch_script_get_history(chunk.iter()).await?;
for (script, history) in chunk.into_iter().zip(call_result.into_iter()) {
trace!("received history for {:?}, size {}", script, history.len());
last_found = index;
let mut check_later_scripts = self
- .check_history(database, script, history, &mut change_max_deriv)?
+ .check_history(database, script, history, &mut change_max_deriv)
+ .await?
.into_iter()
.filter(|x| already_checked.insert(x.clone()))
.collect();
let mut batch = database.begin_batch();
for chunk in ChunksIterator::new(database.iter_utxos()?.into_iter(), batch_query_size) {
let scripts: Vec<_> = chunk.iter().map(|u| &u.txout.script_pubkey).collect();
- let call_result = self.els_batch_script_list_unspent(scripts)?;
+ let call_result = self.els_batch_script_list_unspent(scripts).await?;
// check which utxos are actually still unspent
for (utxo, list_unspent) in chunk.into_iter().zip(call_result.iter()) {
Ok(())
}
- fn check_tx_and_descendant<D: DatabaseUtils + BatchDatabase>(
+ async fn check_tx_and_descendant<D: DatabaseUtils + BatchDatabase>(
&mut self,
database: &mut D,
txid: &Txid,
// went wrong
saved_tx.transaction.unwrap()
}
- None => self.els_transaction_get(&txid)?,
+ None => self.els_transaction_get(&txid).await?,
};
let mut incoming: u64 = 0;
Ok(to_check_later)
}
- fn check_history<D: DatabaseUtils + BatchDatabase>(
+ async fn check_history<D: DatabaseUtils + BatchDatabase>(
&mut self,
database: &mut D,
script_pubkey: Script,
x => u32::try_from(x).ok(),
};
- to_check_later.extend_from_slice(&self.check_tx_and_descendant(
- database,
- &tx.tx_hash,
- height,
- &script_pubkey,
- change_max_deriv,
- )?);
+ to_check_later.extend_from_slice(
+ &self
+ .check_tx_and_descendant(
+ database,
+ &tx.tx_hash,
+ height,
+ &script_pubkey,
+ change_max_deriv,
+ )
+ .await?,
+ );
}
Ok(to_check_later)
#[macro_use]
extern crate lazy_static;
+#[macro_use]
+extern crate async_trait;
+
#[cfg(feature = "electrum")]
pub extern crate electrum_client;
#[cfg(feature = "electrum")]
B: OnlineBlockchain,
D: BatchDatabase,
{
- pub fn new(
+ pub async fn new(
descriptor: &str,
change_descriptor: Option<&str>,
network: Network,
None => None,
};
- let current_height = Some(client.get_height()? as u32);
+ let current_height = Some(client.get_height().await? as u32);
Ok(Wallet {
descriptor,
})
}
- pub fn sync(
+ pub async fn sync(
&self,
max_address: Option<u32>,
_batch_query_size: Option<usize>,
self.database.borrow_mut().commit_batch(address_batch)?;
}
- self.client.borrow_mut().sync(
- None,
- self.database.borrow_mut().deref_mut(),
- noop_progress(),
- )
+ self.client
+ .borrow_mut()
+ .sync(
+ None,
+ self.database.borrow_mut().deref_mut(),
+ noop_progress(),
+ )
+ .await
}
- pub fn broadcast(&self, psbt: PSBT) -> Result<(Txid, Transaction), Error> {
+ pub async fn broadcast(&self, psbt: PSBT) -> Result<(Txid, Transaction), Error> {
let extracted = psbt.extract_tx();
- self.client.borrow_mut().broadcast(&extracted)?;
+ self.client.borrow_mut().broadcast(&extracted).await?;
Ok((extracted.txid(), extracted))
}
projects / bdk / commitdiff