Fuse the pull consumer Stream after terminal error

async-nats/src/jetstream/consumer/pull.rs

@@ -406,6 +406,7 @@ pub struct Stream {
     heartbeat_handle: Option<JoinHandle<()>>,
     last_seen: Arc<Mutex<Instant>>,
     heartbeats_missing: tokio::sync::mpsc::Receiver<()>,
+    terminated: bool,
 }
 
 impl Drop for Stream {
@@ -469,7 +470,11 @@ impl Stream {
                                         pending_reset = true;
                                     }
                                 }
-                                Err(err) => request_result_tx.send(Err(err)).await.unwrap(),
+                                Err(err) => {
+                                     if let Err(err) = request_result_tx.send(Err(err)).await {
+                                        debug!("failed to sent request result: {}", err);
+                                    }
+                                },
                             }
                         },
                         _ = request_rx.changed() => debug!("task received request request"),
@@ -538,6 +543,7 @@ impl Stream {
             pending_request: false,
             last_seen,
             heartbeats_missing: missed_heartbeat_rx,
+            terminated: false,
         })
     }
 }
@@ -549,6 +555,9 @@ impl futures::Stream for Stream {
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Option<Self::Item>> {
+        if self.terminated {
+            return Poll::Ready(None);
+        }
         loop {
             trace!("pending messages: {}", self.pending_messages);
             if (self.pending_messages <= self.batch_config.batch / 2
@@ -564,17 +573,19 @@ impl futures::Stream for Stream {
                 match self.heartbeats_missing.poll_recv(cx) {
                     Poll::Ready(resp) => match resp {
                         Some(()) => {
+                            self.terminated = true;
                             trace!("received missing heartbeats notification");
                             return Poll::Ready(Some(Err(Box::new(std::io::Error::new(
                                 std::io::ErrorKind::TimedOut,
                                 "did not receive idle heartbeat in time",
                             )))));
                         }
                         None => {
+                            self.terminated = true;
                             return Poll::Ready(Some(Err(Box::new(std::io::Error::new(
                                 std::io::ErrorKind::Other,
                                 "unexpected termination of heartbeat checker",
-                            )))))
+                            )))));
                         }
                     },
                     Poll::Pending => {

async-nats/tests/jetstream_tests.rs

@@ -1905,6 +1905,8 @@ mod jetstream {
                 .kind(),
             std::io::ErrorKind::TimedOut
         );
+        // after terminal error, consumer should always return none.
+        assert!(messages.next().await.is_none());
         assert!(now.elapsed().le(&Duration::from_secs(50)));
     }