move bundle logic inside

Author: wlawt

crates/ingress-rpc/src/service.rs

@@ -131,44 +131,30 @@ where
         let start = Instant::now();
         let transaction = self.validate_tx(&data).await?;
 
-        let expiry_timestamp = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_secs()
-            + self.send_transaction_default_lifetime_seconds;
-
-        let bundle = Bundle {
-            txs: vec![data.clone()],
-            max_timestamp: Some(expiry_timestamp),
-            reverting_tx_hashes: vec![transaction.tx_hash()],
-            ..Default::default()
-        };
-        let parsed_bundle: ParsedBundle = bundle
-            .clone()
-            .try_into()
-            .map_err(|e: String| EthApiError::InvalidParams(e).into_rpc_err())?;
+        match self.tx_submission_method {
+            TxSubmissionMethod::Kafka => {
+                let expiry_timestamp = SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs()
+                    + self.send_transaction_default_lifetime_seconds;
+
+                let bundle = Bundle {
+                    txs: vec![data.clone()],
+                    max_timestamp: Some(expiry_timestamp),
+                    reverting_tx_hashes: vec![transaction.tx_hash()],
+                    ..Default::default()
+                };
+                let parsed_bundle: ParsedBundle = bundle
+                    .clone()
+                    .try_into()
+                    .map_err(|e: String| EthApiError::InvalidParams(e).into_rpc_err())?;
 
-        let bundle_hash = &parsed_bundle.bundle_hash();
-        let meter_bundle_response = self.meter_bundle(&bundle, bundle_hash).await?;
+                let bundle_hash = &parsed_bundle.bundle_hash();
+                let meter_bundle_response = self.meter_bundle(&bundle, bundle_hash).await?;
 
-        let accepted_bundle = AcceptedBundle::new(parsed_bundle, meter_bundle_response);
+                let accepted_bundle = AcceptedBundle::new(parsed_bundle, meter_bundle_response);
 
-        match self.tx_submission_method {
-            TxSubmissionMethod::Mempool => {
-                let response = self
-                    .provider
-                    .send_raw_transaction(data.iter().as_slice())
-                    .await;
-                match response {
-                    Ok(_) => {
-                        info!(message = "sent transaction to the mempool", hash=%transaction.tx_hash());
-                    }
-                    Err(e) => {
-                        warn!(message = "Failed to send raw transaction to mempool", error = %e);
-                    }
-                }
-            }
-            TxSubmissionMethod::Kafka => {
                 if let Err(e) = self
                     .bundle_queue
                     .publish(&accepted_bundle, bundle_hash)
@@ -187,6 +173,20 @@ where
                     warn!(message = "Failed to send audit event", error = %e);
                 }
             }
+            TxSubmissionMethod::Mempool => {
+                let response = self
+                    .provider
+                    .send_raw_transaction(data.iter().as_slice())
+                    .await;
+                match response {
+                    Ok(_) => {
+                        info!(message = "sent transaction to the mempool", hash=%transaction.tx_hash());
+                    }
+                    Err(e) => {
+                        warn!(message = "Failed to send raw transaction to mempool", error = %e);
+                    }
+                }
+            }
         }
 
         self.metrics