Make sure Fetch No Wait returns as soon as it gets 404/408 (#1096)

Author: scottf

src/main/java/io/nats/client/BaseConsumeOptions.java

@@ -25,7 +25,6 @@ public class BaseConsumeOptions {
     public static final int DEFAULT_THRESHOLD_PERCENT = 25;
     public static final long DEFAULT_EXPIRES_IN_MILLIS = 30000;
     public static final long MIN_EXPIRES_MILLS = 1000;
-    public static final long MIN_NOWAIT_EXPIRES_MILLS = 100;
     public static final long MAX_HEARTBEAT_MILLIS = 30000;
     public static final int MAX_IDLE_HEARTBEAT_PERCENT = 50;
 
@@ -105,14 +104,7 @@ public B expiresIn(long expiresInMillis) {
                 }
             }
             else if (expiresInMillis < MIN_EXPIRES_MILLS) {
-                if (noWait) {
-                    if (expiresInMillis < MIN_NOWAIT_EXPIRES_MILLS) {
-                        throw new IllegalArgumentException("Expires when No Wait must be greater than or equal to " + MIN_NOWAIT_EXPIRES_MILLS);
-                    }
-                }
-                else {
-                    throw new IllegalArgumentException("Expires must be greater than or equal to " + MIN_EXPIRES_MILLS);
-                }
+                throw new IllegalArgumentException("Expires must be greater than or equal to " + MIN_EXPIRES_MILLS);
             }
             else {
                 expiresIn = expiresInMillis;

src/main/java/io/nats/client/impl/NatsFetchConsumer.java

@@ -14,15 +14,16 @@
 package io.nats.client.impl;
 
 import io.nats.client.*;
+import io.nats.client.api.ConsumerConfiguration;
 import io.nats.client.api.ConsumerInfo;
 
 import java.io.IOException;
 
 import static io.nats.client.BaseConsumeOptions.MIN_EXPIRES_MILLS;
-import static io.nats.client.BaseConsumeOptions.MIN_NOWAIT_EXPIRES_MILLS;
 
 class NatsFetchConsumer extends NatsMessageConsumerBase implements FetchConsumer, PullManagerObserver {
     private final boolean isNoWait;
+    private final boolean isNoWaitNoExpires;
     private final long maxWaitNanos;
     private final String pullSubject;
     private long startNanos;
@@ -35,15 +36,18 @@ class NatsFetchConsumer extends NatsMessageConsumerBase implements FetchConsumer
 
         isNoWait = fetchConsumeOptions.isNoWait();
         long expiresInMillis = fetchConsumeOptions.getExpiresInMillis();
+        isNoWaitNoExpires = isNoWait && expiresInMillis == ConsumerConfiguration.LONG_UNSET;
+
         long inactiveThreshold;
-        if (expiresInMillis <= MIN_NOWAIT_EXPIRES_MILLS ) { // can be for noWait
-            maxWaitNanos = MIN_NOWAIT_EXPIRES_MILLS * 1_000_000;
+        if (expiresInMillis == ConsumerConfiguration.LONG_UNSET) { // can be for noWait
+            maxWaitNanos = MIN_EXPIRES_MILLS * 1_000_000;
             inactiveThreshold = MIN_EXPIRES_MILLS; // no need to do the 10% longer
         }
         else {
             maxWaitNanos = expiresInMillis * 1_000_000;
             inactiveThreshold = expiresInMillis * 110 / 100; // 10% longer than the wait
         }
+
         PullRequestOptions pro = PullRequestOptions.builder(fetchConsumeOptions.getMaxMessages())
             .maxBytes(fetchConsumeOptions.getMaxBytes())
             .expiresIn(expiresInMillis)
@@ -78,7 +82,7 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
                 Message m = sub._nextUnmanagedNoWait(pullSubject);
                 if (m == null) {
                     // if there are no messages in the internal cache AND there are no more pending,
-                    // they all have been read and we can go ahead and close the subscription.
+                    // they all have been read and we can go ahead and finish
                     finished.set(true);
                     lenientClose();
                 }
@@ -96,14 +100,21 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
             // this might happen once, but it should already be noMorePending
             if (timeLeftMillis < 1) {
                 Message m = sub._nextUnmanagedNoWait(pullSubject); // null means don't wait
-                if (m == null && isNoWait) {
+                if (m == null) {
+                    // no message and no time left, go ahead and finish
                     finished.set(true);
                     lenientClose();
                 }
                 return m;
             }
 
-            return sub._nextUnmanaged(timeLeftMillis, pullSubject);
+            Message m = sub._nextUnmanaged(timeLeftMillis, pullSubject);
+            if (m == null && isNoWaitNoExpires) {
+                // no message and no wait, go ahead and finish
+                finished.set(true);
+                lenientClose();
+            }
+            return m;
         }
         catch (JetStreamStatusException e) {
             throw new JetStreamStatusCheckedException(e);