Merge pull request nats-io#593 from nats-io/js_discard_new_per_subject

[ADDED] JetStream: `DiscardNewPerSubject` stream configuration
jsallay · Sep 22, 2022 · 6dd6491 · 6dd6491
2 parents dd3a5aa + af7dc26
commit 6dd6491
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 1 deletion.
diff --git a/src/jsm.c b/src/jsm.c
@@ -613,6 +613,7 @@ js_unmarshalStreamConfig(nats_JSON *json, const char *fieldName, jsStreamConfig
     IFOK(s, _unmarshalRePublish(jcfg, "republish", &(cfg->RePublish)));
     IFOK(s, nats_JSONGetBool(jcfg, "allow_direct", &(cfg->AllowDirect)));
     IFOK(s, nats_JSONGetBool(jcfg, "mirror_direct", &(cfg->MirrorDirect)));
+    IFOK(s, nats_JSONGetBool(jcfg, "discard_new_per_subject", &(cfg->DiscardNewPerSubject)));
 
     if (s == NATS_OK)
         *new_cfg = cfg;
@@ -732,6 +733,8 @@ js_marshalStreamConfig(natsBuffer **new_buf, jsStreamConfig *cfg)
         IFOK(s, natsBuf_Append(buf, ",\"allow_direct\":true", -1));
     if ((s == NATS_OK) && cfg->MirrorDirect)
         IFOK(s, natsBuf_Append(buf, ",\"mirror_direct\":true", -1));
+    if ((s == NATS_OK) && cfg->DiscardNewPerSubject)
+        IFOK(s, natsBuf_Append(buf, ",\"discard_new_per_subject\":true", -1));
 
     IFOK(s, natsBuf_AppendByte(buf, '}'));
 

diff --git a/src/nats.h b/src/nats.h
@@ -504,6 +504,9 @@ typedef struct jsStreamConfig {
         // Allow higher performance and unified direct access for mirrors as well.
         bool                    MirrorDirect;
 
+        // Allow KV like semantics to also discard new on a per subject basis
+        bool                    DiscardNewPerSubject;
+
 } jsStreamConfig;
 
 /**

diff --git a/test/test.c b/test/test.c
@@ -21805,6 +21805,7 @@ test_JetStreamMarshalStreamConfig(void)
     sc.AllowRollup = true;
     sc.AllowDirect = true;
     sc.MirrorDirect = true;
+    sc.DiscardNewPerSubject = true;
 
     test("RePublish init err: ");
     s = jsRePublish_Init(NULL);
@@ -21887,7 +21888,8 @@ test_JetStreamMarshalStreamConfig(void)
                 && (strcmp(rsc->RePublish->Destination, "RP.>") == 0)
                 && rsc->RePublish->HeadersOnly
                 && rsc->AllowDirect
-                && rsc->MirrorDirect);
+                && rsc->MirrorDirect
+                && rsc->DiscardNewPerSubject);
     js_destroyStreamConfig(rsc);
     rsc = NULL;
     // Check that this does not crash