prevent unwrap in get_block_range
This commit is contained in:
@@ -100,49 +100,63 @@ pub fn get_coinsupply(uri: http::Uri, no_cert: bool) -> Result<Coinsupply, Strin
|
|||||||
// tokio::runtime::current_thread::Runtime::new().unwrap().block_on(runner)
|
// 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, pool: ThreadPool, c: F)
|
async fn get_block_range<F : 'static + std::marker::Send>(
|
||||||
-> Result<(), Box<dyn std::error::Error>>
|
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) {
|
where F : Fn(&[u8], u64) {
|
||||||
let mut client = get_client(uri, no_cert).await?;
|
let mut client = get_client(uri, no_cert).await?;
|
||||||
|
|
||||||
let bs = BlockId{ height: start_height, hash: vec!()};
|
let bs = BlockId { height: start_height, hash: vec![] };
|
||||||
let be = BlockId{ height: end_height, hash: vec!()};
|
let be = BlockId { height: end_height, hash: vec![] };
|
||||||
|
|
||||||
let request = Request::new(BlockRange{ start: Some(bs), end: Some(be) });
|
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>>();
|
||||||
let (tx, rx) = channel::<Option<CompactBlock>>();
|
let (ftx, frx) = channel();
|
||||||
|
|
||||||
// Channel that the processor signals it is done, so the method can return
|
pool.execute(move || {
|
||||||
let (ftx, frx) = channel();
|
while let Ok(Some(block)) = rx.recv() {
|
||||||
|
use prost::Message;
|
||||||
// The processor runs on a different thread, so that the network calls don't
|
let mut encoded_buf = vec![];
|
||||||
// block on this
|
|
||||||
pool.execute(move || {
|
match block.encode(&mut encoded_buf) {
|
||||||
while let Some(block) = rx.recv().unwrap() {
|
Ok(_) => c(&encoded_buf, block.height),
|
||||||
use prost::Message;
|
Err(e) => {
|
||||||
let mut encoded_buf = vec![];
|
eprintln!("Error encoding block: {:?}", e);
|
||||||
|
break;
|
||||||
block.encode(&mut encoded_buf).unwrap();
|
}
|
||||||
c(&encoded_buf, block.height);
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
ftx.send(Ok(())).unwrap();
|
if let Err(e) = ftx.send(Ok(())) {
|
||||||
});
|
eprintln!("Error sending completion signal: {:?}", e);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
let mut response = client.get_block_range(request).await?.into_inner();
|
let mut response = client.get_block_range(request).await?.into_inner();
|
||||||
//println!("{:?}", response);
|
|
||||||
while let Some(block) = response.message().await? {
|
|
||||||
tx.send(Some(block)).unwrap();
|
|
||||||
}
|
|
||||||
tx.send(None).unwrap();
|
|
||||||
|
|
||||||
// Wait for the processor to exit
|
while let Some(block) = response.message().await? {
|
||||||
|
if let Err(e) = tx.send(Some(block)) {
|
||||||
|
eprintln!("Error sending block to channel: {:?}", e);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Err(e) = tx.send(None) {
|
||||||
|
eprintln!("Error sending end signal to channel: {:?}", e);
|
||||||
|
}
|
||||||
|
|
||||||
frx.iter().take(1).collect::<Result<Vec<()>, String>>()?;
|
frx.iter().take(1).collect::<Result<Vec<()>, String>>()?;
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
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>
|
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) {
|
where F : Fn(&[u8], u64) {
|
||||||
|
|
||||||
|
|||||||
@@ -1503,8 +1503,7 @@ pub fn start_mempool_monitor(lc: Arc<LightClient>) -> Result<(), String> {
|
|||||||
let block_times = block_times.clone();
|
let block_times = block_times.clone();
|
||||||
|
|
||||||
pool.execute(move || {
|
pool.execute(move || {
|
||||||
println!("Fetching transactions for address: {}", address_clone);
|
|
||||||
|
|
||||||
let r = fetch_transparent_txids(
|
let r = fetch_transparent_txids(
|
||||||
&server_uri,
|
&server_uri,
|
||||||
address,
|
address,
|
||||||
@@ -1545,7 +1544,7 @@ pub fn start_mempool_monitor(lc: Arc<LightClient>) -> Result<(), String> {
|
|||||||
);
|
);
|
||||||
|
|
||||||
match ctx.send(r) {
|
match ctx.send(r) {
|
||||||
Ok(_) => println!("Successfully sent data for address: {}", address_clone),
|
Ok(_) => info!("Successfully sent data for address: {}", address_clone),
|
||||||
Err(e) => println!("Failed to send data for address: {}: {:?}", address_clone, e),
|
Err(e) => println!("Failed to send data for address: {}: {:?}", address_clone, e),
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user