Multi Thread sync, ported from 5d2b85c03a

lib/src/grpcconnector.rs

@@ -2,12 +2,15 @@ use log::{error};
 use std::sync::Arc;
 use zcash_primitives::transaction::{TxId};
 
-use crate::grpc_client::{ChainSpec, BlockId, BlockRange, RawTransaction, 
+use crate::grpc_client::{ChainSpec, BlockId, BlockRange, RawTransaction, CompactBlock,
                          TransparentAddressBlockFilter, TxFilter, Empty, LightdInfo, Coinsupply};
 use tonic::transport::{Channel, ClientTlsConfig};
 use tokio_rustls::{rustls::ClientConfig};
 use tonic::{Request};
 
+use threadpool::ThreadPool;
+use std::sync::mpsc::channel;
+
 use crate::PubCertificate;
 use crate::grpc_client::compact_tx_streamer_client::CompactTxStreamerClient;
 
@@ -95,7 +98,7 @@ pub fn get_coinsupply(uri: http::Uri, no_cert: bool) -> Result<Coinsupply, Strin
    // tokio::runtime::current_thread::Runtime::new().unwrap().block_on(runner)
 }
 
-async fn get_block_range<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, c: F) 
+async fn get_block_range<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, pool: ThreadPool, c: F) 
     -> Result<(), Box<dyn std::error::Error>> 
 where F : Fn(&[u8], u64) {
     let mut client = get_client(uri, no_cert).await?;
@@ -105,20 +108,40 @@ where F : Fn(&[u8], u64) {
 
     let request = Request::new(BlockRange{ start: Some(bs), end: Some(be) });
 
+     // Channel where the blocks are sent. A None signifies end of all blocks
+     let (tx, rx) = channel::<Option<CompactBlock>>();
+
+     // Channel that the processor signals it is done, so the method can return
+     let (ftx, frx) = channel();
+ 
+     // The processor runs on a different thread, so that the network calls don't
+     // block on this
+     pool.execute(move || {
+         while let Some(block) = rx.recv().unwrap() {
+             use prost::Message;
+             let mut encoded_buf = vec![];
+ 
+             block.encode(&mut encoded_buf).unwrap();
+             c(&encoded_buf, block.height);
+         }
+ 
+         ftx.send(Ok(())).unwrap();
+     });
+
     let mut response = client.get_block_range(request).await?.into_inner();
     //println!("{:?}", response);
     while let Some(block) = response.message().await? {
-        use prost::Message;
-        let mut encoded_buf = vec![];
-
-        block.encode(&mut encoded_buf).unwrap();
-        c(&encoded_buf, block.height);
+        tx.send(Some(block)).unwrap();
     }
+    tx.send(None).unwrap();
+
+    // Wait for the processor to exit
+    frx.iter().take(1).collect::<Result<Vec<()>, String>>()?;
 
     Ok(())
 }
 
-pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, c: F) -> Result<(), String>
+pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_height: u64, end_height: u64, no_cert: bool, pool: ThreadPool, c: F) -> Result<(), String>
     where F : Fn(&[u8], u64) {
     
     let mut rt = match tokio::runtime::Runtime::new() {
@@ -131,7 +154,7 @@ pub fn fetch_blocks<F : 'static + std::marker::Send>(uri: &http::Uri, start_heig
         }
     };
 
-    match rt.block_on(get_block_range(uri, start_height, end_height, no_cert, c)) {
+    match rt.block_on(get_block_range(uri, start_height, end_height, no_cert, pool, c)) {
         Ok(o) => Ok(o),
         Err(e) => {
             let e = format!("Error fetching blocks {:?}", e);
@@ -202,26 +225,26 @@ async fn get_transaction(uri: &http::Uri, txid: TxId, no_cert: bool)
     Ok(response.into_inner())
 }
 
-pub fn fetch_full_tx<F : 'static + std::marker::Send>(uri: &http::Uri, txid: TxId, no_cert: bool, c: F)
-        where F : Fn(&[u8]) {
+pub fn fetch_full_tx(uri: &http::Uri, txid: TxId, no_cert: bool) -> Result<Vec<u8>, String> {
     let mut rt = match tokio::runtime::Runtime::new() {
         Ok(r) => r,
         Err(e) => {
-            error!("Error creating runtime {}", e.to_string());
-            eprintln!("{}", e);
-            return;
+            let errstr = format!("Error creating runtime {}", e.to_string());
+            error!("{}", errstr);
+            eprintln!("{}", errstr);
+            return Err(errstr);
         }
     };
 
     match rt.block_on(get_transaction(uri, txid, no_cert)) {
-        Ok(rawtx) => c(&rawtx.data),
+        Ok(rawtx) => Ok(rawtx.data.to_vec()),
         Err(e) => {
-            error!("Error in get_transaction runtime {}", e.to_string());
-            eprintln!("{}", e);
+            let errstr = format!("Error in get_transaction runtime {}", e.to_string());
+            error!("{}", errstr);
+            eprintln!("{}", errstr);
+            Err(errstr)
         }
-    }
-
-    
+    }    
 }
 
 // send_transaction GRPC call
@@ -262,22 +285,19 @@ async fn get_latest_block(uri: &http::Uri, no_cert: bool) -> Result<BlockId, Box
     Ok(response.into_inner())
 }
 
-pub fn fetch_latest_block<F : 'static + std::marker::Send>(uri: &http::Uri, no_cert: bool, mut c : F) 
-    where F : FnMut(BlockId) {
+pub fn fetch_latest_block(uri: &http::Uri, no_cert: bool) -> Result<BlockId, String> {
     let mut rt = match tokio::runtime::Runtime::new() {
         Ok(r) => r,
         Err(e) => {
-            error!("Error creating runtime {}", e.to_string());
-            eprintln!("{}", e);
-            return;
+            let errstr = format!("Error creating runtime {}", e.to_string());
+            eprintln!("{}", errstr);
+            return Err(errstr);
         }
     };
 
-    match rt.block_on(get_latest_block(uri, no_cert)) {
-        Ok(b) => c(b),
-        Err(e) => {
-            error!("Error getting latest block {}", e.to_string());
-            eprintln!("{}", e);
-        }
-    };
+    rt.block_on(get_latest_block(uri, no_cert)).map_err(|e| {
+        let errstr = format!("Error getting latest block {}", e.to_string());
+        eprintln!("{}", errstr);
+        errstr
+    })
 }