bridge: Remove pointless return type from SenderInput::run

Author: svix-jplatte

bridge/svix-bridge-plugin-kafka/src/input.rs

@@ -199,7 +199,7 @@ impl SenderInput for KafkaConsumer {
         self.transformer_tx = tx;
     }
 
-    async fn run(&self) -> std::io::Result<()> {
+    async fn run(&self) {
         let mut fails: u64 = 0;
         let mut last_fail = Instant::now();
 

bridge/svix-bridge-plugin-kafka/tests/it/kafka_consumer.rs

@@ -173,7 +173,7 @@ async fn test_consume_ok() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -245,7 +245,7 @@ async fn test_consume_transformed_json_ok() {
     plugin.set_transformer(Some(transformer_tx));
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -330,7 +330,7 @@ async fn test_consume_transformed_string_ok() {
     plugin.set_transformer(Some(transformer_tx));
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -365,7 +365,7 @@ async fn test_missing_app_id_nack() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
 
     // Wait for the consumer to connect
@@ -415,7 +415,7 @@ async fn test_missing_event_type_nack() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
 
     // Wait for the consumer to connect
@@ -467,7 +467,7 @@ async fn test_consume_svix_503() {
     let plugin = get_test_plugin(mock_server.uri(), topic, None);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;
@@ -510,7 +510,7 @@ async fn test_consume_svix_offline() {
     drop(mock_server);
 
     let handle = tokio::spawn(async move {
-        plugin.run().await.unwrap();
+        plugin.run().await;
     });
     // Wait for the consumer to connect
     tokio::time::sleep(CONNECT_WAIT_TIME).await;

bridge/svix-bridge-plugin-queue/src/lib.rs

@@ -197,7 +197,7 @@ trait Consumer {
     }
 }
 
-async fn run_inner(consumer: &(impl Consumer + Send + Sync)) -> std::io::Result<()> {
+async fn run_inner(consumer: &(impl Consumer + Send + Sync)) -> ! {
     let mut fails: u64 = 0;
     let mut last_fail = Instant::now();
     let system_name = consumer.system();

bridge/svix-bridge-plugin-queue/src/sender_input/mod.rs

@@ -101,10 +101,12 @@ impl SenderInput for QueueSender {
     fn name(&self) -> &str {
         &self.name
     }
+
     fn set_transformer(&mut self, tx: Option<TransformerTx>) {
         self.transformer_tx = tx;
     }
-    async fn run(&self) -> std::io::Result<()> {
+
+    async fn run(&self) {
         run_inner(self).await
     }
 }

bridge/svix-bridge-types/src/lib.rs

@@ -137,7 +137,7 @@ pub trait SenderInput: Send {
     /// For plugins that want to run JS transformations on payloads.
     /// Giving them a sender lets them pass messages to the JS executor.
     fn set_transformer(&mut self, _tx: Option<TransformerTx>) {}
-    async fn run(&self) -> std::io::Result<()>;
+    async fn run(&self);
 }
 
 /// Represents something we can hand a webhook payload to.

bridge/svix-bridge/src/main.rs

@@ -166,17 +166,10 @@ async fn supervise_senders(inputs: Vec<Box<dyn SenderInput>>) -> Result<()> {
         set.spawn(async move {
             // FIXME: needs much better signaling for termination
             loop {
-                let fut = input.run();
                 // If this future returns, the consumer terminated unexpectedly.
-                if let Err(e) = fut.await {
-                    tracing::warn!(
-                        "sender input {} unexpectedly terminated: {}",
-                        input.name(),
-                        e
-                    );
-                } else {
-                    tracing::warn!("sender input {} unexpectedly terminated", input.name());
-                }
+                input.run().await;
+
+                tracing::warn!("sender input {} unexpectedly terminated", input.name());
                 tokio::time::sleep(Duration::from_secs(1)).await;
             }
         });