Skip to content

feat: disable zmq connectivity detection for windows build #53

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
QuantumExplorer merged 2 commits into from
Nov 13, 2024
Merged

feat: disable zmq connectivity detection for windows build #53

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d95ad09
disable zmq for windows
ogabrielides Nov 13, 2024
99c41e9
switch lib in windows
ogabrielides Nov 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
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"
7 changes: 2 additions & 5 deletions src/app.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()),
Expand All @@ -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: [
Expand Down
156 changes: 155 additions & 1 deletion src/components/core_zmq_listener.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand All @@ -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<AtomicBool>,
Expand All @@ -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(
Comment thread
ogabrielides marked this conversation as resolved.
network: Network,
endpoint: &str,
Expand Down Expand Up @@ -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<Sender<ZMQConnectionEvent>>,
) -> Result<Self, Box<dyn Error>> {
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
});
});
Comment on lines +277 to +405

@coderabbitai coderabbitai Bot Nov 13, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Avoid creating a new Tokio runtime inside a thread

In the Windows implementation, a new tokio::runtime::Runtime is created within a spawned thread. Creating multiple runtimes or embedding a runtime inside a thread can lead to unexpected behavior and resource contention. Consider refactoring to use asynchronous functions without spawning an additional thread, or use tokio::spawn to run tasks on the existing runtime.

Apply this refactor to optimize runtime usage:

  • Option 1: If the application already uses Tokio, integrate the listener directly into the existing async context.

  • Option 2: Use tokio::spawn to run the async task without creating a new runtime.

Example using tokio::spawn:

#[cfg(target_os = "windows")]
pub fn spawn_listener(
    network: Network,
    endpoint: &str,
    sender: mpsc::Sender<(ZMQMessage, Network)>,
    tx_zmq_status: Option<Sender<ZMQConnectionEvent>>,
) -> Result<Self, Box<dyn Error>> {
    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();

    tokio::spawn(async move {
        // Async listener code here...
        while !should_stop_clone.load(Ordering::SeqCst) {
            // Receive and handle messages...
        }
    });

    Ok(CoreZMQListener {
        should_stop,
        handle: None, // No thread handle needed
    })
}

This refactor simplifies the code and aligns with best practices for asynchronous Rust applications.


Ok(CoreZMQListener {
should_stop,
handle: Some(handle),
})
}

Comment on lines +277 to +412

@coderabbitai coderabbitai Bot Nov 13, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Reduce code duplication in spawn_listener implementations

The spawn_listener method has separate implementations for Windows and non-Windows platforms, but much of the logic is similar. This duplication can make maintenance more challenging. Consider extracting common functionality into shared helper functions or using conditional compilation only around the specific platform-dependent sections.

Apply this refactor to reduce duplication:

  1. Extract common code: Identify code blocks that are identical or very similar in both implementations and move them into shared helper functions.

  2. Use conditional compilation within functions: Instead of duplicating entire functions, use #[cfg(...)] within the function to handle platform-specific differences.

Example:

pub fn spawn_listener(
    network: Network,
    endpoint: &str,
    sender: mpsc::Sender<(ZMQMessage, Network)>,
    tx_zmq_status: Option<Sender<ZMQConnectionEvent>>,
) -> Result<Self, Box<dyn Error>> {
    // Common setup code here...

    #[cfg(not(target_os = "windows"))]
    {
        // Non-Windows specific code...
    }

    #[cfg(target_os = "windows")]
    {
        // Windows-specific code...
    }

    // Common code to finalize and return...
}

This approach minimizes duplication and simplifies future updates.

/// 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);
Expand Down
5 changes: 4 additions & 1 deletion src/ui/components/top_panel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down