fixed compile error

cmd/db-exporter/src/command_decode_payload.rs

@@ -4,31 +4,30 @@
 use crate::command_progress::{
     ParallelCommand, ParallelCommandFilter, ParallelCommandObserver, ParallelCommandProgress,
 };
+use crate::init_db_obj;
 use anyhow::Result;
 use chrono::{TimeZone, Utc};
 use clap::Parser;
 use csv::{Writer, WriterBuilder};
 use move_binary_format::errors::{Location, PartialVMError};
 use serde::Serialize;
-use starcoin_crypto::hash::CryptoHash;
-use starcoin_crypto::HashValue;
+use starcoin_abi_decoder;
+use starcoin_abi_decoder::DecodedTransactionPayload;
+use starcoin_crypto::{hash::CryptoHash, HashValue};
+use starcoin_statedb::ChainStateDB;
+use starcoin_storage::Storage;
 use starcoin_types::{block::Block, transaction::TransactionPayload};
 use starcoin_vm_types::errors::VMError;
 use std::fs::File;
 use std::sync::{Arc, Mutex};
 use std::{fmt::Debug, path::PathBuf};
-use starcoin_abi_decoder;
-use starcoin_abi_decoder::{DecodedTransactionPayload};
-use starcoin_statedb::ChainStateDB;
-use starcoin_storage::{Storage};
-use crate::init_db_obj;
 
 const DECODE_PAYLOAD_COMMAND_NAME: &str = "decode_payload_command";
 
 #[derive(Debug, Parser)]
 #[clap(
-name = "decode-payload",
-about = "Decode payload for given parameter and function name"
+    name = "decode-payload",
+    about = "Decode payload for given parameter and function name"
 )]
 pub struct DecodePayloadCommandOptions {
     #[clap(long, short = 'i', parse(from_os_str))]
@@ -96,7 +95,6 @@ pub struct CSVHeaders {
 }
 
 pub struct CommandDecodePayload {
-    out_path: PathBuf,
     writer_mutex: Mutex<Writer<File>>,
     storage: Arc<Storage>,
 }
@@ -115,80 +113,84 @@ impl ParallelCommandObserver for CommandDecodePayload {
 
 impl ParallelCommand<CommandDecodePayload, Block, DecodePayloadCommandError> for Block {
     fn execute(&self, command: &CommandDecodePayload) -> (usize, Vec<DecodePayloadCommandError>) {
-        let mut errors = vec![];
-        let mut success_module_size = 0;
+        // let errors = vec![];
+        // let mut success_module_size = 0;
 
         let datetime = Utc.timestamp_opt(self.header.timestamp() as i64, 0);
         let formatted_date = datetime.unwrap().format("%Y-%m-%d %H:%M:%s").to_string();
 
         let root = self.header.state_root();
-        let statedb = ChainStateDB::new(command.storage, Some(root));
+        let statedb = ChainStateDB::new(command.storage.clone(), Some(root));
 
         for txn in self.transactions() {
             let signer = txn.sender().to_string();
-            let decoded_txn_payload = starcoin_abi_decoder::decode_txn_payload(
-                &statedb, txn.payload(),
-            ).expect("Decode transaction payload failed!");
+            let decoded_txn_payload =
+                starcoin_abi_decoder::decode_txn_payload(&statedb, txn.payload())
+                    .expect("Decode transaction payload failed!");
 
             match decoded_txn_payload {
                 DecodedTransactionPayload::ScriptFunction(payload) => {
                     let mut writer = command.writer_mutex.lock().unwrap();
-                    writer.serialize(CSVHeaders {
-                        txn_hash: txn.hash().to_string(),
-                        signer,
-                        func_name: format!("{}::{}", payload.module, payload.function),
-                        //ty_args: payload.ty_args.iter().map(|a| a.to_string() + ",").collect(),
-                        //args: payload.args.iter().map(|dv| format!("{},", dv.0.as_str()).collect(),
-                        ty_args: payload.ty_args.iter().map(|a| a.to_string()).collect::<Vec<_>>().join(","),
-                        args: payload.args.iter().map(|a| a.0.to_string()).collect::<Vec<_>>().join(","),
-                        timestamp: formatted_date.clone(),
-                    }).expect("Write into CSV failed!")
+                    writer
+                        .serialize(CSVHeaders {
+                            txn_hash: txn.hash().to_string(),
+                            signer,
+                            func_name: format!("{}::{}", payload.module, payload.function),
+                            ty_args: payload
+                                .ty_args
+                                .iter()
+                                .map(|a| a.to_string())
+                                .collect::<Vec<_>>()
+                                .join(","),
+                            args: payload
+                                .args
+                                .iter()
+                                .map(|a| a.0.to_string())
+                                .collect::<Vec<_>>()
+                                .join(","),
+                            timestamp: formatted_date.clone(),
+                        })
+                        .expect("Write into CSV failed!")
                 }
                 DecodedTransactionPayload::Script(_) | DecodedTransactionPayload::Package(_) => (),
             }
         }
-        (success_module_size, errors)
-    }
-
-    fn before_command(&self, _cmd: &CommandDecodePayload) -> Result<()> {
-        Ok(())
-    }
-
-    fn after_command(&self, _cmd: &CommandDecodePayload) -> Result<()> {
-        Ok(())
+        //(success_module_size, errors)
+        (0, vec![])
     }
 
     ///
     /// Check whether the conditions are met from the list of all transactions in a block,
     /// and return false if any condition is met.
     ///
-    fn matched(&self, filters: Option<ParallelCommandFilter>) -> bool {
-        filters.as_ref().map_or_else(|| true, |f| {
-            self.transactions().iter().any(|txn| match txn.payload() {
-                TransactionPayload::ScriptFunction(payload) => {
-                    f.match_signer(&txn.sender().to_string())
-                        || f.match_func_name(payload.function().as_str())
-                        || f.match_ty_args(&payload.ty_args().to_vec())
-                        || f.match_args(&payload.args().to_vec())
-                }
-                _ => true,
-            })
-        })
+    fn matched(&self, filters: &Option<ParallelCommandFilter>) -> bool {
+        filters.as_ref().map_or_else(
+            || true,
+            |f| {
+                self.transactions().iter().any(|txn| match txn.payload() {
+                    TransactionPayload::ScriptFunction(payload) => {
+                        f.match_signer(&txn.sender().to_string())
+                            || f.match_func_name(payload.function().as_str())
+                            || f.match_ty_args(&payload.ty_args().to_vec())
+                            || f.match_args(&payload.args().to_vec())
+                    }
+                    _ => true,
+                })
+            },
+        )
     }
 }
 
-
 pub fn decode_payload(
     input_path: PathBuf,
     out_path: PathBuf,
     db_path: PathBuf,
     filter: Option<ParallelCommandFilter>,
 ) -> Result<()> {
-    let file = WriterBuilder::new().from_path(out_path.as_ref())?;
+    let file = WriterBuilder::new().from_path(out_path.clone())?;
     let writer_mutex = Mutex::new(file);
 
     let command = Arc::new(CommandDecodePayload {
-        out_path,
         writer_mutex,
         storage: init_db_obj(db_path)?,
     });
@@ -199,7 +201,8 @@ pub fn decode_payload(
         num_cpus::get(),
         filter,
         Some(command.clone() as Arc<dyn ParallelCommandObserver>),
-    ).progress::<CommandDecodePayload, Block, DecodePayloadCommandError>(&command)
+    )
+    .progress::<CommandDecodePayload, Block, DecodePayloadCommandError>(&command)
 }
 
 #[test]

cmd/db-exporter/src/command_progress.rs

@@ -23,11 +23,9 @@ impl CommandResult {
 
 pub struct ParallelCommandFilter {
     signer: Option<String>,
-    func_name: Option<String>,
-    // function name for filter, none for all
-    ty_args: Option<Vec<String>>,
-    // template parameter for filter, none for all
-    args: Option<Vec<String>>, // arguments type for filter, none for all
+    func_name: Option<String>,    // function name for filter, none for all
+    ty_args: Option<Vec<String>>, // template parameter for filter, none for all
+    args: Option<Vec<String>>,    // arguments type for filter, none for all
 }
 
 impl ParallelCommandFilter {
@@ -98,7 +96,7 @@ impl ParallelCommandProgress {
         }
     }
 
-    pub fn progress<CommandT, BodyT, ErrorT>(self, command: &CommandT) -> Result<()>
+    pub fn progress<CommandT: Sync + Send, BodyT, ErrorT>(self, command: &CommandT) -> Result<()>
     where
         BodyT: ParallelCommand<CommandT, BodyT, ErrorT>
             + Send
@@ -128,7 +126,16 @@ impl ParallelCommandProgress {
         let all_items = lines
             .par_iter()
             .map(|line| Ok(serde_json::from_str::<BodyT>(line.as_str()))?)
-            .filter(|item| item.unwrap().matched(self.filters))
+            .filter(|item| match item {
+                Ok(i) => {
+                    if i.matched(&self.filter) {
+                        true
+                    } else {
+                        false
+                    }
+                }
+                Err(_e) => false,
+            })
             .collect::<Result<Vec<BodyT>, _>>()?;
 
         let progress_bar = ProgressBar::new(all_items.len() as u64).with_style(
@@ -190,9 +197,5 @@ impl ParallelCommandProgress {
 pub trait ParallelCommand<CommandT, BodyT, ErrorT> {
     fn execute(&self, cmd: &CommandT) -> (usize, Vec<ErrorT>);
 
-    fn before_command(&self, cmd: &CommandT) -> Result<()>;
-
-    fn after_command(&self, cmd: &CommandT) -> Result<()>;
-
-    fn matched(&self, filter: Option<ParallelCommandFilter>) -> bool;
+    fn matched(&self, filter: &Option<ParallelCommandFilter>) -> bool;
 }

cmd/db-exporter/src/lib.rs

@@ -1,28 +1,20 @@
 // Copyright (c) The Starcoin Core Contributors
 // SPDX-License-Identifier: Apache-2.0
 
-use std::path::PathBuf;
-use std::sync::Arc;
+use anyhow::Result;
 use starcoin_storage::{
-    db_storage::{
-        DBStorage
-    },
-    Storage,
+    cache_storage::CacheStorage, db_storage::DBStorage, storage::StorageInstance, Storage,
     StorageVersion,
-    storage::StorageInstance,
-    cache_storage::CacheStorage
 };
-use anyhow::Result;
+use std::path::PathBuf;
+use std::sync::Arc;
 
 pub mod command_decode_payload;
 pub mod command_progress;
 pub mod verify_header;
 pub mod verify_module;
 
-
-pub fn init_db_obj(
-    db_path: PathBuf,
-) -> Result<Arc<Storage>> {
+pub fn init_db_obj(db_path: PathBuf) -> Result<Arc<Storage>> {
     let db_storage = DBStorage::open_with_cfs(
         db_path.join("starcoindb/db/starcoindb"),
         StorageVersion::current_version()
@@ -32,8 +24,7 @@ pub fn init_db_obj(
         Default::default(),
         None,
     )?;
-    Ok(Arc::new(Storage::new(StorageInstance::new_cache_and_db_instance(
-        CacheStorage::new(None),
-        db_storage,
-    ))?))
-}
+    Ok(Arc::new(Storage::new(
+        StorageInstance::new_cache_and_db_instance(CacheStorage::new(None), db_storage),
+    )?))
+}

cmd/db-exporter/src/verify_header.rs

@@ -55,15 +55,7 @@ impl ParallelCommand<VerifyHeaderCmdType, Block, VerifyHeaderError> for Block {
         }
     }
 
-    fn before_command(&self, _cmd: &VerifyHeaderCmdType) -> anyhow::Result<()> {
-        Ok(())
-    }
-
-    fn after_command(&self, _cmd: &VerifyHeaderCmdType) -> anyhow::Result<()> {
-        Ok(())
-    }
-
-    fn matched(&self, _filter: Option<ParallelCommandFilter>) -> bool {
+    fn matched(&self, _filter: &Option<ParallelCommandFilter>) -> bool {
         true
     }
 }

cmd/db-exporter/src/verify_module.rs

@@ -73,15 +73,7 @@ impl ParallelCommand<VerifyModulesType, Block, VerifyModuleError> for Block {
         (success_modules, errors)
     }
 
-    fn before_command(&self, _cmd: &VerifyModulesType) -> anyhow::Result<()> {
-        Ok(())
-    }
-
-    fn after_command(&self, _cmd: &VerifyModulesType) -> anyhow::Result<()> {
-        Ok(())
-    }
-
-    fn matched(&self, _filter: Option<ParallelCommandFilter>) -> bool {
+    fn matched(&self, _filter: &Option<ParallelCommandFilter>) -> bool {
         true
     }
 }