diff --git a/Cargo.toml b/Cargo.toml index e6f64dfd8..8d6fae730 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,9 +48,14 @@ image = { version = "0.25.2", default-features = false, features = ["png"] } bitflags = "2.6.0" libsqlite3-sys = { version = "0.30.1", features = ["bundled"] } rust-embed = "8.5.0" -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 -crossbeam-channel = "0.5.13" \ No newline at end of file +crossbeam-channel = "0.5.13" + +[target.'cfg(not(target_os = "windows"))'.dependencies] +zmq = "0.10" + +[target.'cfg(target_os = "windows")'.dependencies] +zeromq = "0.4.1" \ No newline at end of file diff --git a/src/app.rs b/src/app.rs index 396952bde..87d3490ab 100644 --- a/src/app.rs +++ b/src/app.rs @@ -189,14 +189,12 @@ impl AppState { // Create a channel for communication with the InstantSendListener let (core_message_sender, core_message_receiver) = mpsc::channel(); - // Pass the sender to the listener when creating it let mainnet_core_zmq_listener = CoreZMQListener::spawn_listener( 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"); + ).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()), @@ -208,8 +206,7 @@ impl AppState { "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"); + ).expect("Failed to create testnet InstantSend listener"); Self { main_screens: [ diff --git a/src/components/core_zmq_listener.rs b/src/components/core_zmq_listener.rs index 2fd41fd3c..46ca5e2e7 100644 --- a/src/components/core_zmq_listener.rs +++ b/src/components/core_zmq_listener.rs @@ -2,7 +2,6 @@ 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; -use image::EncodableLayout; use std::error::Error; use std::io::Cursor; use std::sync::{ @@ -11,7 +10,18 @@ use std::sync::{ }; use std::thread; use std::time::Duration; + +#[cfg(not(target_os = "windows"))] use zmq::Context; +#[cfg(not(target_os = "windows"))] +use image::EncodableLayout; + +#[cfg(target_os = "windows")] +use futures::StreamExt; +#[cfg(target_os = "windows")] +use tokio::runtime::Runtime; +#[cfg(target_os = "windows")] +use zeromq::{Socket, SocketRecv, SubSocket}; pub struct CoreZMQListener { should_stop: Arc, @@ -30,10 +40,18 @@ pub enum ZMQConnectionEvent { Disconnected, } +#[cfg(not(target_os = "windows"))] pub const IS_LOCK_SIG_MSG: &[u8; 12] = b"rawtxlocksig"; +#[cfg(not(target_os = "windows"))] pub const CHAIN_LOCKED_BLOCK_MSG: &[u8; 12] = b"rawchainlock"; +#[cfg(target_os = "windows")] +pub const IS_LOCK_SIG_MSG: &str = "rawtxlocksig"; +#[cfg(target_os = "windows")] +pub const CHAIN_LOCKED_BLOCK_MSG: &str = "rawchainlock"; + impl CoreZMQListener { + #[cfg(not(target_os = "windows"))] pub fn spawn_listener( network: Network, endpoint: &str, @@ -256,6 +274,142 @@ impl CoreZMQListener { }) } + #[cfg(target_os = "windows")] + pub fn spawn_listener( + 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(); + let should_stop_clone = Arc::clone(&should_stop); + let sender_clone = sender.clone(); + + let handle = thread::spawn(move || { + // Create the runtime inside the thread. + let rt = Runtime::new().unwrap(); + rt.block_on(async move { + // Create the socket inside the async context. + let mut socket = SubSocket::new(); + + // Connect to the endpoint + socket + .connect(&endpoint) + .await + .expect("Failed to connect"); + + // Subscribe to the "rawtxlocksig" events. + socket + .subscribe(IS_LOCK_SIG_MSG) + .await + .expect("Failed to subscribe to rawtxlocksig"); + + // Subscribe to the "rawchainlock" events. + socket + .subscribe(CHAIN_LOCKED_BLOCK_MSG) + .await + .expect("Failed to subscribe to rawchainlock"); + + println!("Subscribed to ZMQ at {}", endpoint); + + while !should_stop_clone.load(Ordering::SeqCst) { + // Receive messages + match socket.recv().await { + Ok(msg) => { + // Access frames using msg.get(n) + if let Some(topic_frame) = msg.get(0) { + let topic = String::from_utf8_lossy(topic_frame).to_string(); + + if let Some(data_frame) = msg.get(1) { + let data_bytes = data_frame; + + match topic.as_str() { + "rawchainlock" => { + // Deserialize the Block + let mut cursor = Cursor::new(data_bytes); + match Block::consensus_decode(&mut cursor) { + Ok(block) => { + if let Err(e) = sender_clone.send(( + ZMQMessage::ChainLockedBlock(block), + network, + )) { + eprintln!( + "Error sending data to main thread: {}", + e + ); + } + } + Err(e) => { + eprintln!( + "Error deserializing chain locked block: {}", + e + ); + } + } + } + "rawtxlocksig" => { + // Deserialize the Transaction and InstantLock + let mut cursor = Cursor::new(data_bytes); + match Transaction::consensus_decode(&mut cursor) { + Ok(tx) => { + match InstantLock::consensus_decode(&mut cursor) + { + Ok(islock) => { + if let Err(e) = sender_clone.send(( + ZMQMessage::ISLockedTransaction( + tx, islock, + ), + network, + )) { + eprintln!( + "Error sending data to main thread: {}", + e + ); + } + } + Err(e) => { + eprintln!( + "Error deserializing InstantLock: {}", + e + ); + } + } + } + Err(e) => { + eprintln!( + "Error deserializing transaction: {}", + e + ); + } + } + } + _ => { + println!("Received unknown topic: {}", topic); + } + } + } + } + } + Err(e) => { + eprintln!("Error receiving message: {}", e); + // Sleep briefly before retrying + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + } + + println!("Listener is stopping."); + // The socket will be dropped here + }); + }); + + Ok(CoreZMQListener { + should_stop, + handle: Some(handle), + }) + } + /// Stops the listener by signaling the thread and waiting for it to finish. pub fn stop(&mut self) { self.should_stop.store(true, Ordering::SeqCst); diff --git a/src/ui/components/top_panel.rs b/src/ui/components/top_panel.rs index f9e8d26b6..bf371054c 100644 --- a/src/ui/components/top_panel.rs +++ b/src/ui/components/top_panel.rs @@ -127,7 +127,10 @@ pub fn add_top_panel( .exact_height(50.0) .show(ctx, |ui| { egui::menu::bar(ui, |ui| { - action |= add_connection_indicator(ui, app_context); + #[cfg(not(target_os = "windows"))] + { + action |= add_connection_indicator(ui, app_context); + } // Left-aligned content with location view action |= add_location_view(ui, location);