From d95ad091f66a4e38b0b5ba74c6691964957ab60d Mon Sep 17 00:00:00 2001 From: Odysseas Gabrielides Date: Wed, 13 Nov 2024 17:53:11 +0200 Subject: [PATCH 1/2] disable zmq for windows --- Cargo.toml | 6 ++-- src/app.rs | 47 ++++++++++++++++++----------- src/components/core_zmq_listener.rs | 3 ++ src/ui/components/top_panel.rs | 5 ++- 4 files changed, 41 insertions(+), 20 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index e6f64dfd8..d87a6664d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,9 +48,11 @@ 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" \ No newline at end of file diff --git a/src/app.rs b/src/app.rs index 396952bde..7cb984e84 100644 --- a/src/app.rs +++ b/src/app.rs @@ -49,8 +49,8 @@ pub struct AppState { pub chosen_network: Network, pub mainnet_app_context: Arc, pub testnet_app_context: Option>, - pub mainnet_core_zmq_listener: CoreZMQListener, - pub testnet_core_zmq_listener: CoreZMQListener, + pub mainnet_core_zmq_listener: Option, + pub testnet_core_zmq_listener: Option, pub core_message_receiver: mpsc::Receiver<(ZMQMessage, Network)>, pub task_result_sender: tokiompsc::Sender, // Channel sender for sending task results pub task_result_receiver: tokiompsc::Receiver, // Channel receiver for receiving task results @@ -189,27 +189,40 @@ 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"); + #[cfg(target_os = "windows")] + let mainnet_core_zmq_listener: Option = None; + + #[cfg(not(target_os = "windows"))] + let mainnet_core_zmq_listener: Option = { + // Pass the sender to the listener when creating it + let 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"); + Some(zmq_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"); + #[cfg(target_os = "windows")] + let testnet_core_zmq_listener: Option = None; + + #[cfg(not(target_os = "windows"))] + let testnet_core_zmq_listener: Option = { + let 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"); + Some(zmq_listener) // or however you construct `Data` + }; Self { main_screens: [ diff --git a/src/components/core_zmq_listener.rs b/src/components/core_zmq_listener.rs index 2fd41fd3c..271467ab5 100644 --- a/src/components/core_zmq_listener.rs +++ b/src/components/core_zmq_listener.rs @@ -11,6 +11,8 @@ use std::sync::{ }; use std::thread; use std::time::Duration; + +#[cfg(not(target_os = "windows"))] use zmq::Context; pub struct CoreZMQListener { @@ -34,6 +36,7 @@ pub const IS_LOCK_SIG_MSG: &[u8; 12] = b"rawtxlocksig"; pub const CHAIN_LOCKED_BLOCK_MSG: &[u8; 12] = b"rawchainlock"; impl CoreZMQListener { + #[cfg(not(target_os = "windows"))] pub fn spawn_listener( network: Network, endpoint: &str, 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); From 99c41e9fd2953ac19ec74811686caf6a45b80417 Mon Sep 17 00:00:00 2001 From: Odysseas Gabrielides Date: Wed, 13 Nov 2024 18:52:38 +0200 Subject: [PATCH 2/2] switch lib in windows --- Cargo.toml | 5 +- src/app.rs | 44 +++----- src/components/core_zmq_listener.rs | 153 +++++++++++++++++++++++++++- 3 files changed, 170 insertions(+), 32 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index d87a6664d..8d6fae730 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -55,4 +55,7 @@ aes-gcm = "0.10" # For AES-256-GCM encryption crossbeam-channel = "0.5.13" [target.'cfg(not(target_os = "windows"))'.dependencies] -zmq = "0.10" \ No newline at end of file +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 7cb984e84..87d3490ab 100644 --- a/src/app.rs +++ b/src/app.rs @@ -49,8 +49,8 @@ pub struct AppState { pub chosen_network: Network, pub mainnet_app_context: Arc, pub testnet_app_context: Option>, - pub mainnet_core_zmq_listener: Option, - pub testnet_core_zmq_listener: Option, + pub mainnet_core_zmq_listener: CoreZMQListener, + pub testnet_core_zmq_listener: CoreZMQListener, pub core_message_receiver: mpsc::Receiver<(ZMQMessage, Network)>, pub task_result_sender: tokiompsc::Sender, // Channel sender for sending task results pub task_result_receiver: tokiompsc::Receiver, // Channel receiver for receiving task results @@ -189,40 +189,24 @@ impl AppState { // Create a channel for communication with the InstantSendListener let (core_message_sender, core_message_receiver) = mpsc::channel(); - #[cfg(target_os = "windows")] - let mainnet_core_zmq_listener: Option = None; - - #[cfg(not(target_os = "windows"))] - let mainnet_core_zmq_listener: Option = { - // Pass the sender to the listener when creating it - let 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"); - Some(zmq_listener) - }; + 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"); let tx_zmq_status_option = match testnet_app_context { Some(ref context) => Some(context.sx_zmq_status.clone()), None => None, }; - #[cfg(target_os = "windows")] - let testnet_core_zmq_listener: Option = None; - - #[cfg(not(target_os = "windows"))] - let testnet_core_zmq_listener: Option = { - let 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"); - Some(zmq_listener) // or however you construct `Data` - }; + 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"); Self { main_screens: [ diff --git a/src/components/core_zmq_listener.rs b/src/components/core_zmq_listener.rs index 271467ab5..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::{ @@ -14,6 +13,15 @@ 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, @@ -32,9 +40,16 @@ 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( @@ -259,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);