Pull Request Validate Idle Heartbeat Against Expiration (#1036)

nats-io · Nov 3, 2023 · 1c63c16 · 1c63c16
1 parent fe91937
commit 1c63c16
Show file tree

Hide file tree

Showing 4 changed files with 52 additions and 10 deletions.
diff --git a/src/main/java/io/nats/client/PullRequestOptions.java b/src/main/java/io/nats/client/PullRequestOptions.java
@@ -196,11 +196,25 @@ public Builder idleHeartbeat(Duration idleHeartbeat) {
         }
 
         /**
-         * Build the PullRequestOptions. Validates that the batch size is greater than 0
+         * Build the PullRequestOptions.
+         * <p>Validates that the batch size is greater than 0</p>
+         * <p>If supplied, validates that the idle heartbeat is valid for the expiration</p>
          * @return the built PullRequestOptions
          */
         public PullRequestOptions build() {
             validateGtZero(batchSize, "Pull batch size");
+            if (idleHeartbeat != null) {
+                long idleNanosTemp = idleHeartbeat.toNanos() * 2;
+                if (idleNanosTemp > 0) {
+                    if (expiresIn == null) {
+                        throw new IllegalArgumentException("Idle Heartbeat not allowed without expiration.");
+                    }
+                    long expiresNanos = expiresIn.toNanos();
+                    if (idleNanosTemp > expiresNanos) {
+                        throw new IllegalArgumentException("Idle Heartbeat cannot be more than half the expiration.");
+                    }
+                }
+            }
             return new PullRequestOptions(this);
         }
     }

diff --git a/src/main/java/io/nats/client/api/SubjectTransform.java b/src/main/java/io/nats/client/api/SubjectTransform.java
@@ -138,4 +138,12 @@ public int hashCode() {
         result = 31 * result + (destination != null ? destination.hashCode() : 0);
         return result;
     }
+
+    @Override
+    public String toString() {
+        return "SubjectTransform{" +
+            "source='" + source + '\'' +
+            ", destination='" + destination + '\'' +
+            '}';
+    }
 }
diff --git a/src/test/java/io/nats/client/impl/JetStreamPullTests.java b/src/test/java/io/nats/client/impl/JetStreamPullTests.java
@@ -16,6 +16,7 @@
 import io.nats.client.*;
 import io.nats.client.api.AckPolicy;
 import io.nats.client.api.ConsumerConfiguration;
+import io.nats.client.support.JsonUtils;
 import io.nats.client.support.Status;
 import io.nats.client.utils.TestBase;
 import org.junit.jupiter.api.Disabled;
@@ -30,6 +31,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static io.nats.client.api.ConsumerConfiguration.builder;
+import static io.nats.client.support.ApiConstants.*;
 import static io.nats.client.support.Status.*;
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -712,6 +714,9 @@ private interface SubscriptionSupplier {
     public void testPullRequestOptionsBuilder() {
         assertThrows(IllegalArgumentException.class, () -> PullRequestOptions.builder(0).build());
         assertThrows(IllegalArgumentException.class, () -> PullRequestOptions.builder(-1).build());
+        assertThrows(IllegalArgumentException.class, () -> PullRequestOptions.builder(1).idleHeartbeat(1).build());
+        assertThrows(IllegalArgumentException.class, () -> PullRequestOptions.builder(1).noWait().idleHeartbeat(1).build());
+        assertThrows(IllegalArgumentException.class, () -> PullRequestOptions.builder(1).expiresIn(30000).idleHeartbeat(15001).build());
 
         PullRequestOptions pro = PullRequestOptions.builder(11).build();
         assertEquals(11, pro.getBatchSize());
@@ -730,24 +735,24 @@ public void testPullRequestOptionsBuilder() {
         pro = PullRequestOptions.builder(31)
             .maxBytes(32)
             .expiresIn(33)
-            .idleHeartbeat(34)
+            .idleHeartbeat(16)
             .noWait()
             .build();
         assertEquals(31, pro.getBatchSize());
         assertEquals(32, pro.getMaxBytes());
         assertEquals(33, pro.getExpiresIn().toMillis());
-        assertEquals(34, pro.getIdleHeartbeat().toMillis());
+        assertEquals(16, pro.getIdleHeartbeat().toMillis());
         assertTrue(pro.isNoWait());
 
         pro = PullRequestOptions.builder(41)
             .expiresIn(Duration.ofMillis(43))
-            .idleHeartbeat(Duration.ofMillis(44))
+            .idleHeartbeat(Duration.ofMillis(21))
             .noWait(false) // just coverage of this method
             .build();
         assertEquals(41, pro.getBatchSize());
         assertEquals(0, pro.getMaxBytes());
         assertEquals(43, pro.getExpiresIn().toMillis());
-        assertEquals(44, pro.getIdleHeartbeat().toMillis());
+        assertEquals(21, pro.getIdleHeartbeat().toMillis());
         assertFalse(pro.isNoWait());
     }
 
@@ -958,12 +963,27 @@ public void testConsumerDeletedAsyncSub() throws Exception {
         });
     }
 
+    static class BadPullRequestOptions extends PullRequestOptions {
+        public BadPullRequestOptions() {
+            super(PullRequestOptions.builder(1));
+        }
+
+        @Override
+        public String toJson() {
+            StringBuilder sb = JsonUtils.beginJson();
+            JsonUtils.addField(sb, BATCH, 1);
+            JsonUtils.addFldWhenTrue(sb, NO_WAIT, true);
+            JsonUtils.addFieldAsNanos(sb, IDLE_HEARTBEAT, Duration.ofMillis(1));
+            return JsonUtils.endJson(sb).toString();
+        }
+    }
+
     @Test
     public void testBadRequestSyncSub() throws Exception {
         testConflictStatus(400, BAD_REQUEST, TYPE_ERROR, "2.9.0", (nc, jsm, js, tsc, handler) -> {
             PullSubscribeOptions so = makePso(b -> b);
             JetStreamSubscription sub = js.subscribe(tsc.subject(), so);
-            sub.pull(PullRequestOptions.builder(1).noWait().idleHeartbeat(1).build());
+            sub.pull(new BadPullRequestOptions());
             return sub;
         });
     }
@@ -974,7 +994,7 @@ public void testBadRequestAsyncSub() throws Exception {
             Dispatcher d = nc.createDispatcher();
             PullSubscribeOptions so = makePso(b -> b);
             JetStreamSubscription sub = js.subscribe(tsc.subject(), d, m -> {}, so);
-            sub.pull(PullRequestOptions.builder(1).noWait().idleHeartbeat(1).build());
+            sub.pull(new BadPullRequestOptions());
             return sub;
         });
     }

diff --git a/src/test/java/io/nats/client/impl/MessageManagerTests.java b/src/test/java/io/nats/client/impl/MessageManagerTests.java
@@ -138,7 +138,7 @@ public void testPullBeforeQueueProcessorAndManage() throws Exception {
             testPullBqpAndManage(sub, handler, pullMgr);
 
             pullMgr = getPullManager(nc, sub, true);
-            pullMgr.startPullRequest("pullSubject", PullRequestOptions.builder(1).idleHeartbeat(100).build(), true, null);
+            pullMgr.startPullRequest("pullSubject", PullRequestOptions.builder(1).expiresIn(10000).idleHeartbeat(100).build(), true, null);
             testPullBqpAndManage(sub, handler, pullMgr);
         });
     }
@@ -253,13 +253,13 @@ public void testPullManagerHeartbeats() throws Exception {
 
             handler.reset();
             handler.prepForHeartbeatAlarm();
-            pullMgr.startPullRequest("pullSubject", PullRequestOptions.builder(1).idleHeartbeat(100).build(), false, null);
+            pullMgr.startPullRequest("pullSubject", PullRequestOptions.builder(1).expiresIn(10000).idleHeartbeat(100).build(), false, null);
             TestHandler.HeartbeatAlarmEvent event = handler.waitForHeartbeatAlarm(1000);
             assertNotNull(event);
 
             handler.reset();
             handler.prepForHeartbeatAlarm();
-            pullMgr.startPullRequest("pullSubject", PullRequestOptions.builder(1).idleHeartbeat(100).build(), false, null);
+            pullMgr.startPullRequest("pullSubject", PullRequestOptions.builder(1).expiresIn(10000).idleHeartbeat(100).build(), false, null);
             event = handler.waitForHeartbeatAlarm(1000);
             assertNotNull(event);