perf(sync): sync addresses and messages concurrently
@@ -53,9 +53,6 @@ async fn sync_addresses(
) -> crate::Result<(Vec<Address>, Vec<(MessageId, IotaMessage)>)> {
let mut address_index = address_index;
- let client = crate::client::get_client(account.client_options());
- let client = client.read().unwrap();
-
let mut generated_addresses = vec![];
let mut found_messages = vec![];
loop {
@@ -69,9 +66,15 @@ async fn sync_addresses(
let mut curr_generated_addresses = vec![];
let mut curr_found_messages = vec![];
+ let mut futures_ = vec![];
for (iota_address_index, iota_address_internal, iota_address) in &generated_iota_addresses {
+ futures_.push(async move {
+ let client = crate::client::get_client(account.client_options());
+ let client = client.read().unwrap();
+
let address_outputs = client.get_address().outputs(&iota_address).await?;
let balance = client.get_address().balance(&iota_address).await?;
+ let mut curr_found_messages = vec![];
let mut curr_found_outputs: Vec<AddressOutput> = vec![];
for output in address_outputs.iter() {
@@ -99,7 +102,7 @@ async fn sync_addresses(
// ignore unused change addresses
if *iota_address_internal && curr_found_outputs.is_empty() {
- continue;
+ return crate::Result::Ok((curr_found_messages, None));
}
if let Some(account_address) = account.addresses().iter().find(|a| a.address() == iota_address) {
@@ -114,8 +117,18 @@ async fn sync_addresses(
.internal(*iota_address_internal)
.build()?;
+ crate::Result::Ok((curr_found_messages, Some(address)))
+ });
+ }
+
+ let results = futures::future::join_all(futures_).await;
+ for result in results {
+ let (found_messages, address_opt) = result?;
+ if let Some(address) = address_opt {
curr_generated_addresses.push(address);
}
+ curr_found_messages.extend(found_messages);
+ }
address_index += gap_limit;
@@ -142,13 +155,23 @@ async fn sync_messages(
stop_at_address_index: usize,
) -> crate::Result<Vec<(MessageId, IotaMessage)>> {
let mut messages = vec![];
- let client = crate::client::get_client(account.client_options());
+ let client_options = account.client_options().clone();
+
+ let futures_ = account
+ .addresses_mut()
+ .iter_mut()
+ .take(stop_at_address_index)
+ .map(|address| {
+ let client_options = client_options.clone();
+ async move {
+ let client = crate::client::get_client(&client_options);
let client = client.read().unwrap();
- for address in account.addresses_mut().iter_mut().take(stop_at_address_index) {
let address_outputs = client.get_address().outputs(address.address()).await?;
let balance = client.get_address().balance(address.address()).await?;
+
let mut outputs = vec![];
+ let mut messages = vec![];
for output in address_outputs.iter() {
let output = client.get_output(output).await?;
let output: AddressOutput = output.try_into()?;
@@ -162,6 +185,13 @@ async fn sync_messages(
address.filter_outputs(outputs);
address.set_balance(balance);
+
+ crate::Result::Ok(messages)
+ }
+ });
+
+ for res in futures::future::join_all(futures_).await {
+ messages.extend(res?);
}
Ok(messages)
