fix(amqp sink): remove unnecessary unwrap & emit event dropped errors

src/internal_events/amqp.rs

@@ -151,4 +151,28 @@ pub mod sink {
             });
         }
     }
+
+    #[derive(Debug)]
+    pub struct AmqpNackError;
+
+    impl InternalEvent for AmqpNackError {
+        fn emit(self) {
+            let deliver_reason = "Received Negative Acknowledgement from AMQP broker.";
+            error!(
+                message = deliver_reason,
+                error_type = error_type::ACKNOWLEDGMENT_FAILED,
+                stage = error_stage::SENDING,
+                internal_log_rate_limit = true,
+            );
+            counter!(
+                "component_errors_total", 1,
+                "error_type" => error_type::ACKNOWLEDGMENT_FAILED,
+                "stage" => error_stage::SENDING,
+            );
+            emit!(ComponentEventsDropped::<UNINTENTIONAL> {
+                count: 1,
+                reason: deliver_reason
+            });
+        }
+    }
 }

src/sinks/amqp/service.rs

@@ -1,7 +1,7 @@
 //! The main tower service that takes the request created by the request builder
 //! and sends it to `AMQP`.
 use crate::{
-    internal_events::sink::{AmqpAcknowledgementError, AmqpDeliveryError},
+    internal_events::sink::{AmqpAcknowledgementError, AmqpDeliveryError, AmqpNackError},
     sinks::prelude::*,
 };
 use bytes::Bytes;
@@ -92,6 +92,9 @@ pub(super) enum AmqpError {
 
     #[snafu(display("Failed AMQP request: {}", error))]
     AmqpDeliveryFailed { error: lapin::Error },
+
+    #[snafu(display("Recieved Negative Acknowledgement from AMQP broker."))]
+    AmqpNack,
 }
 
 impl Service<AmqpRequest> for AmqpService {
@@ -109,11 +112,6 @@ impl Service<AmqpRequest> for AmqpService {
         let channel = Arc::clone(&self.channel);
 
         Box::pin(async move {
-            channel
-                .confirm_select(lapin::options::ConfirmSelectOptions::default())
-                .await
-                .unwrap();
-
             let byte_size = req.body.len();
             let fut = channel
                 .basic_publish(
@@ -128,11 +126,8 @@ impl Service<AmqpRequest> for AmqpService {
             match fut {
                 Ok(result) => match result.await {
                     Ok(lapin::publisher_confirm::Confirmation::Nack(_)) => {
-                        warn!("Received Negative Acknowledgement from AMQP server.");
-                        Ok(AmqpResponse {
-                            json_size: req.metadata.into_events_estimated_json_encoded_byte_size(),
-                            byte_size,
-                        })
+                        emit!(AmqpNackError);
+                        Err(AmqpError::AmqpNack)
                     }
                     Err(error) => {
                         // TODO: In due course the caller could emit these on error.

src/sinks/amqp/sink.rs

@@ -43,6 +43,7 @@ impl AmqpSink {
             .await
             .map_err(|e| BuildError::AmqpCreateFailed { source: e })?;
 
+        // Enable confirmations on the channel.
         channel
             .confirm_select(ConfirmSelectOptions::default())
             .await