Merge pull request #559 from nats-io/kv_sub_pending_limits

[FIXED] KeyValue: set watcher pending limit to unlimited
nats-io · Jul 18, 2022 · b1b81b1 · b1b81b1
2 parents 10ff89d + 2d0d81f
commit b1b81b1
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 0 deletions.
diff --git a/src/kv.c b/src/kv.c
@@ -998,6 +998,7 @@ kvStore_Watch(kvWatcher **new_watcher, kvStore *kv, const char *key, kvWatchOpti
                 w->ignoreDel = true;
         }
         s = js_SubscribeSync(&(w->sub), kv->js, natsBuf_Data(&buf), NULL, &so, NULL);
+        IFOK(s, natsSubscription_SetPendingLimits(w->sub, -1, -1));
         if (s == NATS_OK)
         {
             natsSubscription *sub = w->sub;

diff --git a/test/test.c b/test/test.c
@@ -28440,6 +28440,8 @@ test_KeyValueWatch(void)
     kvWatcher           *w  = NULL;
     kvEntry             *e  = NULL;
     natsThread          *t  = NULL;
+    int                 plc = 0;
+    int                 plb = 0;
     kvConfig            kvc;
     int64_t             start;
 
@@ -28551,6 +28553,12 @@ test_KeyValueWatch(void)
     s = kvStore_Watch(&w, kv, "t.*", NULL);
     testCond(s == NATS_OK);
 
+    test("Check pending limits: ");
+    natsMutex_Lock(w->mu);
+    s = natsSubscription_GetPendingLimits(w->sub, &plc, &plb);
+    natsMutex_Unlock(w->mu);
+    testCond((s == NATS_OK) && (plc == -1) && (plb == -1));
+
     testCond(_expectUpdate(w, "t.name", "ik", 8));
     testCond(_expectUpdate(w, "t.age", "49", 10));
     testCond(_expectInitDone(w));