[do not merge] grpc test

Author: grao1991

rust/processor/src/config.rs

@@ -55,6 +55,8 @@ pub struct IndexerGrpcProcessorConfig {
     // String vector for deprecated tables to skip db writes
     #[serde(default)]
     pub deprecated_tables: HashSet<String>,
+
+    pub db_row_name: String,
 }
 
 impl IndexerGrpcProcessorConfig {
@@ -103,6 +105,7 @@ impl RunnableConfig for IndexerGrpcProcessorConfig {
             self.transaction_filter.clone(),
             self.grpc_response_item_timeout_in_secs,
             self.deprecated_tables.clone(),
+            self.db_row_name.clone(),
         )
         .await
         .context("Failed to build worker")?;

rust/processor/src/processors/mod.rs

@@ -79,6 +79,10 @@ pub struct DefaultProcessingResult {
 pub trait ProcessorTrait: Send + Sync + Debug {
     fn name(&self) -> &'static str;
 
+    fn db_row_name(&self) -> &str {
+        ""
+    }
+
     /// Process all transactions including writing to the database
     async fn process_transactions(
         &self,
@@ -135,7 +139,7 @@ pub trait ProcessorTrait: Send + Sync + Debug {
     ) -> anyhow::Result<()> {
         let timestamp = last_transaction_timestamp.map(|t| parse_timestamp(&t, version as i64));
         let status = ProcessorStatus {
-            processor: self.name().to_string(),
+            processor: self.db_row_name().to_string(),
             last_success_version: version as i64,
             last_transaction_timestamp: timestamp,
         };

rust/processor/src/processors/monitoring_processor.rs

@@ -9,11 +9,15 @@ use std::fmt::Debug;
 
 pub struct MonitoringProcessor {
     connection_pool: ArcDbPool,
+    db_row_name: String,
 }
 
 impl MonitoringProcessor {
-    pub fn new(connection_pool: ArcDbPool) -> Self {
-        Self { connection_pool }
+    pub fn new(connection_pool: ArcDbPool, db_row_name: String) -> Self {
+        Self {
+            connection_pool,
+            db_row_name,
+        }
     }
 }
 
@@ -34,6 +38,10 @@ impl ProcessorTrait for MonitoringProcessor {
         ProcessorName::MonitoringProcessor.into()
     }
 
+    fn db_row_name(&self) -> &str {
+        &self.db_row_name
+    }
+
     async fn process_transactions(
         &self,
         transactions: Vec<Transaction>,

rust/processor/src/worker.rs

@@ -132,6 +132,7 @@ pub struct Worker {
     pub transaction_filter: TransactionFilter,
     pub grpc_response_item_timeout_in_secs: u64,
     pub deprecated_tables: TableFlags,
+    pub db_row_name: String,
 }
 
 impl Worker {
@@ -155,6 +156,7 @@ impl Worker {
         transaction_filter: TransactionFilter,
         grpc_response_item_timeout_in_secs: u64,
         deprecated_tables: HashSet<String>,
+        db_row_name: String,
     ) -> Result<Self> {
         let processor_name = processor_config.name();
         info!(processor_name = processor_name, "[Parser] Kicking off");
@@ -200,6 +202,7 @@ impl Worker {
             transaction_filter,
             grpc_response_item_timeout_in_secs,
             deprecated_tables: deprecated_tables_flags,
+            db_row_name,
         })
     }
 
@@ -332,6 +335,7 @@ impl Worker {
             self.deprecated_tables,
             self.db_pool.clone(),
             maybe_gap_detector_sender,
+            self.db_row_name.clone(),
         );
 
         let gap_detector = if is_parquet_processor {
@@ -376,6 +380,7 @@ impl Worker {
                     receiver.clone(),
                     gap_detector_sender.clone(),
                     gap_detector.clone(),
+                    self.db_row_name.clone(),
                 )
                 .await;
             processor_tasks.push(join_handle);
@@ -401,6 +406,7 @@ impl Worker {
         receiver: kanal::AsyncReceiver<TransactionsPBResponse>,
         gap_detector_sender: AsyncSender<ProcessingResult>,
         mut gap_detector: GapDetector,
+        db_row_name: String,
     ) -> JoinHandle<()> {
         let processor_name = self.processor_config.name();
         let stream_address = self.indexer_grpc_data_service_address.to_string();
@@ -415,6 +421,7 @@ impl Worker {
                 self.deprecated_tables,
                 self.db_pool.clone(),
                 Some(gap_detector_sender.clone()),
+                "".to_string(),
             )
         } else {
             build_processor(
@@ -423,6 +430,7 @@ impl Worker {
                 self.deprecated_tables,
                 self.db_pool.clone(),
                 None,
+                db_row_name,
             )
         };
 
@@ -900,6 +908,7 @@ pub fn build_processor_for_testing(
         deprecated_tables,
         db_pool,
         None,
+        "".to_string(),
     )
 }
 
@@ -914,6 +923,7 @@ pub fn build_processor(
     deprecated_tables: TableFlags,
     db_pool: ArcDbPool,
     gap_detector_sender: Option<AsyncSender<ProcessingResult>>, // Parquet only
+    db_row_name: String,
 ) -> Processor {
     match config {
         ProcessorConfig::AccountTransactionsProcessor => Processor::from(
@@ -938,7 +948,9 @@ pub fn build_processor(
             per_table_chunk_sizes,
             deprecated_tables,
         )),
-        ProcessorConfig::MonitoringProcessor => Processor::from(MonitoringProcessor::new(db_pool)),
+        ProcessorConfig::MonitoringProcessor => {
+            Processor::from(MonitoringProcessor::new(db_pool, db_row_name))
+        },
         ProcessorConfig::NftMetadataProcessor(config) => {
             Processor::from(NftMetadataProcessor::new(db_pool, config.clone()))
         },