Adapt importers to the 'block streamer' api

Author: Alenar

mithril-aggregator/src/services/cardano_transactions_importer.rs

@@ -101,19 +101,20 @@ impl CardanoTransactionsImporter {
             );
             return Ok(());
         }
-
-        // todo: temp algorithm, should be optimized to avoid loading all blocks & transactions
-        // at once in memory (probably using iterators)
-        let scanned_blocks = self.block_scanner.scan(&self.dirpath, from, until).await?;
-        let parsed_transactions: Vec<CardanoTransaction> =
-            scanned_blocks.flat_map(|b| b.into_transactions()).collect();
         debug!(
             self.logger,
-            "TransactionsImporter retrieved '{}' Cardano transactions between immutables '{}' and '{until}'",
-            parsed_transactions.len(),
+            "TransactionsImporter will retrieve Cardano transactions between immutables '{}' and '{until}'",
             from.unwrap_or(0)
         );
 
+        let mut streamer = self.block_scanner.scan(&self.dirpath, from, until).await?;
+        let parsed_transactions: Vec<CardanoTransaction> = streamer
+            .poll_all()
+            .await?
+            .into_iter()
+            .flat_map(|b| b.into_transactions())
+            .collect();
+
         self.transaction_store
             .store_transactions(parsed_transactions)
             .await?;
@@ -178,7 +179,9 @@ impl TransactionsImporter for CardanoTransactionsImporter {
 mod tests {
     use mockall::mock;
 
-    use mithril_common::cardano_block_scanner::{DumbBlockScanner, ScannedBlock};
+    use mithril_common::cardano_block_scanner::{
+        BlockStreamer, DumbBlockScanner, DumbBlockStreamer, ScannedBlock,
+    };
     use mithril_common::crypto_helper::MKTree;
     use mithril_common::entities::{BlockNumber, BlockRangesSequence};
 
@@ -197,7 +200,7 @@ mod tests {
               dirpath: &Path,
               from_immutable: Option<ImmutableFileNumber>,
               until_immutable: ImmutableFileNumber,
-            ) -> StdResult<Box<dyn Iterator<Item = ScannedBlock>>>;
+            ) -> StdResult<Box<dyn BlockStreamer>>;
         }
     }
 
@@ -265,7 +268,7 @@ mod tests {
             scanner_mock
                 .expect_scan()
                 .withf(move |_, from, until| from.is_none() && until == &up_to_beacon)
-                .return_once(move |_, _, _| Ok(Box::new(blocks.into_iter())));
+                .return_once(move |_, _, _| Ok(Box::new(DumbBlockStreamer::new(vec![blocks]))));
             CardanoTransactionsImporter::new_for_test(Arc::new(scanner_mock), repository.clone())
         };
 
@@ -427,7 +430,9 @@ mod tests {
             scanner_mock
                 .expect_scan()
                 .withf(move |_, from, until| from == &Some(12) && until == &up_to_beacon)
-                .return_once(move |_, _, _| Ok(Box::new(scanned_blocks.into_iter())))
+                .return_once(move |_, _, _| {
+                    Ok(Box::new(DumbBlockStreamer::new(vec![scanned_blocks])))
+                })
                 .once();
             CardanoTransactionsImporter::new_for_test(Arc::new(scanner_mock), repository.clone())
         };
@@ -605,13 +610,9 @@ mod tests {
         let transactions = into_transactions(&blocks);
         let importer = {
             let connection = cardano_tx_db_connection().unwrap();
-            let mut scanner = MockBlockScannerImpl::new();
-            scanner
-                .expect_scan()
-                .return_once(move |_, _, _| Ok(Box::new(blocks.into_iter())));
 
             CardanoTransactionsImporter::new_for_test(
-                Arc::new(scanner),
+                Arc::new(DumbBlockScanner::new(blocks.clone())),
                 Arc::new(CardanoTransactionRepository::new(Arc::new(connection))),
             )
         };

mithril-signer/src/cardano_transactions_importer.rs

@@ -101,19 +101,20 @@ impl CardanoTransactionsImporter {
             );
             return Ok(());
         }
-
-        // todo: temp algorithm, should be optimized to avoid loading all blocks & transactions
-        // at once in memory (probably using iterators)
-        let scanned_blocks = self.block_scanner.scan(&self.dirpath, from, until).await?;
-        let parsed_transactions: Vec<CardanoTransaction> =
-            scanned_blocks.flat_map(|b| b.into_transactions()).collect();
         debug!(
             self.logger,
-            "TransactionsImporter retrieved '{}' Cardano transactions between immutables '{}' and '{until}'",
-            parsed_transactions.len(),
+            "TransactionsImporter will retrieve Cardano transactions between immutables '{}' and '{until}'",
             from.unwrap_or(0)
         );
 
+        let mut streamer = self.block_scanner.scan(&self.dirpath, from, until).await?;
+        let parsed_transactions: Vec<CardanoTransaction> = streamer
+            .poll_all()
+            .await?
+            .into_iter()
+            .flat_map(|b| b.into_transactions())
+            .collect();
+
         self.transaction_store
             .store_transactions(parsed_transactions)
             .await?;
@@ -178,7 +179,9 @@ impl TransactionsImporter for CardanoTransactionsImporter {
 mod tests {
     use mockall::mock;
 
-    use mithril_common::cardano_block_scanner::{DumbBlockScanner, ScannedBlock};
+    use mithril_common::cardano_block_scanner::{
+        BlockStreamer, DumbBlockScanner, DumbBlockStreamer, ScannedBlock,
+    };
     use mithril_common::crypto_helper::MKTree;
     use mithril_common::entities::{BlockNumber, BlockRangesSequence};
 
@@ -197,7 +200,7 @@ mod tests {
               dirpath: &Path,
               from_immutable: Option<ImmutableFileNumber>,
               until_immutable: ImmutableFileNumber,
-            ) -> StdResult<Box<dyn Iterator<Item = ScannedBlock>>>;
+            ) -> StdResult<Box<dyn BlockStreamer>>;
         }
     }
 
@@ -265,7 +268,7 @@ mod tests {
             scanner_mock
                 .expect_scan()
                 .withf(move |_, from, until| from.is_none() && until == &up_to_beacon)
-                .return_once(move |_, _, _| Ok(Box::new(blocks.into_iter())));
+                .return_once(move |_, _, _| Ok(Box::new(DumbBlockStreamer::new(vec![blocks]))));
             CardanoTransactionsImporter::new_for_test(Arc::new(scanner_mock), repository.clone())
         };
 
@@ -427,7 +430,9 @@ mod tests {
             scanner_mock
                 .expect_scan()
                 .withf(move |_, from, until| from == &Some(12) && until == &up_to_beacon)
-                .return_once(move |_, _, _| Ok(Box::new(scanned_blocks.into_iter())))
+                .return_once(move |_, _, _| {
+                    Ok(Box::new(DumbBlockStreamer::new(vec![scanned_blocks])))
+                })
                 .once();
             CardanoTransactionsImporter::new_for_test(Arc::new(scanner_mock), repository.clone())
         };
@@ -605,13 +610,9 @@ mod tests {
         let transactions = into_transactions(&blocks);
         let importer = {
             let connection = cardano_tx_db_connection().unwrap();
-            let mut scanner = MockBlockScannerImpl::new();
-            scanner
-                .expect_scan()
-                .return_once(move |_, _, _| Ok(Box::new(blocks.into_iter())));
 
             CardanoTransactionsImporter::new_for_test(
-                Arc::new(scanner),
+                Arc::new(DumbBlockScanner::new(blocks.clone())),
                 Arc::new(CardanoTransactionRepository::new(Arc::new(connection))),
             )
         };