frost/bdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make the blockchain interface async again on wasm32-unknown-unknown

Browse Source 
The procedural macro `#[maybe_async]` makes a method or every method of a trait
"async" whenever the target_arch is `wasm32`, and leaves them untouched on
every other platform.

The macro `maybe_await!($e:expr)` can be used to call `maybe_async` methods on
multi-platform code: it expands to `$e` on non-wasm32 platforms and to
`$e.await` on wasm32.

The macro `await_or_block!($e:expr)` can be used to contain async code as much
as possible: it expands to `tokio::runtime::Runtime::new().unwrap().block_on($e)`
on non-wasm32 platforms, and to `$e.await` on wasm32.
This commit is contained in:
Alekos Filini
2020-07-20 15:51:57 +02:00
parent 4a51d50e1f
commit 4fcf7ac89e
10 changed files with 254 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
src/blockchain/esplora.rs
View File

@@ -1,10 +1,7 @@
use std::collections::HashSet;
use std::sync::Mutex;
use futures::stream::{self, StreamExt, TryStreamExt};
use tokio::runtime::Runtime;
#[allow(unused_imports)]
use log::{debug, error, info, trace};
@@ -26,11 +23,8 @@ use crate::error::Error;
pub struct UrlClient {
url: String,
// We use the async client instead of the blocking one because it automatically uses `fetch`
// when the target platform is wasm32. For some reason the blocking client doesn't, so we are
// stuck with this
// when the target platform is wasm32.
client: Client,
runtime: Mutex<Runtime>,
}
#[derive(Debug)]
@@ -47,8 +41,6 @@ impl EsploraBlockchain {
EsploraBlockchain(Some(UrlClient {
url: base_url.to_string(),
client: Client::new(),
runtime: Mutex::new(Runtime::new().unwrap()),
}))
}
}
@@ -63,6 +55,7 @@ impl Blockchain for EsploraBlockchain {
}
}
#[maybe_async]
impl OnlineBlockchain for EsploraBlockchain {
fn get_capabilities(&self) -> HashSet<Capability> {
vec![Capability::FullHistory, Capability::GetAnyTx]
@@ -76,26 +69,35 @@ impl OnlineBlockchain for EsploraBlockchain {
database: &mut D,
progress_update: P,
) -> Result<(), Error> {
self.0
.as_mut()
.ok_or(Error::OfflineClient)?
.electrum_like_setup(stop_gap, database, progress_update)
}
fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
Ok(self.0.as_mut().ok_or(Error::OfflineClient)?._get_tx(txid)?)
}
fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
Ok(self
maybe_await!(self
.0
.as_mut()
.ok_or(Error::OfflineClient)?
._broadcast(tx)?)
.electrum_like_setup(stop_gap, database, progress_update))
}
fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
Ok(await_or_block!(self
.0
.as_mut()
.ok_or(Error::OfflineClient)?
._get_tx(txid))?)
}
fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
Ok(await_or_block!(self
.0
.as_mut()
.ok_or(Error::OfflineClient)?
._broadcast(tx))?)
}
fn get_height(&mut self) -> Result<usize, Error> {
Ok(self.0.as_mut().ok_or(Error::OfflineClient)?._get_height()?)
Ok(await_or_block!(self
.0
.as_mut()
.ok_or(Error::OfflineClient)?
._get_height())?)
}
}
@@ -104,54 +106,39 @@ impl UrlClient {
sha256::Hash::hash(script.as_bytes()).into_inner().to_hex()
}
fn _get_tx(&self, txid: &Txid) -> Result<Option<Transaction>, EsploraError> {
let resp = self.runtime.lock().unwrap().block_on(
self.client
.get(&format!("{}/api/tx/{}/raw", self.url, txid))
.send(),
)?;
async fn _get_tx(&self, txid: &Txid) -> Result<Option<Transaction>, EsploraError> {
let resp = self
.client
.get(&format!("{}/api/tx/{}/raw", self.url, txid))
.send()
.await?;
if let StatusCode::NOT_FOUND = resp.status() {
return Ok(None);
}
Ok(Some(deserialize(
&self
.runtime
.lock()
.unwrap()
.block_on(resp.error_for_status()?.bytes())?,
)?))
Ok(Some(deserialize(&resp.error_for_status()?.bytes().await?)?))
}
fn _broadcast(&self, transaction: &Transaction) -> Result<(), EsploraError> {
self.runtime
.lock()
.unwrap()
.block_on(
self.client
.post(&format!("{}/api/tx", self.url))
.body(serialize(transaction).to_hex())
.send(),
)?
async fn _broadcast(&self, transaction: &Transaction) -> Result<(), EsploraError> {
self.client
.post(&format!("{}/api/tx", self.url))
.body(serialize(transaction).to_hex())
.send()
.await?
.error_for_status()?;
Ok(())
}
fn _get_height(&self) -> Result<usize, EsploraError> {
let req = self.runtime.lock().unwrap().block_on(
self.client
.get(&format!("{}/api/blocks/tip/height", self.url))
.send(),
)?;
async fn _get_height(&self) -> Result<usize, EsploraError> {
let req = self
.client
.get(&format!("{}/api/blocks/tip/height", self.url))
.send()
.await?;
Ok(self
.runtime
.lock()
.unwrap()
.block_on(req.error_for_status()?.text())?
.parse()?)
Ok(req.error_for_status()?.text().await?.parse()?)
}
async fn _script_get_history(
@@ -247,34 +234,38 @@ impl UrlClient {
}
}
#[maybe_async]
impl ElectrumLikeSync for UrlClient {
fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error> {
self.runtime.lock().unwrap().block_on(async {
let future = async {
Ok(stream::iter(scripts)
.then(|script| self._script_get_history(&script))
.try_collect()
.await?)
})
};
await_or_block!(future)
}
fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
scripts: I,
) -> Result<Vec<Vec<ELSListUnspentRes>>, Error> {
self.runtime.lock().unwrap().block_on(async {
let future = async {
Ok(stream::iter(scripts)
.then(|script| self._script_list_unspent(&script))
.try_collect()
.await?)
})
};
await_or_block!(future)
}
fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
Ok(self
._get_tx(txid)?
Ok(await_or_block!(self._get_tx(txid))?
.ok_or_else(|| EsploraError::TransactionNotFound(*txid))?)
}
}

3
src/blockchain/mod.rs
View File

@@ -41,6 +41,7 @@ impl Blockchain for OfflineBlockchain {
}
}
#[maybe_async]
pub trait OnlineBlockchain: Blockchain {
fn get_capabilities(&self) -> HashSet<Capability>;
@@ -56,7 +57,7 @@ pub trait OnlineBlockchain: Blockchain {
database: &mut D,
progress_update: P,
) -> Result<(), Error> {
self.setup(stop_gap, database, progress_update)
maybe_await!(self.setup(stop_gap, database, progress_update))
}
fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error>;

25
src/blockchain/utils.rs
View File

@@ -27,6 +27,7 @@ pub struct ELSListUnspentRes {
}
/// Implements the synchronization logic for an Electrum-like client.
#[maybe_async]
pub trait ElectrumLikeSync {
fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
&mut self,
@@ -85,7 +86,7 @@ pub trait ElectrumLikeSync {
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 = maybe_await!(self.els_batch_script_get_history(chunk.iter()))?;
for (script, history) in chunk.into_iter().zip(call_result.into_iter()) {
trace!("received history for {:?}, size {}", script, history.len());
@@ -93,11 +94,15 @@ pub trait ElectrumLikeSync {
if !history.is_empty() {
last_found = index;
let mut check_later_scripts = self
.check_history(database, script, history, &mut change_max_deriv)?
.into_iter()
.filter(|x| already_checked.insert(x.clone()))
.collect();
let mut check_later_scripts = maybe_await!(self.check_history(
database,
script,
history,
&mut change_max_deriv
))?
.into_iter()
.filter(|x| already_checked.insert(x.clone()))
.collect();
to_check_later.append(&mut check_later_scripts);
}
@@ -124,7 +129,7 @@ pub trait ElectrumLikeSync {
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 = maybe_await!(self.els_batch_script_list_unspent(scripts))?;
// check which utxos are actually still unspent
for (utxo, list_unspent) in chunk.into_iter().zip(call_result.iter()) {
@@ -199,7 +204,7 @@ pub trait ElectrumLikeSync {
// went wrong
saved_tx.transaction.unwrap()
}
None => self.els_transaction_get(&txid)?,
None => maybe_await!(self.els_transaction_get(&txid))?,
};
let mut incoming: u64 = 0;
@@ -284,13 +289,13 @@ pub trait ElectrumLikeSync {
x => u32::try_from(x).ok(),
};
to_check_later.extend_from_slice(&self.check_tx_and_descendant(
to_check_later.extend_from_slice(&maybe_await!(self.check_tx_and_descendant(
database,
&tx.tx_hash,
height,
&script_pubkey,
change_max_deriv,
)?);
))?);
}
Ok(to_check_later)