Use TryStreamExt for error handling of exporters

deepeth · Jul 5, 2022 · bd934b7 · bd934b7
1 parent f123f61
commit bd934b7
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 9 deletions.
diff --git a/src/exporters/blocks.rs b/src/exporters/blocks.rs
@@ -13,7 +13,8 @@
 // limitations under the License.
 
 use futures::stream;
-use futures::StreamExt;
+use futures::stream::StreamExt;
+use futures::stream::TryStreamExt;
 
 use crate::exceptions::Result;
 use crate::BlockWorker;
@@ -36,7 +37,8 @@ impl BlockExporter {
     pub async fn export(&self) -> Result<()> {
         let jobs = self.numbers.chunks(self.ctx.get_batch_size()).len();
         stream::iter(0..jobs)
-            .for_each_concurrent(self.ctx.get_max_worker(), |job| async move {
+            .map(Ok)
+            .try_for_each_concurrent(self.ctx.get_max_worker(), |job| async move {
                 let mut block_worker = BlockWorker::create(&self.ctx);
                 let mut chunks = self.numbers.chunks(self.ctx.get_batch_size());
                 let chunk = chunks.nth(job).unwrap();
@@ -53,9 +55,8 @@ impl BlockExporter {
 
                 // Receipts.
                 let receipt_worker = ReceiptExporter::create(&self.ctx, tx_hashes);
-                receipt_worker.export().await.unwrap();
+                receipt_worker.export().await
             })
-            .await;
-        Ok(())
+            .await
     }
 }
diff --git a/src/exporters/receipts.rs b/src/exporters/receipts.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use futures::stream;
+use futures::stream::TryStreamExt;
 use futures::StreamExt;
 use web3::types::H256;
 
@@ -36,15 +37,16 @@ impl ReceiptExporter {
     pub async fn export(&self) -> Result<()> {
         let jobs = self.hashes.chunks(self.ctx.get_batch_size()).len();
         stream::iter(0..jobs)
-            .for_each_concurrent(self.ctx.get_max_worker(), |job| async move {
+            .map(Ok)
+            .try_for_each_concurrent(self.ctx.get_max_worker(), |job| async move {
                 let mut worker = ReceiptWorker::create(&self.ctx);
                 let mut chunks = self.hashes.chunks(self.ctx.get_batch_size());
                 let chunk = chunks.nth(job).unwrap();
 
                 worker.push_batch(chunk.to_vec()).unwrap();
-                let _res = worker.execute().await.unwrap();
+                worker.execute().await?;
+                Ok(())
             })
-            .await;
-        Ok(())
+            .await
     }
 }