Merge pull request #157 from nats-io/tyler_0.9.7

Cut 0.9.7

CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.9.7
+
+## Improvements
+
+- Improved error log output and avoid panicking
+  when problems are encountered while parsing the
+  JetStream reply subject.
+
 # 0.9.6
 
 ## New Features

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nats"
-version = "0.9.6"
+version = "0.9.7"
 description = "A Rust NATS client"
 authors = ["Derek Collison <[email protected]>", "Tyler Neely <[email protected]>", "Stjepan Glavina <[email protected]>"]
 edition = "2018"

async-nats/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "async-nats"
-version = "0.9.6"
+version = "0.9.7"
 description = "An async Rust NATS client"
 authors = ["Derek Collison <[email protected]>", "Tyler Neely <[email protected]>", "Stjepan Glavina <[email protected]>"]
 edition = "2018"
@@ -17,7 +17,7 @@ maintenance = { status = "actively-developed" }
 
 [dependencies]
 blocking = "1.0.2"
-nats = { path = "..", version = "0.9.6" }
+nats = { path = "..", version = "0.9.7" }
 
 [dev-dependencies]
 smol = "1.2.5"

src/jetstream.rs

@@ -857,7 +857,7 @@ impl Consumer {
                     return Err(Error::new(
                         ErrorKind::InvalidInput,
                         "process and process_batch are only usable from \
-                Pull-based Consumers if there is a durable_name set",
+                        Pull-based Consumers if there is a durable_name set",
                     ));
                 }
 
@@ -871,7 +871,15 @@ impl Consumer {
                 self.nc.request(&subject, AckKind::Ack)?
             };
 
-            let next_id = next.jetstream_message_info().unwrap().stream_seq;
+            let next_id = if let Some(jmi) = next.jetstream_message_info() {
+                jmi.stream_seq
+            } else {
+                return Err(Error::new(
+                    ErrorKind::Other,
+                    "failed to process jetstream message info \
+                    from message reply subject. Is your nats-server up to date?"
+                ));
+            };
 
             if self.dedupe_window.already_processed(next_id) {
                 let _dont_care = next.ack();

src/message.rs

@@ -170,18 +170,50 @@ impl Message {
         if split.next()? != "$JS" || split.next()? != "ACK" {
             return None;
         }
+
+        macro_rules! try_parse {
+            () => {
+                match str::parse(try_parse!(str)) {
+                    Ok(parsed) => parsed,
+                    Err(e) => {
+                        log::error!(
+                            "failed to parse jetstream reply \
+                            subject: {}, error: {:?}. Is your \
+                            nats-server up to date?",
+                            reply,
+                            e
+                        );
+                        return None;
+                    }
+                }
+            };
+            (str) => {
+                if let Some(next) = split.next() {
+                    next
+                } else {
+                    log::error!(
+                        "unexpectedly few tokens while parsing \
+                        jetstream reply subject: {}. Is your \
+                        nats-server up to date?",
+                        reply
+                    );
+                    return None;
+                }
+            };
+        }
+
         Some(crate::jetstream::JetStreamMessageInfo {
-            stream: split.next()?,
-            consumer: split.next()?,
-            delivered: str::parse(split.next()?).ok()?,
-            stream_seq: str::parse(split.next()?).ok()?,
-            consumer_seq: str::parse(split.next()?).ok()?,
+            stream: try_parse!(str),
+            consumer: try_parse!(str),
+            delivered: try_parse!(),
+            stream_seq: try_parse!(),
+            consumer_seq: try_parse!(),
             published: {
-                let nanos: u64 = str::parse(split.next()?).ok()?;
+                let nanos: u64 = try_parse!();
                 let offset = std::time::Duration::from_nanos(nanos);
                 std::time::UNIX_EPOCH + offset
             },
-            pending: str::parse(split.next()?).ok()?,
+            pending: try_parse!(),
         })
     }
 }

tests/jetstream.rs

@@ -72,7 +72,7 @@ fn jetstream_basics() -> io::Result<()> {
     nc.stream_info("test2")?;
     nc.create_consumer("test2", "consumer1")?;
 
-    let mut consumer2_cfg = ConsumerConfig {
+    let consumer2_cfg = ConsumerConfig {
         durable_name: Some("consumer2".to_string()),
         ack_policy: AckPolicy::All,
         deliver_subject: Some("consumer2_ds".to_string()),