[ADDED] JetStream: Nak delay and BackOff list

src/js.c

@@ -42,7 +42,7 @@ const int64_t    jsDefaultRequestWait    = 5000;
 const int64_t    jsDefaultStallWait      = 200;
 const char       *jsDigits               = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
 const int        jsBase                  = 62;
-const int64_t    jsOrderedHBInterval     = 5*(int64_t)1E9;
+const int64_t    jsOrderedHBInterval     = NATS_SECONDS_TO_NANOS(5);
 
 #define jsReplyTokenSize    (8)
 #define jsDefaultMaxMsgs    (512 * 1024)
@@ -2071,7 +2071,7 @@ _subscribe(natsSubscription **new_sub, jsCtx *js, const char *subject, const cha
             cfg->FlowControl = true;
             cfg->AckPolicy   = js_AckNone;
             cfg->MaxDeliver  = 1;
-            cfg->AckWait     = (24*60*60)*(int64_t)1E9; // Just set to something known, not utilized.
+            cfg->AckWait     = NATS_SECONDS_TO_NANOS(24*60*60); // Just set to something known, not utilized.
             if (opts->Config.Heartbeat <= 0)
                 cfg->Heartbeat = jsOrderedHBInterval;
         }
@@ -2296,14 +2296,27 @@ js_PullSubscribe(natsSubscription **sub, jsCtx *js, const char *subject, const c
     return NATS_UPDATE_ERR_STACK(s);
 }
 
+typedef struct __ackOpts
+{
+    const char  *ackType;
+    bool        inProgress;
+    bool        sync;
+    int64_t     nakDelay;
+
+} _ackOpts;
+
 static natsStatus
-_ackMsg(natsMsg *msg, jsOptions *opts, const char *ackType, bool inProgress, bool sync, jsErrCode *errCode)
+_ackMsg(natsMsg *msg, jsOptions *opts, _ackOpts *o, jsErrCode *errCode)
 {
     natsSubscription    *sub = NULL;
     natsConnection      *nc  = NULL;
     jsCtx               *js  = NULL;
     jsSub               *jsi = NULL;
     natsStatus          s    = NATS_OK;
+    const char          *body= o->ackType;
+    bool                sync = o->sync;
+    int64_t             wait = 0;
+    char                tmp[64];
 
     if (msg == NULL)
         return nats_setDefaultError(NATS_INVALID_ARG);
@@ -2323,10 +2336,21 @@ _ackMsg(natsMsg *msg, jsOptions *opts, const char *ackType, bool inProgress, boo
     js = jsi->js;
     nc = sub->conn;
 
+    // If option with Wait is specified, transform all Acks as sync operation.
+    if ((opts != NULL) && (opts->Wait > 0))
+    {
+        wait = opts->Wait;
+        sync = true;
+    }
+    if (o->nakDelay > 0)
+    {
+        int64_t v = NATS_MILLIS_TO_NANOS(o->nakDelay);
+        snprintf(tmp, sizeof(tmp), "%s {\"delay\":%" PRId64 "}", o->ackType, v);
+        body = (const char*) tmp;
+    }
     if (sync)
     {
         natsMsg *rply   = NULL;
-        int64_t wait    = (opts != NULL ? opts->Wait : 0);
 
         if (wait == 0)
         {
@@ -2336,15 +2360,15 @@ _ackMsg(natsMsg *msg, jsOptions *opts, const char *ackType, bool inProgress, boo
             wait = js->opts.Wait;
             js_unlock(js);
         }
-        IFOK_JSR(s, natsConnection_RequestString(&rply, nc, msg->reply, ackType, wait));
+        IFOK_JSR(s, natsConnection_RequestString(&rply, nc, msg->reply, body, wait));
         natsMsg_Destroy(rply);
     }
     else
     {
-        s = natsConnection_PublishString(nc, msg->reply, ackType);
+        s = natsConnection_PublishString(nc, msg->reply, body);
     }
     // Indicate that we have ack'ed the message
-    if ((s == NATS_OK) && !inProgress)
+    if ((s == NATS_OK) && !o->inProgress)
         natsMsg_setAcked(msg);
 
     return NATS_UPDATE_ERR_STACK(s);
@@ -2353,31 +2377,61 @@ _ackMsg(natsMsg *msg, jsOptions *opts, const char *ackType, bool inProgress, boo
 natsStatus
 natsMsg_Ack(natsMsg *msg, jsOptions *opts)
 {
-    return _ackMsg(msg, opts, jsAckAck, false, false, NULL);
+    natsStatus  s;
+    _ackOpts    o = {jsAckAck, false, false, 0};
+
+    s = _ackMsg(msg, opts, &o, NULL);
+    return NATS_UPDATE_ERR_STACK(s);
 }
 
 natsStatus
 natsMsg_AckSync(natsMsg *msg, jsOptions *opts, jsErrCode *errCode)
 {
-    return _ackMsg(msg, opts, jsAckAck, false, true, errCode);
+    natsStatus  s;
+    _ackOpts    o = {jsAckAck, false, true, 0};
+
+    s = _ackMsg(msg, opts, &o, errCode);
+    return NATS_UPDATE_ERR_STACK(s);
 }
 
 natsStatus
 natsMsg_Nak(natsMsg *msg, jsOptions *opts)
 {
-    return _ackMsg(msg, opts, jsAckNak, false, false, NULL);
+    natsStatus  s;
+    _ackOpts    o = {jsAckNak, false, false, 0};
+
+    s = _ackMsg(msg, opts, &o, NULL);
+    return NATS_UPDATE_ERR_STACK(s);
+}
+
+natsStatus
+natsMsg_NakWithDelay(natsMsg *msg, int64_t delay, jsOptions *opts)
+{
+    natsStatus  s;
+    _ackOpts    o = {jsAckNak, false, false, delay};
+
+    s = _ackMsg(msg, opts, &o, NULL);
+    return NATS_UPDATE_ERR_STACK(s);
 }
 
 natsStatus
 natsMsg_InProgress(natsMsg *msg, jsOptions *opts)
 {
-    return _ackMsg(msg, opts, jsAckInProgress, true, false, NULL);
+    natsStatus  s;
+    _ackOpts    o = {jsAckInProgress, true, false, 0};
+
+    s = _ackMsg(msg, opts, &o, NULL);
+    return NATS_UPDATE_ERR_STACK(s);
 }
 
 natsStatus
 natsMsg_Term(natsMsg *msg, jsOptions *opts)
 {
-    return _ackMsg(msg, opts, jsAckTerm, false, false, NULL);
+    natsStatus  s;
+    _ackOpts    o = {jsAckTerm, false, false, 0};
+
+    s = _ackMsg(msg, opts, &o, NULL);
+    return NATS_UPDATE_ERR_STACK(s);
 }
 
 natsStatus
@@ -2548,7 +2602,7 @@ _recreateOrderedCons(void *closure)
         cc.FlowControl      = true;
         cc.AckPolicy        = js_AckNone;
         cc.MaxDeliver       = 1;
-        cc.AckWait          = (24*60*60)*(int64_t)1E9; // Just set to something known, not utilized.
+        cc.AckWait          = NATS_SECONDS_TO_NANOS(24*60*60); // Just set to something known, not utilized.
         cc.Heartbeat        = jsi->hbi * 1000000;
         cc.DeliverSubject   = sub->subject;
         cc.DeliverPolicy    = js_DeliverByStartSequence;

src/jsm.c

@@ -1710,6 +1710,21 @@ _marshalConsumerCreateReq(natsBuffer **new_buf, const char *stream, jsConsumerCo
         s = nats_marshalLong(buf, true, "max_batch", cfg->MaxRequestBatch);
     if ((s == NATS_OK) && (cfg->MaxRequestExpires > 0))
         s = nats_marshalLong(buf, true, "max_expires", cfg->MaxRequestExpires);
+    if ((s == NATS_OK) && (cfg->BackOff != NULL) && (cfg->BackOffLen > 0))
+    {
+        char    tmp[32];
+        int     i;
+
+        s = natsBuf_Append(buf, ",\"backoff\":[", -1);
+        for (i=0; (s == NATS_OK) && (i<cfg->BackOffLen); i++)
+        {
+            snprintf(tmp, sizeof(tmp), "%" PRId64, cfg->BackOff[i]);
+            if (i > 0)
+                s = natsBuf_AppendByte(buf, ',');
+            IFOK(s, natsBuf_Append(buf, tmp, -1));
+        }
+        IFOK(s, natsBuf_AppendByte(buf, ']'));
+    }
     IFOK(s, natsBuf_Append(buf, "}}", -1));
 
     if (s == NATS_OK)
@@ -1732,6 +1747,7 @@ _destroyConsumerConfig(jsConsumerConfig *cc)
     NATS_FREE((char*) cc->DeliverGroup);
     NATS_FREE((char*) cc->FilterSubject);
     NATS_FREE((char*) cc->SampleFrequency);
+    NATS_FREE(cc->BackOff);
     NATS_FREE(cc);
 }
 
@@ -1846,6 +1862,7 @@ _unmarshalConsumerConfig(nats_JSON *json, const char *fieldName, jsConsumerConfi
         IFOK(s, nats_JSONGetBool(cjson, "headers_only", &(cc->HeadersOnly)));
         IFOK(s, nats_JSONGetLong(cjson, "max_batch", &(cc->MaxRequestBatch)));
         IFOK(s, nats_JSONGetLong(cjson, "max_expires", &(cc->MaxRequestExpires)));
+        IFOK(s, nats_JSONGetArrayLong(cjson, "backoff", &(cc->BackOff), &(cc->BackOffLen)));
     }
 
     if (s == NATS_OK)

src/nats.h

@@ -611,6 +611,8 @@ typedef struct jsConsumerConfig
         jsAckPolicy             AckPolicy;
         int64_t                 AckWait;
         int64_t                 MaxDeliver;
+        int64_t                 *BackOff;               ///< Redelivery durations expressed in nanoseconds
+        int                     BackOffLen;
         const char              *FilterSubject;
         jsReplayPolicy          ReplayPolicy;
         uint64_t                RateLimit;
@@ -5801,6 +5803,21 @@ natsMsg_AckSync(natsMsg *msg, jsOptions *opts, jsErrCode *errCode);
 NATS_EXTERN natsStatus
 natsMsg_Nak(natsMsg *msg, jsOptions *opts);
 
+/** \brief Negatively acknowledges a message.
+ *
+ * This tells the server to redeliver the message after the given `delay`
+ * duration expressed in milliseconds. You can configure the number of
+ * redeliveries by passing `MaxDeliver` when you subscribe.
+ *
+ * The default is infinite redeliveries.
+ *
+ * @param msg the pointer to the #natsMsg object.
+ * @param delay the amount of time before the redelivery expressed in milliseconds.
+ * @param opts the pointer to the #jsOptions object, possibly `NULL`.
+ */
+NATS_EXTERN natsStatus
+natsMsg_NakWithDelay(natsMsg *msg, int64_t delay, jsOptions *opts);
+
 /** \brief Resets redelivery timer on the server.
  *
  * This tells the server that this message is being worked on. It resets

src/natsp.h

@@ -1,4 +1,4 @@
-// Copyright 2015-2021 The NATS Authors
+// Copyright 2015-2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -133,6 +133,9 @@
 
 #define IFOK(s, c)      if (s == NATS_OK) { s = (c); }
 
+#define NATS_MILLIS_TO_NANOS(d)     (((int64_t)d)*(int64_t)1E6)
+#define NATS_SECONDS_TO_NANOS(d)    (((int64_t)d)*(int64_t)1E9)
+
 extern int64_t gLockSpinCount;
 
 typedef void (*natsInitOnceCb)(void);

src/status.h

@@ -1,4 +1,4 @@
-// Copyright 2015-2021 The NATS Authors
+// Copyright 2015-2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -243,6 +243,11 @@ typedef enum {
     JSStreamSealedErr = 10109,                          ///< Invalid operation on sealed stream
     JSStreamPurgeFailedErr = 10110,                     ///< Generic stream purge failure
     JSStreamRollupFailedErr = 10111,                    ///< Generic stream rollup failure
+    JSConsumerInvalidDeliverSubjectErr = 10112,         ///< Invalid push consumer deliver subject
+    JSStreamMaxBytesRequiredErr = 10113,                ///< Account requires a stream config to have max bytes set
+    JSConsumerMaxRequestBatchNegativeErr = 10114,       ///< Consumer max request batch needs to be > 0
+    JSConsumerMaxRequestExpiresToSmallErr = 10115,      ///< Consumer max request expires needs to be > 1ms
+    JSConsumerMaxDeliverBackoffErr = 10116,             ///< Max deliver is required to be > length of backoff values
 
 } jsErrCode;
 

test/list.txt

@@ -230,6 +230,8 @@ JetStreamOrderedConsAutoUnsub
 JetStreamSubscribeWithFWC
 JetStreamStreamsSealAndRollup
 JetStreamGetMsgAndLastMsg
+JetStreamNakWithDelay
+JetStreamBackOffRedeliveries
 KeyValueManager
 KeyValueBasics
 KeyValueWatch