Importing transactions won't load anymore all transactions in memory

Instead the streamer api is used to loaded them by chunks, right now
those chunks are by immutable files but the API is agnostic to them and
could be used with the chain sync protocol.

Author: Alenar

mithril-aggregator/src/services/cardano_transactions_importer.rs

@@ -108,16 +108,18 @@ impl CardanoTransactionsImporter {
         );
 
         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?;
+        while let Some(blocks) = streamer.poll_next().await? {
+            let parsed_transactions: Vec<CardanoTransaction> = blocks
+                .into_iter()
+                .flat_map(|b| b.into_transactions())
+                .collect();
+
+            self.transaction_store
+                .store_transactions(parsed_transactions)
+                .await?;
+        }
+
         Ok(())
     }
 

mithril-signer/src/cardano_transactions_importer.rs

@@ -108,16 +108,18 @@ impl CardanoTransactionsImporter {
         );
 
         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?;
+        while let Some(blocks) = streamer.poll_next().await? {
+            let parsed_transactions: Vec<CardanoTransaction> = blocks
+                .into_iter()
+                .flat_map(|b| b.into_transactions())
+                .collect();
+
+            self.transaction_store
+                .store_transactions(parsed_transactions)
+                .await?;
+        }
+
         Ok(())
     }