Fix pull consumer without hearbeat

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>
nats-io · Oct 19, 2022 · 246b1c8 · 246b1c8
1 parent 1ac548d
commit 246b1c8
Show file tree

Hide file tree

Showing 2 changed files with 67 additions and 15 deletions.
diff --git a/async-nats/src/jetstream/consumer/pull.rs b/async-nats/src/jetstream/consumer/pull.rs
@@ -542,21 +542,23 @@ impl futures::Stream for Stream {
                 self.request_tx.send(()).unwrap();
                 self.pending_request = true;
             }
-            match self.heartbeats_missing.try_recv() {
-                Ok(_) => {
-                    return Poll::Ready(Some(Err(Box::new(std::io::Error::new(
-                        std::io::ErrorKind::TimedOut,
-                        "did not receive idle heartbeat in time",
-                    )))))
-                }
-                // ignore this error as that means we haven't got any missing heartbeats that we
-                // haven't read.
-                Err(TryRecvError::Empty) => (),
-                Err(TryRecvError::Closed) => {
-                    return Poll::Ready(Some(Err(Box::new(std::io::Error::new(
-                        std::io::ErrorKind::Other,
-                        "unexpected heartbeat error closure",
-                    )))))
+            if self.heartbeat_handle.is_some() {
+                match self.heartbeats_missing.try_recv() {
+                    Ok(_) => {
+                        return Poll::Ready(Some(Err(Box::new(std::io::Error::new(
+                            std::io::ErrorKind::TimedOut,
+                            "did not receive idle heartbeat in time",
+                        )))))
+                    }
+                    // ignore this error as that means we haven't got any missing heartbeats that we
+                    // haven't read.
+                    Err(TryRecvError::Empty) => (),
+                    Err(TryRecvError::Closed) => {
+                        return Poll::Ready(Some(Err(Box::new(std::io::Error::new(
+                            std::io::ErrorKind::Other,
+                            "unexpected heartbeat error closure",
+                        )))))
+                    }
                 }
             }
             match self.request_result_rx.poll_recv(cx) {

diff --git a/async-nats/tests/jetstream_tests.rs b/async-nats/tests/jetstream_tests.rs
@@ -1590,6 +1590,56 @@ mod jetstream {
         }
     }
     #[tokio::test]
+    async fn pull_consumer_stream_without_heartbeat() {
+        tracing_subscriber::fmt::init();
+        let server = nats_server::run_server("tests/configs/jetstream.conf");
+        let client = ConnectOptions::new()
+            .event_callback(|err| async move { println!("error: {:?}", err) })
+            .connect(server.client_url())
+            .await
+            .unwrap();
+
+        let context = async_nats::jetstream::new(client);
+
+        context
+            .create_stream(stream::Config {
+                name: "events".to_string(),
+                subjects: vec!["events".to_string()],
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        let stream = context.get_stream("events").await.unwrap();
+        stream
+            .create_consumer(consumer::pull::Config {
+                durable_name: Some("pull".to_string()),
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        let consumer: PullConsumer = stream.get_consumer("pull").await.unwrap();
+
+        context
+            .publish("events".to_string(), "dat".into())
+            .await
+            .unwrap();
+
+        let mut messages = consumer
+            .stream()
+            .max_messages_per_batch(3)
+            .messages()
+            .await
+            .unwrap();
+
+        messages.next().await.unwrap().unwrap().ack().await.unwrap();
+        let name = &consumer.cached_info().name;
+        stream.delete_consumer(name).await.unwrap();
+        let now = Instant::now();
+        tokio::time::sleep(Duration::from_secs(10)).await;
+        println!("time elapsed {:?}", now.elapsed());
+    }
+    #[tokio::test]
     async fn pull_consumer_stream_with_heartbeat() {
         let server = nats_server::run_server("tests/configs/jetstream.conf");
         let client = ConnectOptions::new()