refactor(chainsync): chainsync refactor

Cargo.toml

@@ -34,19 +34,10 @@ tokio = { version = "1.15.0", features = ["full"]}
 [dev-dependencies]
 env_logger = "0.9.0"
 futures = "0.3.8"
-rusqlite = { version = "0.26.0", features = ["bundled"] }
 oura = "1.1.0"
 pallas = "0.4.0"
 sled = "0.34.7"
 
 [[example]]
 name = "common"
 crate-type = ["staticlib"]
-
-[[example]]
-name = "sqlite"
-crate-type = ["staticlib"]
-
-[[example]]
-name = "pooltool"
-crate-type = ["staticlib"]

examples/blockfetch.rs

@@ -13,31 +13,6 @@ use cardano_ouroboros_network::{
     protocols::handshake::Handshake,
 };
 
-use std::sync::Arc;
-
-use pallas::ledger::alonzo::{
-    crypto::hash_block_header,
-    BlockWrapper,
-    Fragment,
-};
-
-use blake2b_simd::Params;
-
-use oura::{
-    mapper::ChainWellKnownInfo,
-    mapper::Config,
-    mapper::EventWriter,
-    pipelining::new_inter_stage_channel,
-    pipelining::SinkProvider,
-    sources::MagicArg,
-    utils::{
-        Utils,
-        WithUtils,
-    },
-};
-
-use log::debug;
-
 mod common;
 
 async fn blockfetch() -> Result<(), Box<dyn std::error::Error>> {
@@ -64,37 +39,13 @@ async fn blockfetch() -> Result<(), Box<dyn std::error::Error>> {
             26250057,
             hex::decode("5fec758c8aaff4a7683c27b075dc3984d8d982839cc56470a682d1411c9f8198")?,
         )
-        .build()?;
-    let mut blocks = blockfetch.run(&mut connection).await?;
-
-    let (tx, rx) = new_inter_stage_channel(None);
-    let config = Config {
-        include_block_end_events: true,
-        ..Default::default()
-    };
-
-    let well_known = ChainWellKnownInfo::try_from_magic(*MagicArg::default()).unwrap();
-    let utils = Arc::new(Utils::new(well_known, None));
-    let writer = EventWriter::standalone(tx, None, config);
-    let sink_handle = WithUtils::new(
-        oura::sinks::terminal::Config {
-            throttle_min_span_millis: Some(0),
-        },
-        utils,
-    )
-    .bootstrap(rx)?;
-    let block_db = cfg.sdb.open_tree("blocks").unwrap();
+        .build(&mut connection)?;
+    let mut blocks = blockfetch.run().await?;
 
     while let Some(block) = blocks.next().await? {
-        let block_raw = block.clone();
-        let block = BlockWrapper::decode_fragment(&block[..])?;
-        let hash = hash_block_header(&block.1.header);
-        //debug!("HASH: {}", hash);
-        block_db.insert(&hash, &*block_raw);
-        writer.crawl(&block.1).unwrap();
+        cfg.handle_block(&block)?;
     }
 
-    sink_handle.join().unwrap();
     Ok(())
 }
 

examples/common.rs

@@ -7,22 +7,103 @@
 // SPDX-License-Identifier: MPL-2.0
 //
 
-use std::path::PathBuf;
+use std::sync::Arc;
+
+use pallas::ledger::alonzo::{
+    crypto::hash_block_header,
+    BlockWrapper,
+    Fragment,
+};
+
+use oura::{
+    mapper::ChainWellKnownInfo,
+    mapper::Config as MapperConfig,
+    mapper::EventWriter,
+    pipelining::new_inter_stage_channel,
+    pipelining::SinkProvider,
+    sources::MagicArg,
+    utils::{
+        Utils,
+        WithUtils,
+    },
+};
+
+use cardano_ouroboros_network::model::Point;
 
 #[derive(Clone)]
 pub struct Config {
-    pub db: PathBuf,
     pub sdb: sled::Db,
     pub host: String,
     pub magic: u32,
+    pub writer: EventWriter,
+    pub byron_mainnet: Point,
+    pub byron_testnet: Point,
+    pub byron_guild: Point,
 }
 
 pub fn init() -> Config {
-    env_logger::init();
-    Config {
-        db: PathBuf::from("sqlite.db"),
-        sdb: sled::open(".db").unwrap(),
-        host: "relays-new.cardano-mainnet.iohk.io:3001".to_string(),
-        magic: 764824073,
+    Config::new()
+}
+
+impl Config {
+    fn new() -> Config {
+        env_logger::init();
+        Config {
+            sdb: sled::open(".db").unwrap(),
+            host: "relays-new.cardano-mainnet.iohk.io:3001".to_string(),
+            magic: 764824073,
+            writer: oura_init().unwrap(),
+            byron_mainnet: (
+                4492799,
+                "f8084c61b6a238acec985b59310b6ecec49c0ab8352249afd7268da5cff2a457",
+            )
+                .try_into()
+                .unwrap(),
+            byron_testnet: (
+                1598399,
+                "7e16781b40ebf8b6da18f7b5e8ade855d6738095ef2f1c58c77e88b6e45997a4",
+            )
+                .try_into()
+                .unwrap(),
+            byron_guild: (
+                719,
+                "e5400faf19e712ebc5ff5b4b44cecb2b140d1cca25a011e36a91d89e97f53e2e",
+            )
+                .try_into()
+                .unwrap(),
+        }
     }
+
+    #[allow(dead_code)]
+    pub fn handle_block(&self, data: &[u8]) -> Result<(), Box<dyn std::error::Error>> {
+        let block_db = self.sdb.open_tree("blocks").unwrap();
+
+        let block = BlockWrapper::decode_fragment(&data[..])?;
+        let hash = hash_block_header(&block.1.header);
+        //debug!("HASH: {}", hash);
+        block_db.insert(&hash, &*data)?;
+        self.writer.crawl(&block.1).unwrap();
+        Ok(())
+    }
+}
+
+fn oura_init() -> Result<EventWriter, Box<dyn std::error::Error>> {
+    let (tx, rx) = new_inter_stage_channel(None);
+    let config = MapperConfig {
+        include_block_end_events: true,
+        ..Default::default()
+    };
+
+    let well_known = ChainWellKnownInfo::try_from_magic(*MagicArg::default()).unwrap();
+    let utils = Arc::new(Utils::new(well_known, None));
+    let writer = EventWriter::standalone(tx, None, config);
+    let _sink_handle = WithUtils::new(
+        oura::sinks::terminal::Config {
+            throttle_min_span_millis: Some(0),
+        },
+        utils,
+    )
+    .bootstrap(rx)?;
+    Ok(writer)
+    // sink_handle.join().unwrap();
 }