introduce ZMQ listener thread and arg

Author: RCasatta

Cargo.lock

@@ -56,6 +56,7 @@ tokio = { version = "1", features = ["sync", "macros"] }
 
 # optional dependencies for electrum-discovery
 electrum-client = { version = "0.8", optional = true }
+zmq = "0.10.0"
 
 
 [dev-dependencies]

Cargo.toml

@@ -15,7 +15,7 @@ use electrs::{
     electrum::RPC as ElectrumRPC,
     errors::*,
     metrics::Metrics,
-    new_index::{precache, ChainQuery, FetchFrom, Indexer, Mempool, Query, Store},
+    new_index::{precache, zmq, ChainQuery, FetchFrom, Indexer, Mempool, Query, Store},
     rest,
     signal::Waiter,
 };
@@ -45,6 +45,10 @@ fn run_server(config: Arc<Config>) -> Result<()> {
     let metrics = Metrics::new(config.monitoring_addr);
     metrics.start();
 
+    if let Some(zmq_addr) = config.zmq_addr.as_ref() {
+        zmq::start(&format!("tcp://{zmq_addr}"), None);
+    }
+
     let daemon = Arc::new(Daemon::new(
         &config.daemon_dir,
         &config.blocks_dir,
@@ -85,9 +89,12 @@ fn run_server(config: Arc<Config>) -> Result<()> {
         match Mempool::update(&mempool, &daemon) {
             Ok(_) => break,
             Err(e) => {
-                warn!("Error performing initial mempool update, trying again in 5 seconds: {}", e.display_chain());
+                warn!(
+                    "Error performing initial mempool update, trying again in 5 seconds: {}",
+                    e.display_chain()
+                );
                 signal.wait(Duration::from_secs(5), false)?;
-            },
+            }
         }
     }
 
@@ -117,7 +124,6 @@ fn run_server(config: Arc<Config>) -> Result<()> {
     ));
 
     loop {
-
         main_loop_count.inc();
 
         if let Err(err) = signal.wait(Duration::from_secs(5), true) {
@@ -137,7 +143,10 @@ fn run_server(config: Arc<Config>) -> Result<()> {
         // Update mempool
         if let Err(e) = Mempool::update(&mempool, &daemon) {
             // Log the error if the result is an Err
-            warn!("Error updating mempool, skipping mempool update: {}", e.display_chain());
+            warn!(
+                "Error updating mempool, skipping mempool update: {}",
+                e.display_chain()
+            );
         }
 
         // Update subscribed clients