fix order of batch calls by including the response id in the ChannelMessage

fixes bitcoindevkit#160

Author: ValuedMammal

Cargo.toml

@@ -20,6 +20,7 @@ path = "src/lib.rs"
 
 [dependencies]
 log = "^0.4"
+pretty_env_logger = "0.5"
 bitcoin = { version = "0.32", features = ["serde"] }
 serde = { version = "^1.0", features = ["derive"] }
 serde_json = { version = "^1.0" }

examples/plaintext.rs

@@ -1,9 +1,59 @@
 extern crate electrum_client;
+extern crate log;
+extern crate pretty_env_logger;
 
+use bitcoin::block::Header;
 use electrum_client::{Client, ElectrumApi};
 
 fn main() {
-    let client = Client::new("tcp://electrum.blockstream.info:50001").unwrap();
-    let res = client.server_features();
-    println!("{:#?}", res);
+    pretty_env_logger::init_timed();
+
+    for _ in 0..10 {
+        // let client = Client::new("electrum.blockstream.info:50001").unwrap();
+        let client = Client::new("exs.dyshek.org:50001").unwrap();
+        // let client = Client::new("tcp://bitcoin.aranguren.org:50001").unwrap();
+
+        let heights: Vec<u32> = vec![1, 4, 8, 12, 222, 6666];
+        let headers: Vec<Header> = client.batch_block_header(heights).unwrap();
+
+        let mut err_counter = 0;
+        if headers.get(0).unwrap().time != 1231469665 {
+            // time of block 1
+            err_counter += 1;
+            log::error!("error 0");
+        }
+        if headers.get(1).unwrap().time != 1231470988 {
+            // time of block 4
+            err_counter += 1;
+            log::error!("error 1");
+        }
+        if headers.get(2).unwrap().time != 1231472743 {
+            // time of block 8
+            err_counter += 1;
+            log::error!("error 2");
+        }
+        if headers.get(3).unwrap().time != 1231474888 {
+            // time of block 12
+            err_counter += 1;
+            log::error!("error 3");
+        }
+        if headers.get(4).unwrap().time != 1231770653 {
+            // time of block 222
+            err_counter += 1;
+            log::error!("error 4");
+        }
+        if headers.get(5).unwrap().time != 1236456633 {
+            // time of block 6666
+            err_counter += 1;
+            log::error!("error 5");
+        }
+        if err_counter > 0 {
+            log::error!(
+                "Result: {} headers were not at expected index in result vector",
+                err_counter
+            );
+            break;
+        }
+        println!("==========================================")
+    }
 }

src/raw_client.rs

@@ -511,7 +511,7 @@ impl RawClient<ElectrumProxyStream> {
 
 #[derive(Debug)]
 enum ChannelMessage {
-    Response(serde_json::Value),
+    Response((usize, serde_json::Value)),
     WakeUp,
     Error(Arc<std::io::Error>),
 }
@@ -605,7 +605,8 @@ impl<S: Read + Write> RawClient<S> {
                             trace!("Reader thread received response for {}", resp_id);
 
                             if let Some(sender) = self.waiting_map.lock()?.remove(&resp_id) {
-                                sender.send(ChannelMessage::Response(resp))?;
+                                log::debug!("Sending resp_id {resp_id}");
+                                sender.send(ChannelMessage::Response((resp_id, resp)))?;
                             } else {
                                 warn!("Missing listener for {}", resp_id);
                             }
@@ -657,12 +658,12 @@ impl<S: Read + Write> RawClient<S> {
         self.increment_calls();
 
         let mut resp = match self.recv(&receiver, req.id) {
-            Ok(resp) => resp,
-            e @ Err(_) => {
+            Ok((_, resp)) => resp,
+            Err(e) => {
                 // In case of error our sender could still be left in the map, depending on where
                 // the error happened. Just in case, try to remove it here
                 self.waiting_map.lock()?.remove(&req.id);
-                return e;
+                return Err(e);
             }
         };
         Ok(resp["result"].take())
@@ -672,16 +673,16 @@ impl<S: Read + Write> RawClient<S> {
         &self,
         receiver: &Receiver<ChannelMessage>,
         req_id: usize,
-    ) -> Result<serde_json::Value, Error> {
+    ) -> Result<(usize, serde_json::Value), Error> {
         loop {
             // Try to take the lock on the reader. If we manage to do so, we'll become the reader
             // thread until we get our reponse
             match self._reader_thread(Some(req_id)) {
-                Ok(response) => break Ok(response),
+                Ok(response) => break Ok((req_id, response)),
                 Err(Error::CouldntLockReader) => {
                     match receiver.recv()? {
-                        // Received our response, returning it
-                        ChannelMessage::Response(received) => break Ok(received),
+                        // Received a response, returning it
+                        ChannelMessage::Response((resp_id, resp)) => break Ok((resp_id, resp)),
                         ChannelMessage::WakeUp => {
                             // We have been woken up, this means that we should try becoming the
                             // reader thread ourselves
@@ -696,7 +697,7 @@ impl<S: Read + Write> RawClient<S> {
                         }
                     }
                 }
-                e @ Err(_) => break e,
+                Err(e) => break Err(e),
             }
         }
     }
@@ -798,7 +799,10 @@ impl<T: Read + Write> ElectrumApi for RawClient<T> {
 
         for req_id in missing_responses.iter() {
             match self.recv(&receiver, *req_id) {
-                Ok(mut resp) => answers.insert(req_id, resp["result"].take()),
+                Ok((resp_id, mut resp)) => {
+                    log::debug!("Received resp_id {resp_id}");
+                    answers.insert(resp_id, resp["result"].take());
+                }
                 Err(e) => {
                     // In case of error our sender could still be left in the map, depending on where
                     // the error happened. Just in case, try to remove it here