diff --git a/Cargo.toml b/Cargo.toml index 2978cb12b..dac5cdc08 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,4 +52,5 @@ zmq = "0.10" zeroize = "1.8.1" zxcvbn = "3.1.0" argon2 = "0.5" # For Argon2 key derivation -aes-gcm = "0.10"# For AES-256-GCM encryption \ No newline at end of file +aes-gcm = "0.10" # For AES-256-GCM encryption +crossbeam-channel = "0.5.13" \ No newline at end of file diff --git a/src/app.rs b/src/app.rs index 651ea78fe..bd39af25f 100644 --- a/src/app.rs +++ b/src/app.rs @@ -193,13 +193,20 @@ impl AppState { Network::Dash, "tcp://127.0.0.1:23708", core_message_sender.clone(), // Clone the sender for each listener + Some(mainnet_app_context.sx_zmq_status.clone()), ) .expect("Failed to create mainnet InstantSend listener"); + let tx_zmq_status_option = match testnet_app_context { + Some(ref context) => Some(context.sx_zmq_status.clone()), + None => None, + }; + let testnet_core_zmq_listener = CoreZMQListener::spawn_listener( Network::Testnet, "tcp://127.0.0.1:23709", core_message_sender, // Use the original sender or create a new one if needed + tx_zmq_status_option, ) .expect("Failed to create testnet InstantSend listener"); @@ -364,6 +371,11 @@ impl AppState { impl App for AppState { fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) { + if let Ok(event) = self.current_app_context().rx_zmq_status.try_recv() { + if let Ok(mut status) = self.current_app_context().zmq_connection_status.lock() { + *status = event; + } + } // Poll the receiver for any new task results while let Ok(task_result) = self.task_result_receiver.try_recv() { // Handle the result on the main thread diff --git a/src/components/core_zmq_listener.rs b/src/components/core_zmq_listener.rs index c264d568e..01807fad0 100644 --- a/src/components/core_zmq_listener.rs +++ b/src/components/core_zmq_listener.rs @@ -1,3 +1,4 @@ +use crossbeam_channel::Sender; use dash_sdk::dpp::dashcore::consensus::Decodable; use dash_sdk::dpp::dashcore::{Block, InstantLock, Network, Transaction}; use dash_sdk::dpp::prelude::CoreBlockHeight; @@ -23,6 +24,12 @@ pub enum ZMQMessage { ChainLockedLockedTransaction(Transaction, CoreBlockHeight), } +#[derive(Debug)] +pub enum ZMQConnectionEvent { + Connected, + Disconnected, +} + pub const IS_LOCK_SIG_MSG: &[u8; 12] = b"rawtxlocksig"; pub const CHAIN_LOCKED_BLOCK_MSG: &[u8; 12] = b"rawchainlock"; @@ -31,6 +38,7 @@ impl CoreZMQListener { network: Network, endpoint: &str, sender: mpsc::Sender<(ZMQMessage, Network)>, + tx_zmq_status: Option>, ) -> Result> { let should_stop = Arc::new(AtomicBool::new(false)); let endpoint = endpoint.to_string(); @@ -42,6 +50,27 @@ impl CoreZMQListener { let context = Context::new(); let socket = context.socket(zmq::SUB).expect("Failed to create socket"); + // Set heartbeat options + socket + .set_heartbeat_ivl(5000) + .expect("Failed to set heartbeat interval"); // Send a heartbeat every 5000 ms + socket + .set_heartbeat_timeout(10000) + .expect("Failed to set heartbeat timeout"); // Timeout after 10000 ms without response + + let monitor_addr = "inproc://socket-monitor"; + socket + .monitor(monitor_addr, zmq::SocketEvent::ALL as i32) + .expect("Failed to monitor socket"); + + // Create the PAIR socket for monitoring + let monitor_socket = context + .socket(zmq::PAIR) + .expect("Failed to create monitor socket"); + monitor_socket + .connect(monitor_addr) + .expect("Failed to connect monitor socket"); + // Connect to the zmqpubhashtxlock endpoint. socket.connect(&endpoint).expect("Failed to connect"); @@ -55,110 +84,163 @@ impl CoreZMQListener { .set_subscribe(CHAIN_LOCKED_BLOCK_MSG) .expect("Failed to subscribe to rawchainlock"); - println!("Connected to ZMQ at {}", endpoint); + println!("Subscribed to ZMQ at {}", endpoint); + + let mut items = [ + socket.as_poll_item(zmq::POLLIN), + monitor_socket.as_poll_item(zmq::POLLIN), + ]; while !should_stop_clone.load(Ordering::SeqCst) { - // Receive the topic part of the message - let mut topic_message = zmq::Message::new(); - - // Use non-blocking receive with DONTWAIT. - match socket.recv(&mut topic_message, zmq::DONTWAIT) { - Ok(_) => { - let topic = topic_message.as_str().unwrap_or(""); - let has_more = socket.get_rcvmore().unwrap_or(false); - - if has_more { - // Receive the data part of the message - let mut data_message = zmq::Message::new(); - if let Err(e) = socket.recv(&mut data_message, 0) { - eprintln!("Error receiving data part: {}", e); - continue; - } + zmq::poll(&mut items, -1).expect("Failed to poll sockets"); - let data_bytes = data_message.as_bytes(); + if items[0].is_readable() { + // Handle messages from the SUB socket + // Receive the topic part of the message + let mut topic_message = zmq::Message::new(); - match topic { - "rawchainlock" => { - // println!("Received raw chain locked block:"); - // println!("Data (hex): {}", hex::encode(data_bytes)); + // Use non-blocking receive with DONTWAIT. + match socket.recv(&mut topic_message, zmq::DONTWAIT) { + Ok(_) => { + let topic = topic_message.as_str().unwrap_or(""); + let has_more = socket.get_rcvmore().unwrap_or(false); - // Create a cursor over the data_bytes - let mut cursor = Cursor::new(data_bytes); + if has_more { + // Receive the data part of the message + let mut data_message = zmq::Message::new(); + if let Err(e) = socket.recv(&mut data_message, 0) { + eprintln!("Error receiving data part: {}", e); + continue; + } - // Deserialize the LLMQChainLock - match Block::consensus_decode(&mut cursor) { - Ok(block) => { - // Send the ChainLock and Network back to the main thread - if let Err(e) = sender_clone.send(( - ZMQMessage::ChainLockedBlock(block), - network, - )) { + let data_bytes = data_message.as_bytes(); + + match topic { + "rawchainlock" => { + println!("Received raw chain locked block:"); + println!("Data (hex): {}", hex::encode(data_bytes)); + + // Create a cursor over the data_bytes + let mut cursor = Cursor::new(data_bytes); + + // Deserialize the LLMQChainLock + match Block::consensus_decode(&mut cursor) { + Ok(block) => { + // Send the ChainLock and Network back to the main thread + if let Err(e) = sender_clone.send(( + ZMQMessage::ChainLockedBlock(block), + network, + )) { + eprintln!( + "Error sending data to main thread: {}", + e + ); + } + } + Err(e) => { eprintln!( - "Error sending data to main thread: {}", + "Error deserializing chain locked block: {}", e ); } } - Err(e) => { - eprintln!( - "Error deserializing chain locked block: {}", - e - ); - } } - } - "rawtxlocksig" => { - // println!("Received rawtxlocksig for InstantSend:"); - // println!("Data (hex): {}", hex::encode(data_bytes)); - - // Create a cursor over the data_bytes - let mut cursor = Cursor::new(data_bytes); - - // Deserialize the transaction - match Transaction::consensus_decode(&mut cursor) { - Ok(tx) => { - // Deserialize the InstantLock from the remaining bytes - match InstantLock::consensus_decode(&mut cursor) { - Ok(islock) => { - // Send the Transaction, InstantLock, and Network back to the main thread - if let Err(e) = sender_clone.send(( - ZMQMessage::ISLockedTransaction(tx, islock), - network, - )) { + "rawtxlocksig" => { + println!("Received rawtxlocksig for InstantSend:"); + println!("Data (hex): {}", hex::encode(data_bytes)); + + // Create a cursor over the data_bytes + let mut cursor = Cursor::new(data_bytes); + + // Deserialize the transaction + match Transaction::consensus_decode(&mut cursor) { + Ok(tx) => { + // Deserialize the InstantLock from the remaining bytes + match InstantLock::consensus_decode(&mut cursor) { + Ok(islock) => { + // Send the Transaction, InstantLock, and Network back to the main thread + if let Err(e) = sender_clone.send(( + ZMQMessage::ISLockedTransaction( + tx, islock, + ), + network, + )) { + eprintln!( + "Error sending data to main thread: {}", + e + ); + } + } + Err(e) => { eprintln!( - "Error sending data to main thread: {}", + "Error deserializing InstantLock: {}", e ); } } - Err(e) => { - eprintln!( - "Error deserializing InstantLock: {}", - e - ); - } } - } - Err(e) => { - eprintln!("Error deserializing transaction: {}", e); + Err(e) => { + eprintln!("Error deserializing transaction: {}", e); + } } } + _ => { + println!("Received unknown topic: {}", topic); + } } - _ => { - println!("Received unknown topic: {}", topic); - } + } + } + Err(e) => { + if e == zmq::Error::EAGAIN { + // No message received, sleep briefly. + thread::sleep(Duration::from_millis(100)); + continue; + } else { + eprintln!("Error receiving message: {}", e); + break; } } } - Err(e) => { - if e == zmq::Error::EAGAIN { - // No message received, sleep briefly. - thread::sleep(Duration::from_millis(100)); - continue; - } else { - eprintln!("Error receiving message: {}", e); - break; + } + + if items[1].is_readable() { + let mut event_msg = zmq::Message::new(); + monitor_socket + .recv(&mut event_msg, 0) + .expect("Failed to receive event message"); + + let mut addr_msg = zmq::Message::new(); + monitor_socket + .recv(&mut addr_msg, 0) + .expect("Failed to receive address message"); + + let data = event_msg.as_ref(); + if data.len() >= 6 { + let event_number = u16::from_le_bytes([data[0], data[1]]); + let endpoint = addr_msg.as_str().unwrap_or(""); + + match zmq::SocketEvent::from_raw(event_number) { + zmq::SocketEvent::CONNECTED => { + if let Some(ref tx) = tx_zmq_status { + println!("ODY Socket connected to {}", endpoint); + tx.send(ZMQConnectionEvent::Connected) + .expect("Failed to send connected event"); + } + // Connection is successful + } + zmq::SocketEvent::DISCONNECTED => { + if let Some(ref tx) = tx_zmq_status { + println!("ODY Socket disconnected from {}", endpoint); + tx.send(ZMQConnectionEvent::Disconnected) + .expect("Failed to send connected event"); + } + // Connection is lost + } + // Handle other events as needed + _ => {} } + } else { + println!("Invalid event message received"); } } } diff --git a/src/context.rs b/src/context.rs index c93a81cc0..e5d40d37a 100644 --- a/src/context.rs +++ b/src/context.rs @@ -1,3 +1,4 @@ +use crate::components::core_zmq_listener::ZMQConnectionEvent; use crate::config::{Config, NetworkConfig}; use crate::context_provider::Provider; use crate::database::Database; @@ -8,6 +9,7 @@ use crate::model::qualified_identity::{DPNSNameInfo, QualifiedIdentity}; use crate::model::wallet::Wallet; use crate::sdk_wrapper::initialize_sdk; use crate::ui::RootScreenType; +use crossbeam_channel::{Receiver, Sender}; use dash_sdk::dashcore_rpc::dashcore::{InstantLock, Transaction}; use dash_sdk::dashcore_rpc::{Auth, Client}; use dash_sdk::dpp::dashcore::hashes::Hash; @@ -35,6 +37,9 @@ pub struct AppContext { pub(crate) db: Arc, pub(crate) sdk: Sdk, pub(crate) config: NetworkConfig, + pub(crate) rx_zmq_status: Receiver, + pub(crate) sx_zmq_status: Sender, + pub(crate) zmq_connection_status: Mutex, pub(crate) dpns_contract: Arc, pub(crate) withdraws_contract: Arc, pub(crate) core_client: Client, @@ -60,6 +65,7 @@ impl AppContext { }; let network_config = config.config_for_network(network).clone()?; + let (sx_zmq_status, rx_zmq_status) = crossbeam_channel::unbounded(); // we create provider, but we need to set app context to it later, as we have a circular dependency let provider = @@ -102,6 +108,8 @@ impl AppContext { db, sdk, config: network_config, + sx_zmq_status, + rx_zmq_status, dpns_contract: Arc::new(dpns_contract), withdraws_contract: Arc::new(withdrawal_contract), core_client, @@ -110,6 +118,7 @@ impl AppContext { password_info, transactions_waiting_for_finality: Mutex::new(BTreeMap::new()), platform_version: PlatformVersion::latest(), + zmq_connection_status: Mutex::new(ZMQConnectionEvent::Disconnected), }; let app_context = Arc::new(app_context); diff --git a/src/ui/components/top_panel.rs b/src/ui/components/top_panel.rs index 37dc22ecf..d04bd2c06 100644 --- a/src/ui/components/top_panel.rs +++ b/src/ui/components/top_panel.rs @@ -1,4 +1,5 @@ use crate::app::{AppAction, DesiredAppAction}; +use crate::components::core_zmq_listener::ZMQConnectionEvent; use crate::context::AppContext; use dash_sdk::dashcore_rpc::dashcore::Network; use egui::{Align, Color32, Context, Frame, Layout, Margin, RichText, Stroke, TopBottomPanel, Ui}; @@ -67,6 +68,18 @@ pub fn add_top_panel( // Right-aligned content with buttons ui.with_layout(Layout::right_to_left(Align::Center), |ui| { + let connection_status = { + if let Ok(status) = app_context.zmq_connection_status.lock() { + match *status { + ZMQConnectionEvent::Connected => "zmq connected", + ZMQConnectionEvent::Disconnected => "zmq disconnected", + } + } else { + "" + } + }; + ui.label(connection_status); + for (text, right_button_action) in right_buttons.into_iter().rev() { ui.add_space(8.0);