Add live P2Pool websocket integration for shares and peers

Cargo.toml

@@ -17,8 +17,17 @@ unicode-width = "0.2"
 p2poolv2_config = { git = "https://github.com/p2poolv2/p2poolv2", package = "p2poolv2_config" }
 bitcoin = "0.32.5"
 toml_edit = "0.22"
+reqwest = { version = "0.12", features = ["json", "rustls-tls", "blocking"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time"] }
+base64 = "0.22.1"
+futures-util = "0.3"
+tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] }
+url = "2"
+serde_json = "1.0.135"
 
 [dev-dependencies]
 insta = "1.44.3"
 serial_test = "3"
 tempfile = "3"
+mockito = "1.7.2"
+serde_json = "1.0.149"

config/config.toml

@@ -0,0 +1,7 @@
+[api]
+base_url = "http://127.0.0.1:46884"
+# fallback_base_url = "http://127.0.0.1:46885"
+host = "127.0.0.1"
+port = 46884
+auth_user = "p2pool"
+auth_pass = "p2pool"

src/app.rs

@@ -5,11 +5,16 @@
 use crate::bitcoin_config::ConfigEntry as BitcoinEntry;
 use crate::components::bitcoin_config_view::BitcoinConfigView;
 use crate::components::file_explorer::FileExplorer;
+use crate::components::p2pool_client::{ChainInfo, P2PoolClient, PeerInfo, SharesResponse};
 use crate::components::p2pool_config_view::P2PoolConfigView;
+use crate::components::p2pool_websocket::{
+    LiveP2PoolEvent, LivePeerEvent, LiveShare, P2PoolWebSocketClient,
+};
 use crate::components::settings_view::SettingsView;
 use crate::settings::Settings;
 use p2poolv2_config::Config as P2PoolConfig;
 use std::path::PathBuf;
+use tokio::sync::mpsc;
 
 /// Sidebar items labels
 pub const SIDEBAR_ITEMS: &[(&str, CurrentScreen)] = &[
@@ -31,6 +36,11 @@ pub const BITCOIN_STATUS_TABS: &[&str] = &["Chain Info", "System", "Logs", "Peer
 
 pub const MAX_BITCOIN_STATUS_TAB: usize = BITCOIN_STATUS_TABS.len() - 1;
 
+/// Tab labels for the P2Pool Status view
+pub const P2POOL_STATUS_TABS: &[&str] = &["Chain Info", "Shares", "Peers Info"];
+
+pub const MAX_P2POOL_STATUS_TAB: usize = P2POOL_STATUS_TABS.len() - 1;
+
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum CurrentScreen {
     Home,
@@ -96,17 +106,44 @@ pub struct App {
     pub bitcoin_data: Vec<BitcoinEntry>,
     pub bitcoin_status_tab: usize,
     pub settings: Settings,
+    pub p2pool_client: P2PoolClient,
+    pub p2pool_websocket_client: P2PoolWebSocketClient,
     /// Cached value of the `HOME` environment variable, used for path display.
     /// Populated once at startup to avoid repeated syscalls during rendering.
     pub home_dir: String,
     /// Cached result of `settings::config_dir()`, used to display the default
     /// settings storage path without repeated env-var lookups during rendering.
     pub config_dir: PathBuf,
+    pub p2pool_status_tab: usize,
+    pub chain_info: Option<ChainInfo>,
+    pub p2pool_chain_info_error: Option<String>,
+    pub share_info: Option<SharesResponse>,
+    pub p2pool_share_info_error: Option<String>,
+    pub peer_info: Option<Vec<PeerInfo>>,
+    pub p2pool_peer_info_error: Option<String>,
+    pub live_shares: Vec<LiveShare>,
+    pub live_peer_events: Vec<LivePeerEvent>,
+    pub p2pool_live_error: Option<String>,
+    pub p2pool_live_stream_started: bool,
+    pub p2pool_live_tx: mpsc::UnboundedSender<anyhow::Result<LiveP2PoolEvent>>,
+    pub p2pool_live_rx: mpsc::UnboundedReceiver<anyhow::Result<LiveP2PoolEvent>>,
+    // async channel to receive chain info updates from the background task that
+    // fetches it when the P2Pool Status screen is opened.
+    pub chain_info_tx: mpsc::UnboundedSender<anyhow::Result<ChainInfo>>,
+    pub chain_info_rx: mpsc::UnboundedReceiver<anyhow::Result<ChainInfo>>,
+    pub share_info_tx: mpsc::UnboundedSender<anyhow::Result<SharesResponse>>,
+    pub share_info_rx: mpsc::UnboundedReceiver<anyhow::Result<SharesResponse>>,
+    pub peer_info_tx: mpsc::UnboundedSender<anyhow::Result<Vec<PeerInfo>>>,
+    pub peer_info_rx: mpsc::UnboundedReceiver<anyhow::Result<Vec<PeerInfo>>>,
 }
 
 impl App {
     #[must_use]
     pub fn new() -> App {
+        let (chain_info_tx, chain_info_rx) = mpsc::unbounded_channel();
+        let (peer_info_tx, peer_info_rx) = mpsc::unbounded_channel();
+        let (share_info_tx, share_info_rx) = mpsc::unbounded_channel();
+        let (p2pool_live_tx, p2pool_live_rx) = mpsc::unbounded_channel();
         App {
             current_screen: CurrentScreen::Home,
             sidebar_index: 0,
@@ -121,8 +158,134 @@ impl App {
             bitcoin_data: Vec::new(),
             bitcoin_status_tab: 0,
             settings: Settings::default(),
+            p2pool_client: P2PoolClient::new(),
+            p2pool_websocket_client: P2PoolWebSocketClient::new(),
             home_dir: std::env::var("HOME").unwrap_or_default(),
             config_dir: crate::settings::config_dir().unwrap_or_default(),
+            p2pool_status_tab: 0,
+            chain_info: None,
+            p2pool_chain_info_error: None,
+            share_info: None,
+            p2pool_share_info_error: None,
+            peer_info: None,
+            p2pool_peer_info_error: None,
+            live_shares: Vec::new(),
+            live_peer_events: Vec::new(),
+            p2pool_live_error: None,
+            p2pool_live_stream_started: false,
+            p2pool_live_tx,
+            p2pool_live_rx,
+            chain_info_tx,
+            chain_info_rx,
+            share_info_tx,
+            share_info_rx,
+            peer_info_tx,
+            peer_info_rx,
+        }
+    }
+
+    #[must_use]
+    pub fn new_with_client(client: P2PoolClient) -> App {
+        let mut app = App::new();
+        app.p2pool_websocket_client = client.websocket_client();
+        app.p2pool_client = client;
+        app
+    }
+
+    /// Non-blocking result handler
+    pub fn poll_chain_info(&mut self) {
+        while let Ok(result) = self.chain_info_rx.try_recv() {
+            match result {
+                Ok(info) => {
+                    self.chain_info = Some(info);
+                    self.p2pool_chain_info_error = None;
+                }
+                Err(e) => {
+                    self.chain_info = None;
+                    self.p2pool_chain_info_error = Some(e.to_string());
+                }
+            }
+        }
+    }
+
+    pub fn poll_peer_info(&mut self) {
+        while let Ok(result) = self.peer_info_rx.try_recv() {
+            match result {
+                Ok(info) => {
+                    self.peer_info = Some(info);
+                    self.p2pool_peer_info_error = None;
+                }
+                Err(e) => {
+                    self.peer_info = None;
+                    self.p2pool_peer_info_error = Some(e.to_string());
+                }
+            }
+        }
+    }
+
+    pub fn poll_share_info(&mut self) {
+        while let Ok(result) = self.share_info_rx.try_recv() {
+            match result {
+                Ok(info) => {
+                    self.share_info = Some(info);
+                    self.p2pool_share_info_error = None;
+                }
+                Err(e) => {
+                    self.share_info = None;
+                    self.p2pool_share_info_error = Some(e.to_string());
+                }
+            }
+        }
+    }
+
+    pub fn poll_live_p2pool_events(&mut self) {
+        while let Ok(result) = self.p2pool_live_rx.try_recv() {
+            match result {
+                Ok(LiveP2PoolEvent::Share(share)) => {
+                    Self::push_limited(&mut self.live_shares, share, 50);
+                    self.p2pool_live_error = None;
+                }
+                Ok(LiveP2PoolEvent::Peer(peer_event)) => {
+                    self.apply_live_peer_event(&peer_event);
+                    Self::push_limited(&mut self.live_peer_events, peer_event, 50);
+                    self.p2pool_live_error = None;
+                }
+                Err(e) => {
+                    self.p2pool_live_error = Some(e.to_string());
+                    self.p2pool_live_stream_started = false;
+                }
+            }
+        }
+    }
+
+    pub fn poll_live_shares(&mut self) {
+        self.poll_live_p2pool_events();
+    }
+
+    fn push_limited<T>(items: &mut Vec<T>, item: T, max_len: usize) {
+        items.push(item);
+        if items.len() > max_len {
+            let extra = items.len() - max_len;
+            items.drain(0..extra);
+        }
+    }
+
+    fn apply_live_peer_event(&mut self, event: &LivePeerEvent) {
+        if event.status.eq_ignore_ascii_case("disconnected") {
+            if let Some(peers) = &mut self.peer_info {
+                peers.retain(|peer| peer.peer_id != event.peer_id);
+            }
+            return;
+        }
+
+        let peers = self.peer_info.get_or_insert_with(Vec::new);
+        if let Some(peer) = peers.iter_mut().find(|peer| peer.peer_id == event.peer_id) {
+            peer.status = Some(event.status.clone());
+        } else {
+            peers.push(PeerInfo {
+                peer_id: event.peer_id.clone(),
+                status: Some(event.status.clone()),
+            });
         }
     }
 
@@ -142,6 +305,46 @@ impl App {
         }
         if let Some(&(_, screen)) = SIDEBAR_ITEMS.get(self.sidebar_index) {
             self.current_screen = screen;
+            if self.current_screen == CurrentScreen::P2PoolStatus {
+                let chain_client = self.p2pool_client.clone();
+                let chain_tx = self.chain_info_tx.clone();
+                let share_client = self.p2pool_client.clone();
+                let share_tx = self.share_info_tx.clone();
+                let peer_client = self.p2pool_client.clone();
+                let peer_tx = self.peer_info_tx.clone();
+                let websocket_client = self.p2pool_websocket_client.clone();
+                let live_tx = self.p2pool_live_tx.clone();
+                let start_live_stream = !self.p2pool_live_stream_started;
+
+                if let Ok(handle) = tokio::runtime::Handle::try_current() {
+                    handle.spawn(async move {
+                        let res = chain_client.fetch_chain_info().await;
+                        let _ = chain_tx.send(res.map_err(anyhow::Error::from));
+                    });
+
+                    handle.spawn(async move {
+                        let res = share_client.fetch_recent_shares(10).await;
+                        let _ = share_tx.send(res.map_err(anyhow::Error::from));
+                    });
+
+                    handle.spawn(async move {
+                        let res = peer_client.fetch_peer_info().await;
+                        let _ = peer_tx.send(res.map_err(anyhow::Error::from));
+                    });
+
+                    if start_live_stream {
+                        self.p2pool_live_stream_started = true;
+                        handle.spawn(async move {
+                            if let Err(error) = websocket_client
+                                .subscribe_live_events(live_tx.clone())
+                                .await
+                            {
+                                let _ = live_tx.send(Err(error));
+                            }
+                        });
+                    }
+                }
+            }
         }
     }
 }

src/components/mod.rs

@@ -8,8 +8,10 @@ pub mod file_explorer;
 pub mod home_view;
 pub mod ln_config_view;
 pub mod ln_status_view;
+pub mod p2pool_client;
 pub mod p2pool_config_view;
 pub mod p2pool_status_view;
+pub mod p2pool_websocket;
 pub mod settings_view;
 pub mod shares_market_view;
 pub mod status_bar;