Add subject to slow consumers event

Author: MasterPtato

async-nats/src/lib.rs

@@ -748,11 +748,14 @@ impl ConnectionHandler {
                                 }
                             }
                         }
-                        Err(mpsc::error::TrySendError::Full(_)) => {
+                        Err(mpsc::error::TrySendError::Full(returned_message)) => {
                             debug!("slow consumer detected for subscription {}", sid);
                             self.connector
                                 .events_tx
-                                .try_send(Event::SlowConsumer(sid))
+                                .try_send(Event::SlowConsumer(SlowConsumer {
+                                    sid,
+                                    subject: returned_message.subject,
+                                }))
                                 .ok();
                         }
                         Err(mpsc::error::TrySendError::Closed(_)) => {
@@ -1103,7 +1106,7 @@ pub enum Event {
     LameDuckMode,
     Draining,
     Closed,
-    SlowConsumer(u64),
+    SlowConsumer(SlowConsumer),
     ServerError(ServerError),
     ClientError(ClientError),
 }
@@ -1116,7 +1119,11 @@ impl fmt::Display for Event {
             Event::LameDuckMode => write!(f, "lame duck mode detected"),
             Event::Draining => write!(f, "draining"),
             Event::Closed => write!(f, "closed"),
-            Event::SlowConsumer(sid) => write!(f, "slow consumers for subscription {sid}"),
+            Event::SlowConsumer(slow_consumer) => write!(
+                f,
+                "slow consumers for subscription {} on subject {}",
+                slow_consumer.sid, slow_consumer.subject
+            ),
             Event::ServerError(err) => write!(f, "server error: {err}"),
             Event::ClientError(err) => write!(f, "client error: {err}"),
         }
@@ -1437,7 +1444,7 @@ impl From<ClientError> for CallbackError {
 #[derive(Clone, Debug, Eq, PartialEq, Error)]
 pub enum ServerError {
     AuthorizationViolation,
-    SlowConsumer(u64),
+    SlowConsumer(SlowConsumer),
     Other(String),
 }
 
@@ -1469,12 +1476,28 @@ impl std::fmt::Display for ServerError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::AuthorizationViolation => write!(f, "nats: authorization violation"),
-            Self::SlowConsumer(sid) => write!(f, "nats: subscription {sid} is a slow consumer"),
+            Self::SlowConsumer(slow_consumer) => write!(
+                f,
+                "nats: subscription {} on subject {} is a slow consumer",
+                slow_consumer.sid, slow_consumer.subject,
+            ),
             Self::Other(error) => write!(f, "nats: {error}"),
         }
     }
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SlowConsumer {
+    pub sid: u64,
+    pub subject: Subject,
+}
+
+impl std::fmt::Display for SlowConsumer {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "slow consumer {} on subject {}", self.sid, self.subject)
+    }
+}
+
 /// Info to construct a CONNECT message.
 #[derive(Clone, Debug, Serialize)]
 pub struct ConnectInfo {

async-nats/tests/client_tests.rs

@@ -569,8 +569,8 @@ mod client {
             .event_callback(move |event| {
                 let tx = tx.clone();
                 async move {
-                    if let Event::SlowConsumer(_) = event {
-                        tx.send(()).await.unwrap()
+                    if let Event::SlowConsumer(sc) = event {
+                        tx.send(sc).await.unwrap()
                     }
                 }
             })
@@ -587,15 +587,18 @@ mod client {
 
         tokio::time::sleep(Duration::from_secs(1)).await;
 
-        tokio::time::timeout(Duration::from_secs(5), rx.recv())
+        let sc = tokio::time::timeout(Duration::from_secs(5), rx.recv())
             .await
             .unwrap()
             .unwrap();
-        tokio::time::timeout(Duration::from_secs(5), rx.recv())
+        assert_eq!(sc.subject.as_str(), "data");
+        let sc = tokio::time::timeout(Duration::from_secs(5), rx.recv())
             .await
             .unwrap()
             .unwrap();
+        assert_eq!(sc.subject.as_str(), "data");
     }
+
     #[tokio::test]
     async fn no_echo() {
         // no_echo disabled.