Merge branch 'zmq' into new-index

Cargo.toml

@@ -57,6 +57,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]

src/bin/electrs.rs

@@ -4,6 +4,7 @@ extern crate log;
 
 extern crate electrs;
 
+use crossbeam_channel::{self as channel};
 use error_chain::ChainedError;
 use std::process;
 use std::sync::{Arc, RwLock};
@@ -15,7 +16,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,
 };
@@ -41,10 +42,15 @@ fn fetch_from(config: &Config, store: &Store) -> FetchFrom {
 }
 
 fn run_server(config: Arc<Config>) -> Result<()> {
-    let signal = Waiter::start();
+    let (block_hash_notify, block_hash_receive) = channel::bounded(1);
+    let signal = Waiter::start(block_hash_receive);
     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}"), block_hash_notify);
+    }
+
     let daemon = Arc::new(Daemon::new(
         &config.daemon_dir,
         &config.blocks_dir,

src/bin/tx-fingerprint-stats.rs

@@ -21,7 +21,7 @@ fn main() {
         util::has_prevout,
     };
 
-    let signal = Waiter::start();
+    let signal = Waiter::start(crossbeam_channel::never());
     let config = Config::from_args();
     let store = Arc::new(Store::open(&config.db_path.join("newindex"), &config));
 
@@ -83,13 +83,15 @@ fn main() {
 
         //info!("{:?},{:?}", txid, blockid);
 
-        let prevouts = chain.lookup_txos(
-            tx.input
-                .iter()
-                .filter(|txin| has_prevout(txin))
-                .map(|txin| txin.previous_output)
-                .collect(),
-        ).unwrap();
+        let prevouts = chain
+            .lookup_txos(
+                tx.input
+                    .iter()
+                    .filter(|txin| has_prevout(txin))
+                    .map(|txin| txin.previous_output)
+                    .collect(),
+            )
+            .unwrap();
 
         let total_out: u64 = tx.output.iter().map(|out| out.value.to_sat()).sum();
         let small_out = tx