nik/liana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix subscriptions

Browse Source
This commit is contained in:
edouardparis 2025-01-28 17:34:27 +01:00
parent ad297d1ef8
commit a4adab5cd1
9 changed files with 494 additions and 487 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Cargo.lock generated
View File

@ -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"

2
liana-gui/Cargo.toml
View File

@ -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"

5
liana-gui/src/app/state/export.rs
View File

@ -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,

155
liana-gui/src/download.rs
View File

@ -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<I: 'static + Hash + Copy + Send + Sync, T: ToString>(
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<Progress, DownloadError>)> {
Subscription::run_with_id(
id,
download(url.to_string()).map(move |progress| (id, progress)),
)
}
fn download(url: String) -> impl Stream<Item = Result<Progress, DownloadError>> {
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<I> {
id: I,
url: String,
#[derive(Debug, Clone)]
pub enum Progress {
Downloading(f32),
Finished(Vec<u8>),
}
/// Possible errors with download.
#[derive(PartialEq, Eq, Debug, Clone)]
#[derive(Debug, Clone)]
pub enum DownloadError {
UnknownContentLength,
RequestError(String),
RequestFailed(Arc<reqwest::Error>),
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<I: Copy>(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<reqwest::Error> for DownloadError {
fn from(error: reqwest::Error) -> Self {
DownloadError::RequestFailed(Arc::new(error))
}
}
#[derive(Debug, Clone)]
pub enum Progress {
Started,
Advanced(f32),
Finished(Vec<u8>),
Errored(DownloadError),
}
pub enum State {
Ready(String),
Downloading {
response: reqwest::Response,
total: u64,
downloaded: u64,
bytes: Vec<u8>,
},
Finished,
}

93
liana-gui/src/export.rs
View File

@ -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<dyn Daemon + Sync + Send>,
path: PathBuf,
) -> impl Stream<Item = ExportProgress> {
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<PathBuf> {

579
liana-gui/src/hw.rs
View File

@ -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<HardwareWalletMessage> {
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<HardwareWallet> = Vec::new();
let mut still: Vec<String> = 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<Item = HardwareWalletMessage> {
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::<specter::SerialTransport>::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<HardwareWallet> = Vec::new();
let mut still: Vec<String> = 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::<specter::SerialTransport>::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<dyn HWI + Send + Sync> = 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<dyn HWI + Send + Sync> =
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::<ledger::TransportHID>::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::<ledger::TransportHID>::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::<ledger::TransportHID>::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::<ledger::TransportHID>::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>(

4
liana-gui/src/installer/message.rs
View File

@ -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<Progress, DownloadError>),
Install,
Start,
}

11
liana-gui/src/installer/step/node/bitcoind.rs
View File

@ -70,19 +70,16 @@ impl Download {
}
}
pub fn progress(&mut self, new_progress: download::Progress) {
pub fn progress(&mut self, new_progress: Result<download::Progress, download::DownloadError>) {
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);
}
}

117
liana-gui/src/loader.rs
View File

@ -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<Message> {
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<Item = Option<String>> {
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<dyn Daemon + Sync + Send>,
info: GetInfoResult,