Induce data availability

.gitignore

@@ -15,5 +15,6 @@
 /target
 
 *.log
-
+.vscode
+.cargo
 config/block_submitter.yaml

Cargo.toml

@@ -12,6 +12,7 @@ config_rs = { package = "config", version = "0.10.1" }
 const_format = "0.2.15"
 crossbeam-channel = "0.5.1"
 ctrlc = { version = "3.1", features = [ "termination" ] }
+clap = "=3.0.0-beta.5" #todo: need to update the dependency after official released
 dotenv = "0.15.0"
 ethers = { git = "https://github.com/gakonst/ethers-rs" }
 fluidex-common = { git = "https://github.com/fluidex/common-rs", branch = "master", features = [ "kafka", "non-blocking-tracing", "rollup-state-db" ] }
@@ -26,13 +27,15 @@ serde_json = "1.0.64"
 sqlx = { version = "0.5.1", features = [ "runtime-tokio-rustls", "postgres", "chrono", "decimal", "json", "migrate" ] }
 tokio = { version = "1.0", features = [ "full" ] }
 tonic = "0.5.2"
+async-trait = "0.1.52"
 
 [build-dependencies]
 prost = "0.7.0"
 tonic-build = "0.4.0"
 
 [features]
 default = ["ganache"]
+windows_build = [ "fluidex-common/rdkafka-dynamic" ]
 ganache = []
 
 [[bin]]

src/bin/block_submitter.rs

@@ -1,8 +1,32 @@
+use clap::Parser;
 use fluidex_common::non_blocking_tracing;
 use futures::{channel::mpsc, executor::block_on, SinkExt, StreamExt};
-use regnbue_bridge::block_submitter::{storage, EthSender, Settings, TaskFetcher};
+use regnbue_bridge::block_submitter::{storage, types, EthSenderConfigure, Settings, TaskFetcher};
 use std::cell::RefCell;
 
+#[derive(Parser, Debug)]
+#[clap(version = "0.1")]
+struct Opts {
+    #[clap(subcommand)]
+    command: Option<SubCommand>,
+}
+
+#[derive(Parser, Debug)]
+enum SubCommand {
+    /// Verify a block with specified block id
+    Verify(VerifyBlock),
+    /// manual submit a block
+    Manual(ManualSubmit),
+}
+
+#[derive(Parser, Debug)]
+struct VerifyBlock {
+    block_id: i64,
+}
+
+#[derive(Parser, Debug)]
+struct ManualSubmit {}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     dotenv::dotenv().ok();
@@ -15,6 +39,37 @@ async fn main() -> anyhow::Result<()> {
     let settings: Settings = conf.try_into().unwrap();
     log::debug!("{:?}", settings);
 
+    let opts: Opts = Opts::parse();
+    log::debug!("{:?}", opts);
+
+    // TODO: maybe separate and have: 1. consumer 2. producer 3. sender
+    let dbpool = storage::from_config(&settings).await?;
+    //let eth_sender = eth
+    let client = EthSenderConfigure::from_config(&settings).await?.build(dbpool.clone());
+
+    // one-block mode
+    if let Some(sub_cmd) = opts.command {
+        match sub_cmd {
+            SubCommand::Verify(opts) => {
+                let block_id = opts.block_id;
+                let block = types::SubmitBlockArgs::fetch_by_blockid(block_id, &dbpool).await?;
+                let ret = client
+                    .verify_block(block.ok_or_else(|| anyhow::anyhow!("block {} not existed", block_id))?)
+                    .await?;
+                println!("verify block {} result: {}", block_id, ret);
+            }
+            SubCommand::Manual(_) => {
+                let block = types::SubmitBlockArgs::fetch_latest(None, &dbpool).await?;
+                let block = block.ok_or_else(|| anyhow::anyhow!("no pending block for submitting"))?;
+                client.submit_block(block).await?;
+            }
+        };
+
+        return Ok(());
+    }
+
+    // continuous mode
+
     // handle ctrl+c
     let (stop_signal_sender, mut stop_signal_receiver) = mpsc::channel(256);
     {
@@ -26,13 +81,10 @@ async fn main() -> anyhow::Result<()> {
         .expect("Error setting Ctrl-C handler");
     }
 
-    // TODO: maybe separate and have: 1. consumer 2. producer 3. sender
-    let dbpool = storage::from_config(&settings).await?;
     let (tx, rx) = crossbeam_channel::unbounded();
     let mut fetcher = TaskFetcher::from_config_with_pool(&settings, dbpool.clone());
     let fetcher_task_handle = tokio::spawn(async move { fetcher.run(tx).await });
-    let eth_sender = EthSender::from_config_with_pool(&settings, dbpool).await?;
-    let eth_sender_task_handle = tokio::spawn(async move { eth_sender.run(rx).await });
+    let eth_sender_task_handle = tokio::spawn(async move { client.run(rx).await });
 
     tokio::select! {
         _ = async { fetcher_task_handle.await } => {

src/block_submitter/config.rs

@@ -7,7 +7,7 @@ pub struct Settings {
     pub contract_abi_file_path: String,
     pub confirmations: usize, // TODO: default
     pub web3_url: String,
-    pub keystore: String,
-    pub password: String,
-    pub chain_id: u64,
+    pub keystore: Option<String>,
+    pub password: Option<String>,
+    pub chain_id: Option<u64>,
 }