From a4adab5cd1c99fb0a29526112ca8c541bd040655 Mon Sep 17 00:00:00 2001 From: edouardparis Date: Tue, 28 Jan 2025 17:34:27 +0100 Subject: [PATCH] fix subscriptions --- Cargo.lock | 15 + liana-gui/Cargo.toml | 2 +- liana-gui/src/app/state/export.rs | 5 +- liana-gui/src/download.rs | 155 ++--- liana-gui/src/export.rs | 93 ++- liana-gui/src/hw.rs | 579 +++++++++--------- liana-gui/src/installer/message.rs | 4 +- liana-gui/src/installer/step/node/bitcoind.rs | 11 +- liana-gui/src/loader.rs | 117 ++-- 9 files changed, 494 insertions(+), 487 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 12d890f0..2a315295 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4625,10 +4625,12 @@ dependencies = [ "system-configuration", "tokio", "tokio-rustls", + "tokio-util", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", "webpki-roots", "winreg", @@ -6025,6 +6027,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasm-timer" version = "0.2.5" diff --git a/liana-gui/Cargo.toml b/liana-gui/Cargo.toml index 0028c53b..2fe9d695 100644 --- a/liana-gui/Cargo.toml +++ b/liana-gui/Cargo.toml @@ -48,7 +48,7 @@ chrono = "0.4.38" # Used for managing internal bitcoind base64 = "0.21" bitcoin_hashes = "0.12" -reqwest = { version = "0.11", default-features=false, features = ["json", "rustls-tls"] } +reqwest = { version = "0.11", default-features=false, features = ["json", "rustls-tls", "stream"] } rust-ini = "0.19.0" rfd = "0.15.1" diff --git a/liana-gui/src/app/state/export.rs b/liana-gui/src/app/state/export.rs index 65e232a3..a517c0ee 100644 --- a/liana-gui/src/app/state/export.rs +++ b/liana-gui/src/app/state/export.rs @@ -115,10 +115,9 @@ impl ExportModal { if let Some(path) = &self.path { match &self.state { ExportState::Started | ExportState::Progress(_) => { - Some(iced::subscription::unfold( + Some(iced::Subscription::run_with_id( "transactions", - export::State::new(self.daemon.clone(), Box::new(path.to_path_buf())), - export::export_subscription, + export::export_subscription(self.daemon.clone(), path.to_path_buf()), )) } _ => None, diff --git a/liana-gui/src/download.rs b/liana-gui/src/download.rs index d2182a82..6423d6f1 100644 --- a/liana-gui/src/download.rs +++ b/liana-gui/src/download.rs @@ -1,135 +1,82 @@ // This is based on https://github.com/iced-rs/iced/blob/master/examples/download_progress/src/download.rs // with some modifications to store the downloaded bytes in `Progress::Finished` and `State::Downloading` // and to keep track of any download errors. -use iced::subscription; +use iced::futures::{SinkExt, Stream, StreamExt}; +use iced::stream::try_channel; +use iced::Subscription; use std::hash::Hash; +use std::sync::Arc; // Just a little utility function pub fn file( id: I, url: T, -) -> iced::Subscription<(I, Progress)> { - subscription::unfold(id, State::Ready(url.to_string()), move |state| { - download(id, state) +) -> iced::Subscription<(I, Result)> { + Subscription::run_with_id( + id, + download(url.to_string()).map(move |progress| (id, progress)), + ) +} + +fn download(url: String) -> impl Stream> { + try_channel(100, move |mut output| async move { + let response = reqwest::get(&url).await?; + let total = response + .content_length() + .ok_or(DownloadError::NoContentLength)?; + + let _ = output.send(Progress::Downloading(0.0)).await; + + let mut byte_stream = response.bytes_stream(); + let mut downloaded = 0; + let mut bytes = Vec::new(); + + while let Some(next_bytes) = byte_stream.next().await { + let chunk = next_bytes?; + downloaded += chunk.len(); + bytes.append(&mut chunk.to_vec()); + + let _ = output + .send(Progress::Downloading( + 100.0 * downloaded as f32 / total as f32, + )) + .await; + } + + let _ = output.send(Progress::Finished(bytes)).await; + + Ok(()) }) } -#[derive(Debug, Hash, Clone)] -pub struct Download { - id: I, - url: String, +#[derive(Debug, Clone)] +pub enum Progress { + Downloading(f32), + Finished(Vec), } -/// Possible errors with download. -#[derive(PartialEq, Eq, Debug, Clone)] +#[derive(Debug, Clone)] pub enum DownloadError { - UnknownContentLength, - RequestError(String), + RequestFailed(Arc), + NoContentLength, } impl std::fmt::Display for DownloadError { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { match self { - Self::UnknownContentLength => { + Self::NoContentLength => { write!(f, "Response has unknown content length.") } - Self::RequestError(e) => { + Self::RequestFailed(e) => { write!(f, "Request error: '{}'.", e) } } } } -async fn download(id: I, state: State) -> ((I, Progress), State) { - match state { - State::Ready(url) => { - let response = reqwest::get(&url).await; - - match response { - Ok(response) => { - if let Some(total) = response.content_length() { - ( - (id, Progress::Started), - State::Downloading { - response, - total, - downloaded: 0, - bytes: Vec::new(), - }, - ) - } else { - ( - (id, Progress::Errored(DownloadError::UnknownContentLength)), - State::Finished, - ) - } - } - Err(e) => ( - ( - id, - Progress::Errored(DownloadError::RequestError(e.to_string())), - ), - State::Finished, - ), - } - } - State::Downloading { - mut response, - total, - downloaded, - mut bytes, - } => match response.chunk().await { - Ok(Some(chunk)) => { - let downloaded = downloaded + chunk.len() as u64; - - let percentage = (downloaded as f32 / total as f32) * 100.0; - - bytes.append(&mut chunk.to_vec()); - - ( - (id, Progress::Advanced(percentage)), - State::Downloading { - response, - total, - downloaded, - bytes, - }, - ) - } - Ok(None) => ((id, Progress::Finished(bytes)), State::Finished), - Err(e) => ( - ( - id, - Progress::Errored(DownloadError::RequestError(e.to_string())), - ), - State::Finished, - ), - }, - State::Finished => { - // We do not let the stream die, as it would start a - // new download repeatedly if the user is not careful - // in case of errors. - iced::futures::future::pending().await - } +impl From for DownloadError { + fn from(error: reqwest::Error) -> Self { + DownloadError::RequestFailed(Arc::new(error)) } } - -#[derive(Debug, Clone)] -pub enum Progress { - Started, - Advanced(f32), - Finished(Vec), - Errored(DownloadError), -} - -pub enum State { - Ready(String), - Downloading { - response: reqwest::Response, - total: u64, - downloaded: u64, - bytes: Vec, - }, - Finished, -} diff --git a/liana-gui/src/export.rs b/liana-gui/src/export.rs index e7bec3c6..b530719a 100644 --- a/liana-gui/src/export.rs +++ b/liana-gui/src/export.rs @@ -17,6 +17,8 @@ use tokio::{ time::sleep, }; +use iced::futures::{SinkExt, Stream}; + use crate::{ app::view, daemon::{ @@ -343,42 +345,65 @@ impl State { } } -pub async fn export_subscription(mut state: State) -> (ExportProgress, State) { - match state.state() { - Status::Init => { - state.start().await; - } - Status::Stopped => { - sleep(time::Duration::from_millis(1000)).await; - return (ExportProgress::None, state); - } - Status::Running => { /* continue */ } - } - let msg = state.receiver.try_recv(); - let disconnected = match msg { - Ok(m) => return (m, state), - Err(e) => match e { - std::sync::mpsc::TryRecvError::Empty => false, - std::sync::mpsc::TryRecvError::Disconnected => true, - }, - }; +pub fn export_subscription( + daemon: Arc, + path: PathBuf, +) -> impl Stream { + iced::stream::channel(100, move |mut output| async move { + let mut state = State::new(daemon, Box::new(path)); + loop { + match state.state() { + Status::Init => { + state.start().await; + } + Status::Stopped => { + break; + } + Status::Running => { + sleep(time::Duration::from_millis(100)).await; + continue; + } + } + let msg = state.receiver.try_recv(); + let disconnected = match msg { + Ok(m) => { + let _ = output.send(m).await; + continue; + } + Err(e) => match e { + std::sync::mpsc::TryRecvError::Empty => false, + std::sync::mpsc::TryRecvError::Disconnected => true, + }, + }; - let handle = match state.handle.take() { - Some(h) => h, - None => return (ExportProgress::Error(Error::HandleLost), state), - }; - { - let h = handle.lock().expect("should not fail"); - if h.is_finished() { - return (ExportProgress::Finished, state); - } else if disconnected { - return (ExportProgress::Error(Error::ChannelLost), state); - } - } // => release handle lock - state.handle = Some(handle); + let handle = match state.handle.take() { + Some(h) => h, + None => { + let _ = output.send(ExportProgress::Error(Error::HandleLost)).await; + continue; + } + }; + let msg = { + let h = handle.lock().expect("should not fail"); + if h.is_finished() { + Some(ExportProgress::Finished) + } else if disconnected { + Some(ExportProgress::Error(Error::ChannelLost)) + } else { + None + } + }; + if let Some(msg) = msg { + let _ = output.send(msg).await; + continue; + } + // => release handle lock + state.handle = Some(handle); - sleep(time::Duration::from_millis(100)).await; - (ExportProgress::None, state) + sleep(time::Duration::from_millis(100)).await; + let _ = output.send(ExportProgress::None).await; + } + }) } pub async fn get_path() -> Option { diff --git a/liana-gui/src/hw.rs b/liana-gui/src/hw.rs index b217fa94..5ed01b60 100644 --- a/liana-gui/src/hw.rs +++ b/liana-gui/src/hw.rs @@ -12,6 +12,7 @@ use async_hwi::{ jade::{self, Jade}, ledger, specter, DeviceKind, Error as HWIError, Version, HWI, }; +use iced::futures::{SinkExt, Stream}; use liana::miniscript::bitcoin::{bip32::Fingerprint, hashes::hex::FromHex, Network}; use serde::{Deserialize, Serialize}; use tracing::{debug, warn}; @@ -313,17 +314,16 @@ impl HardwareWallets { } pub fn refresh(&self) -> iced::Subscription { - iced::subscription::unfold( + iced::Subscription::run_with_id( format!("refresh-{}", self.network), - State { + refresh(State { network: self.network, keys_aliases: self.aliases.clone(), wallet: self.wallet.clone(), connected_supported_hws: Vec::new(), api: None, datadir_path: self.datadir_path.clone(), - }, - refresh, + }), ) } } @@ -383,318 +383,333 @@ struct State { datadir_path: PathBuf, } -async fn refresh(mut state: State) -> (HardwareWalletMessage, State) { - let api = if let Some(api) = &mut state.api { - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - if let Err(e) = api.refresh_devices() { - return (HardwareWalletMessage::Error(e.to_string()), state); - }; - api - } else { - match ledger::HidApi::new() { - Ok(api) => { - state.api = Some(api); - state.api.as_mut().unwrap() - } - Err(e) => { - return (HardwareWalletMessage::Error(e.to_string()), state); - } - } - }; - - let mut hws: Vec = Vec::new(); - let mut still: Vec = Vec::new(); - match specter::SpecterSimulator::try_connect().await { - Ok(device) => { - let id = "specter-simulator".to_string(); - if state.connected_supported_hws.contains(&id) { - still.push(id); +fn refresh(mut state: State) -> impl Stream { + iced::stream::channel(100, move |mut output| async move { + loop { + let api = if let Some(api) = &mut state.api { + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + if let Err(e) = api.refresh_devices() { + let _ = output + .send(HardwareWalletMessage::Error(e.to_string())) + .await; + continue; + }; + api } else { - match HardwareWallet::new(id, Arc::new(device), Some(&state.keys_aliases)).await { - Ok(hw) => hws.push(hw), + match ledger::HidApi::new() { + Ok(api) => { + state.api = Some(api); + state.api.as_mut().unwrap() + } Err(e) => { - debug!("{}", e); + let _ = output + .send(HardwareWalletMessage::Error(e.to_string())) + .await; + continue; } } - } - } - Err(HWIError::DeviceNotFound) => {} - Err(e) => { - debug!("{}", e); - } - } + }; - match specter::SerialTransport::enumerate_potential_ports() { - Ok(ports) => { - for port in ports { - let id = format!("specter-{}", port); - if state.connected_supported_hws.contains(&id) { - still.push(id); - } else { - match specter::Specter::::new(port.clone()) { - Err(e) => { - warn!("{}", e); - } - Ok(device) => { - if tokio::time::timeout( - std::time::Duration::from_millis(500), - device.fingerprint(), - ) + let mut hws: Vec = Vec::new(); + let mut still: Vec = Vec::new(); + match specter::SpecterSimulator::try_connect().await { + Ok(device) => { + let id = "specter-simulator".to_string(); + if state.connected_supported_hws.contains(&id) { + still.push(id); + } else { + match HardwareWallet::new(id, Arc::new(device), Some(&state.keys_aliases)) .await - .is_ok() - { - match HardwareWallet::new( - id, - Arc::new(device), - Some(&state.keys_aliases), - ) - .await - { - Ok(hw) => hws.push(hw), - Err(e) => { - debug!("{}", e); + { + Ok(hw) => hws.push(hw), + Err(e) => { + debug!("{}", e); + } + } + } + } + Err(HWIError::DeviceNotFound) => {} + Err(e) => { + debug!("{}", e); + } + } + + match specter::SerialTransport::enumerate_potential_ports() { + Ok(ports) => { + for port in ports { + let id = format!("specter-{}", port); + if state.connected_supported_hws.contains(&id) { + still.push(id); + } else { + match specter::Specter::::new(port.clone()) { + Err(e) => { + warn!("{}", e); + } + Ok(device) => { + if tokio::time::timeout( + std::time::Duration::from_millis(500), + device.fingerprint(), + ) + .await + .is_ok() + { + match HardwareWallet::new( + id, + Arc::new(device), + Some(&state.keys_aliases), + ) + .await + { + Ok(hw) => hws.push(hw), + Err(e) => { + debug!("{}", e); + } + } } } } } } } + Err(e) => warn!("Error while listing specter wallets: {}", e), } - } - Err(e) => warn!("Error while listing specter wallets: {}", e), - } - match jade::SerialTransport::enumerate_potential_ports() { - Ok(ports) => { - for port in ports { - let id = format!("jade-{}", port); - if state.connected_supported_hws.contains(&id) { - still.push(id); - } else { - match jade::SerialTransport::new(port) { - Err(e) => { - warn!("{:?}", e); - } - Ok(device) => { - match handle_jade_device( - id, - state.network, - Jade::new(device).with_network(state.network), - state.wallet.as_ref().map(|w| w.as_ref()), - Some(&state.keys_aliases), - ) - .await - { - Ok(hw) => { - hws.push(hw); - } + match jade::SerialTransport::enumerate_potential_ports() { + Ok(ports) => { + for port in ports { + let id = format!("jade-{}", port); + if state.connected_supported_hws.contains(&id) { + still.push(id); + } else { + match jade::SerialTransport::new(port) { Err(e) => { warn!("{:?}", e); } + Ok(device) => { + match handle_jade_device( + id, + state.network, + Jade::new(device).with_network(state.network), + state.wallet.as_ref().map(|w| w.as_ref()), + Some(&state.keys_aliases), + ) + .await + { + Ok(hw) => { + hws.push(hw); + } + Err(e) => { + warn!("{:?}", e); + } + } + } } } } } + Err(e) => warn!("Error while listing jade devices: {}", e), } - } - Err(e) => warn!("Error while listing jade devices: {}", e), - } - match ledger::LedgerSimulator::try_connect().await { - Ok(device) => { - let id = "ledger-simulator".to_string(); - if state.connected_supported_hws.contains(&id) { - still.push(id); - } else { - match handle_ledger_device( - id, - device, - state.wallet.as_ref().map(|w| w.as_ref()), - &state.keys_aliases, - ) - .await - { - Ok(hw) => { - hws.push(hw); - } - Err(e) => { - warn!("{:?}", e); - } - } - } - } - Err(HWIError::DeviceNotFound) => {} - Err(e) => { - debug!("{}", e); - } - } - - for device_info in api.device_list() { - if async_hwi::bitbox::is_bitbox02(device_info) { - let id = format!( - "bitbox-{:?}-{}-{}", - device_info.path(), - device_info.vendor_id(), - device_info.product_id() - ); - if state.connected_supported_hws.contains(&id) { - still.push(id); - continue; - } - if let Ok(device) = device_info.open_device(api) { - if let Ok(device) = PairingBitbox02::connect( - device, - Some(Box::new(settings::global::PersistedBitboxNoiseConfig::new( - &state.datadir_path, - ))), - ) - .await - { - hws.push(HardwareWallet::Locked { - id, - kind: DeviceKind::BitBox02, - pairing_code: device.pairing_code().map(|s| s.replace('\n', " ")), - device: Arc::new(Mutex::new(Some(LockedDevice::BitBox02(Box::new( - device, - ))))), - }); - } - } - } - if device_info.vendor_id() == coldcard::api::COINKITE_VID - && device_info.product_id() == coldcard::api::CKCC_PID - { - let id = format!( - "coldcard-{:?}-{}-{}", - device_info.path(), - device_info.vendor_id(), - device_info.product_id() - ); - if state.connected_supported_hws.contains(&id) { - still.push(id); - continue; - } - if let Some(sn) = device_info.serial_number() { - if let Ok((cc, _)) = - coldcard::api::Coldcard::open(AsRefWrap { inner: api }, sn, None) - { - let device: Arc = if let Some(wallet) = &state.wallet { - coldcard::Coldcard::from(cc) - .with_wallet_name(wallet.name.clone()) - .into() + match ledger::LedgerSimulator::try_connect().await { + Ok(device) => { + let id = "ledger-simulator".to_string(); + if state.connected_supported_hws.contains(&id) { + still.push(id); } else { - coldcard::Coldcard::from(cc).into() - }; - match ( - device.get_master_fingerprint().await, - device.get_version().await, - ) { - (Ok(fingerprint), Ok(version)) => { - if version - >= (Version { - major: 6, - minor: 2, - patch: 1, - prerelease: None, - }) - { - hws.push(HardwareWallet::Supported { - id, - device, - kind: DeviceKind::Coldcard, - fingerprint, - version: Some(version), - registered: None, - alias: state.keys_aliases.get(&fingerprint).cloned(), - }); - } else { - hws.push(HardwareWallet::Unsupported { - id, - kind: device.device_kind(), - version: Some(version), - reason: UnsupportedReason::Version { - minimal_supported_version: "Edge firmware v6.2.1", - }, - }); + match handle_ledger_device( + id, + device, + state.wallet.as_ref().map(|w| w.as_ref()), + &state.keys_aliases, + ) + .await + { + Ok(hw) => { + hws.push(hw); + } + Err(e) => { + warn!("{:?}", e); + } + } + } + } + Err(HWIError::DeviceNotFound) => {} + Err(e) => { + debug!("{}", e); + } + } + + for device_info in api.device_list() { + if async_hwi::bitbox::is_bitbox02(device_info) { + let id = format!( + "bitbox-{:?}-{}-{}", + device_info.path(), + device_info.vendor_id(), + device_info.product_id() + ); + if state.connected_supported_hws.contains(&id) { + still.push(id); + continue; + } + if let Ok(device) = device_info.open_device(api) { + if let Ok(device) = PairingBitbox02::connect( + device, + Some(Box::new(settings::global::PersistedBitboxNoiseConfig::new( + &state.datadir_path, + ))), + ) + .await + { + hws.push(HardwareWallet::Locked { + id, + kind: DeviceKind::BitBox02, + pairing_code: device.pairing_code().map(|s| s.replace('\n', " ")), + device: Arc::new(Mutex::new(Some(LockedDevice::BitBox02( + Box::new(device), + )))), + }); + } + } + } + if device_info.vendor_id() == coldcard::api::COINKITE_VID + && device_info.product_id() == coldcard::api::CKCC_PID + { + let id = format!( + "coldcard-{:?}-{}-{}", + device_info.path(), + device_info.vendor_id(), + device_info.product_id() + ); + if state.connected_supported_hws.contains(&id) { + still.push(id); + continue; + } + if let Some(sn) = device_info.serial_number() { + if let Ok((cc, _)) = + coldcard::api::Coldcard::open(AsRefWrap { inner: api }, sn, None) + { + let device: Arc = + if let Some(wallet) = &state.wallet { + coldcard::Coldcard::from(cc) + .with_wallet_name(wallet.name.clone()) + .into() + } else { + coldcard::Coldcard::from(cc).into() + }; + match ( + device.get_master_fingerprint().await, + device.get_version().await, + ) { + (Ok(fingerprint), Ok(version)) => { + if version + >= (Version { + major: 6, + minor: 2, + patch: 1, + prerelease: None, + }) + { + hws.push(HardwareWallet::Supported { + id, + device, + kind: DeviceKind::Coldcard, + fingerprint, + version: Some(version), + registered: None, + alias: state.keys_aliases.get(&fingerprint).cloned(), + }); + } else { + hws.push(HardwareWallet::Unsupported { + id, + kind: device.device_kind(), + version: Some(version), + reason: UnsupportedReason::Version { + minimal_supported_version: "Edge firmware v6.2.1", + }, + }); + } + } + _ => tracing::error!("Failed to connect to coldcard"), } } - _ => tracing::error!("Failed to connect to coldcard"), } } } - } - } - for detected in ledger::Ledger::::enumerate(api) { - let id = format!( - "ledger-{:?}-{}-{}", - detected.path(), - detected.vendor_id(), - detected.product_id() - ); - if state.connected_supported_hws.contains(&id) { - still.push(id); - continue; - } - - match ledger::Ledger::::connect(api, detected) { - Ok(device) => match handle_ledger_device( - id, - device, - state.wallet.as_ref().map(|w| w.as_ref()), - &state.keys_aliases, - ) - .await - { - Ok(hw) => { - hws.push(hw); + for detected in ledger::Ledger::::enumerate(api) { + let id = format!( + "ledger-{:?}-{}-{}", + detected.path(), + detected.vendor_id(), + detected.product_id() + ); + if state.connected_supported_hws.contains(&id) { + still.push(id); + continue; } - Err(e) => { - warn!("{:?}", e); - } - }, - Err(HWIError::DeviceNotFound) => {} - Err(e) => { - debug!("{}", e); - } - } - } - if let Some(wallet) = &state.wallet { - let wallet_keys = wallet.descriptor_keys(); - for hw in &mut hws { - if let HardwareWallet::Supported { - fingerprint, - id, - kind, - version, - .. - } = &hw - { - if !wallet_keys.contains(fingerprint) { - *hw = HardwareWallet::Unsupported { - id: id.clone(), - kind: *kind, - version: version.clone(), - reason: UnsupportedReason::NotPartOfWallet(*fingerprint), - }; + match ledger::Ledger::::connect(api, detected) { + Ok(device) => match handle_ledger_device( + id, + device, + state.wallet.as_ref().map(|w| w.as_ref()), + &state.keys_aliases, + ) + .await + { + Ok(hw) => { + hws.push(hw); + } + Err(e) => { + warn!("{:?}", e); + } + }, + Err(HWIError::DeviceNotFound) => {} + Err(e) => { + debug!("{}", e); + } } } - } - } - state.connected_supported_hws = still - .iter() - .chain(hws.iter().filter_map(|hw| match hw { - HardwareWallet::Locked { id, .. } => Some(id), - HardwareWallet::Supported { id, .. } => Some(id), - HardwareWallet::Unsupported { .. } => None, - })) - .cloned() - .collect(); - ( - HardwareWalletMessage::List(ConnectedList { new: hws, still }), - state, - ) + if let Some(wallet) = &state.wallet { + let wallet_keys = wallet.descriptor_keys(); + for hw in &mut hws { + if let HardwareWallet::Supported { + fingerprint, + id, + kind, + version, + .. + } = &hw + { + if !wallet_keys.contains(fingerprint) { + *hw = HardwareWallet::Unsupported { + id: id.clone(), + kind: *kind, + version: version.clone(), + reason: UnsupportedReason::NotPartOfWallet(*fingerprint), + }; + } + } + } + } + + state.connected_supported_hws = still + .iter() + .chain(hws.iter().filter_map(|hw| match hw { + HardwareWallet::Locked { id, .. } => Some(id), + HardwareWallet::Supported { id, .. } => Some(id), + HardwareWallet::Unsupported { .. } => None, + })) + .cloned() + .collect(); + let _ = output + .send(HardwareWalletMessage::List(ConnectedList { + new: hws, + still, + })) + .await; + } + }) } async fn handle_ledger_device<'a, T: async_hwi::ledger::Transport + Sync + Send + 'static>( diff --git a/liana-gui/src/installer/message.rs b/liana-gui/src/installer/message.rs index 96f5acd1..551f08e9 100644 --- a/liana-gui/src/installer/message.rs +++ b/liana-gui/src/installer/message.rs @@ -6,7 +6,7 @@ use std::path::PathBuf; use super::{context, Error}; use crate::{ - download::Progress, + download::{DownloadError, Progress}, hw::HardwareWalletMessage, installer::step::descriptor::editor::key::Key, lianalite::client::{auth::AuthClient, backend::api}, @@ -103,7 +103,7 @@ pub enum InternalBitcoindMsg { Reload, DefineConfig, Download, - DownloadProgressed(Progress), + DownloadProgressed(Result), Install, Start, } diff --git a/liana-gui/src/installer/step/node/bitcoind.rs b/liana-gui/src/installer/step/node/bitcoind.rs index 4916b828..c23e6efb 100644 --- a/liana-gui/src/installer/step/node/bitcoind.rs +++ b/liana-gui/src/installer/step/node/bitcoind.rs @@ -70,19 +70,16 @@ impl Download { } } - pub fn progress(&mut self, new_progress: download::Progress) { + pub fn progress(&mut self, new_progress: Result) { if let DownloadState::Downloading { progress } = &mut self.state { match new_progress { - download::Progress::Started => { - *progress = 0.0; - } - download::Progress::Advanced(percentage) => { + Ok(download::Progress::Downloading(percentage)) => { *progress = percentage; } - download::Progress::Finished(bytes) => { + Ok(download::Progress::Finished(bytes)) => { self.state = DownloadState::Finished(bytes); } - download::Progress::Errored(e) => { + Err(e) => { self.state = DownloadState::Errored(e); } } diff --git a/liana-gui/src/loader.rs b/liana-gui/src/loader.rs index ac6a93a2..005f5bd4 100644 --- a/liana-gui/src/loader.rs +++ b/liana-gui/src/loader.rs @@ -5,6 +5,8 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; +use iced::futures::{SinkExt, Stream}; +use iced::stream::channel; use iced::{Alignment, Length, Subscription, Task}; use tokio::runtime::Handle; use tracing::{debug, info, warn}; @@ -313,60 +315,8 @@ impl Loader { pub fn subscription(&self) -> Subscription { if self.internal_bitcoind.is_some() { let log_path = internal_bitcoind_debug_log_path(&self.datadir_path, self.network); - iced::Subscription::unfold(0, log_path, move |log_path| async move { - // Reduce the io load. - tokio::time::sleep(Duration::from_millis(500)).await; - - // Open the log file and seek to its end, with some breathing room to make sure - // we don't skip all "UpdateTip" lines. This is to avoid making BufReader read - // the whole file every single time below. - let mut file = match File::open(&log_path) { - Ok(file) => file, - Err(e) => { - log::warn!("Opening bitcoind log file: {}", e); - return (Message::None, log_path); - } - }; - match file.metadata() { - Ok(m) => { - let file_len = m.len(); - let offset = 1024 * 1024; - if file_len > offset { - if let Err(e) = - file.seek(SeekFrom::Start(file_len.saturating_sub(offset))) - { - log::error!("Seeking to end of bitcoind log file: {}", e); - } - } - } - Err(e) => { - log::error!("Getting bitcoind log file metadata: {}", e); - } - }; - - // Find the latest tip update line in bitcoind's debug.log. BufReader is only - // used to facilitates searching through the lines. - let reader = BufReader::new(file); - let last_update_tip = reader - .lines() - .filter(|l| { - l.as_ref() - .map(|l| l.contains("UpdateTip") || l.contains("blockheaders")) - .unwrap_or(false) - }) - .last(); - match last_update_tip { - Some(Ok(line)) => (Message::BitcoindLog(Some(line)), log_path), - res => { - if let Some(Err(e)) = res { - log::error!("Reading bitcoind log file: {}", e); - } else { - log::warn!("Couldn't find an UpdateTip line in bitcoind log file."); - } - (Message::None, log_path) - } - } - }) + iced::Subscription::run_with_id("bitcoind_log", get_bitcoind_log(log_path)) + .map(|msg| Message::BitcoindLog(msg)) } else { Subscription::none() } @@ -377,6 +327,65 @@ impl Loader { } } +fn get_bitcoind_log(log_path: PathBuf) -> impl Stream> { + channel(1, move |mut output| async move { + loop { + // Reduce the io load. + tokio::time::sleep(Duration::from_millis(500)).await; + + // Open the log file and seek to its end, with some breathing room to make sure + // we don't skip all "UpdateTip" lines. This is to avoid making BufReader read + // the whole file every single time below. + let mut file = match File::open(&log_path) { + Ok(file) => file, + Err(e) => { + log::warn!("Opening bitcoind log file: {}", e); + continue; + } + }; + match file.metadata() { + Ok(m) => { + let file_len = m.len(); + let offset = 1024 * 1024; + if file_len > offset { + if let Err(e) = file.seek(SeekFrom::Start(file_len.saturating_sub(offset))) + { + log::error!("Seeking to end of bitcoind log file: {}", e); + } + } + } + Err(e) => { + log::error!("Getting bitcoind log file metadata: {}", e); + } + }; + + // Find the latest tip update line in bitcoind's debug.log. BufReader is only + // used to facilitates searching through the lines. + let reader = BufReader::new(file); + let last_update_tip = reader + .lines() + .filter(|l| { + l.as_ref() + .map(|l| l.contains("UpdateTip") || l.contains("blockheaders")) + .unwrap_or(false) + }) + .last(); + match last_update_tip { + Some(Ok(line)) => { + let _ = output.send(Some(line)).await; + } + res => { + if let Some(Err(e)) = res { + log::error!("Reading bitcoind log file: {}", e); + } else { + log::warn!("Couldn't find an UpdateTip line in bitcoind log file."); + } + } + } + } + }) +} + pub async fn load_application( daemon: Arc, info: GetInfoResult,