From 10f3f6ec26bd10e9d25a0ebd48cf2a97fd2b44ff Mon Sep 17 00:00:00 2001 From: Lev Brouk Date: Mon, 7 Apr 2025 13:13:34 -0700 Subject: [PATCH 1/4] [ADDED] Per-message TTL support (nats-server v2.11 feature) See https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-43.md and https://github.com/nats-io/nats.go/pull/1825. Also added all of the new JS error status codes from the server. --- src/js.c | 9 +++ src/js.h | 1 + src/jsm.c | 7 +++ src/msg.c | 5 +- src/nats.h | 9 +++ src/status.h | 33 ++++++++++ src/util.c | 41 ++++++++----- src/util.h | 8 +++ test/list_test.txt | 2 + test/test.c | 149 +++++++++++++++++++++++++++++++++++++++++++++ 10 files changed, 249 insertions(+), 15 deletions(-) diff --git a/src/js.c b/src/js.c index d90c99e99..ef9b8f567 100644 --- a/src/js.c +++ b/src/js.c @@ -476,6 +476,15 @@ _setHeadersFromOptions(natsMsg *msg, jsPubOptions *opts) s = natsMsgHeader_Set(msg, jsExpectedLastSubjSeqHdr, temp); } } + if ((s == NATS_OK) && (opts->MsgTTL != 0)) + { + if (opts->MsgTTL < 0) + s = nats_setError(NATS_INVALID_ARG, "option 'MsgTTL' (%" PRId64 ") cannot be negative", opts->MsgTTL); + if (s == NATS_OK) + { + s = natsMsgHeader_Set(msg, jsMsgTTLHdr, nats_formatDurationMilli(temp, opts->MsgTTL)); + } + } return NATS_UPDATE_ERR_STACK(s); } diff --git a/src/js.h b/src/js.h index 5c8fbd4f9..3f0cc8b7f 100644 --- a/src/js.h +++ b/src/js.h @@ -34,6 +34,7 @@ extern const char* jsDefaultAPIPrefix; extern const int64_t jsDefaultRequestWait; #define jsMsgIdHdr "Nats-Msg-Id" +#define jsMsgTTLHdr "Nats-TTL" #define jsExpectedStreamHdr "Nats-Expected-Stream" #define jsExpectedLastSeqHdr "Nats-Expected-Last-Sequence" #define jsExpectedLastSubjSeqHdr "Nats-Expected-Last-Subject-Sequence" diff --git a/src/jsm.c b/src/jsm.c index 6d62687f5..7cc49ac1c 100644 --- a/src/jsm.c +++ b/src/jsm.c @@ -763,6 +763,8 @@ js_unmarshalStreamConfig(nats_JSON *json, const char *fieldName, jsStreamConfig obj = NULL; IFOK(s, nats_JSONGetObject(jcfg, "consumer_limits", &obj)); IFOK(s, _unmarshalStreamConsumerLimits(obj, &(cfg->ConsumerLimits))); + IFOK(s, nats_JSONGetLong(jcfg, "subject_delete_marker_ttl", &(cfg->SubjectDeleteMarkerTTL))); + IFOK(s, nats_JSONGetBool(jcfg, "allow_msg_ttl", &(cfg->AllowMsgTTL))); if (s == NATS_OK) *new_cfg = cfg; @@ -891,6 +893,11 @@ js_marshalStreamConfig(natsBuffer **new_buf, jsStreamConfig *cfg) IFOK(s, _marshalSubjectTransformConfig(&cfg->SubjectTransform, buf)); IFOK(s, _marshalStreamConsumerLimits(&cfg->ConsumerLimits, buf)); + if ((s == NATS_OK) && cfg->AllowMsgTTL) + s = natsBuf_Append(buf, ",\"allow_msg_ttl\":true", -1); + if ((s == NATS_OK) && cfg->SubjectDeleteMarkerTTL > 0) + s = nats_marshalLong(buf, true, "subject_delete_marker_ttl", cfg->SubjectDeleteMarkerTTL); + IFOK(s, natsBuf_AppendByte(buf, '}')); if (s == NATS_OK) diff --git a/src/msg.c b/src/msg.c index 740b4abf5..8d90dd6a2 100644 --- a/src/msg.c +++ b/src/msg.c @@ -213,7 +213,10 @@ _processKeyValue(int line, natsMsg *msg, char *endPtr, char **pPtr, char **lastK start = ptr; if (*ptr == '\r') { - if ((++ptr == endPtr) || ((*ptr == '\n') && (++ptr == endPtr))) + if ((++ptr == endPtr) || + ((*ptr == '\n') && (++ptr == endPtr)) || + ((*ptr == '\r') && (++ptr == endPtr)) || + ((*ptr == '\n') && (++ptr == endPtr))) { *pPtr = ptr; return NATS_OK; diff --git a/src/nats.h b/src/nats.h index 97d62525b..4aebba810 100644 --- a/src/nats.h +++ b/src/nats.h @@ -295,6 +295,7 @@ typedef struct jsPubOptions uint64_t ExpectLastSeq; ///< Expected last message sequence in the stream. uint64_t ExpectLastSubjectSeq; ///< Expected last message sequence for the subject in the stream. bool ExpectNoMessage; ///< Expected no message (that is, sequence == 0) for the subject in the stream. + int64_t MsgTTL; ///< Message time to live (TTL) in milliseconds, used by the server to expire the message. } jsPubOptions; @@ -611,6 +612,14 @@ typedef struct jsStreamConfig { /// @brief Sets the limits on certain options on all consumers of the /// stream. jsStreamConsumerLimits ConsumerLimits; + + /// @brief Allow the message to be sent with a time to live (TTL) value. + bool AllowMsgTTL; + + /// @brief Enables and sets a duration for adding server markers for + /// delete, purge and max age limits. + int64_t SubjectDeleteMarkerTTL; + } jsStreamConfig; /** diff --git a/src/status.h b/src/status.h index e8f90c7ed..47cf33287 100644 --- a/src/status.h +++ b/src/status.h @@ -268,6 +268,39 @@ typedef enum { JSConsumerCreateFilterSubjectMismatchErr = 10131, ///< Consumer create request did not match filtered subject from create subject JSConsumerCreateDurableAndNameMismatchErr = 10132, ///< Consumer Durable and Name have to be equal if both are provided JSReplicasCountCannotBeNegativeErr = 10133, ///< Replicas count cannot be negative + JSConsumerReplicasShouldMatchStreamErr = 10134, ///< Consumer config replicas must match interest retention stream's replicas + JSConsumerMetadataLengthErr = 10135, ///< Consumer metadata exceeds maximum size + JSConsumerDuplicateFilterSubjectsErr = 10136, ///< Consumer cannot have both FilterSubject and FilterSubjects specified + JSConsumerMultipleFiltersNotAllowedErr = 10137, ///< Consumer with multiple subject filters cannot use subject based API + JSConsumerOverlappingSubjectFiltersErr = 10138, ///< Consumer subject filters cannot overlap + JSConsumerEmptyFilterErr = 10139, ///< Consumer filter in FilterSubjects cannot be empty + JSSourceDuplicateDetectedErr = 10140, ///< Source stream, filter and transform (plus external if present) must form a unique combination + JSSourceInvalidStreamNameErr = 10141, ///< Sourced stream name is invalid + JSMirrorInvalidStreamNameErr = 10142, ///< Mirrored stream name is invalid + JSSourceMultipleFiltersNotAllowedErr = 10144, ///< Source with multiple subject transforms cannot also have a single subject filter + JSSourceInvalidSubjectFilterErr = 10145, ///< Source transform source error + JSSourceInvalidTransformDestinationErr = 10146, ///< Source transform error + JSSourceOverlappingSubjectFiltersErr = 10147, ///< Source subject filters cannot overlap + JSConsumerAlreadyExistsErr = 10148, ///< Action CREATE is used for a existing consumer with a different config + JSConsumerDoesNotExistErr = 10149, ///< Action UPDATE is used for a non-existing consumer + JSMirrorMultipleFiltersNotAllowedErr = 10150, ///< Mirror with multiple subject transforms cannot also have a single subject filter + JSMirrorInvalidSubjectFilterErr = 10151, ///< Mirror transform source error + JSMirrorOverlappingSubjectFiltersErr = 10152, ///< Mirror subject filters cannot overlap + JSConsumerInactiveThresholdExcessErr = 10153, ///< Consumer inactive threshold exceeds system limit + JSMirrorInvalidTransformDestinationErr = 10154, ///< Mirror transform error + JSStreamTransformInvalidSourceErr = 10155, ///< Stream transform source error + JSStreamTransformInvalidDestinationErr = 10156, ///< Stream transform destination error + JSPedanticErr = 10157, ///< Pedantic mode error + JSStreamDuplicateMessageConflictErr = 10158, ///< Duplicate message id is in process + JSConsumerPriorityPolicyWithoutGroupErr = 10159, ///< Setting PriorityPolicy requires at least one PriorityGroup to be set + JSConsumerInvalidPriorityGroupErr = 10160, ///< Provided priority group does not exist for this consumer" + JSConsumerEmptyGroupNameErr = 10161, ///< Group name cannot be an empty string + JSConsumerInvalidGroupNameErr = 10162, ///< Valid priority group name must match A-Z, a-z, 0-9, -_/=)+ and may not exceed 16 characters + JSStreamExpectedLastSeqPerSubjectNotReadyErr = 10163, ///< Expected last sequence per subject temporarily unavailable + JSStreamWrongLastSequenceConstantErr = 10164, ///< Wrong last sequence + JSMessageTTLInvalidErr = 10165, ///< Invalid per-message TTL + JSMessageTTLDisabledErr = 10166, ///< Per-message TTL is disabled + JSStreamTooManyRequestsErr = 10167, ///< Too many requests } jsErrCode; diff --git a/src/util.c b/src/util.c index 2b2be6478..24ac60ec1 100644 --- a/src/util.c +++ b/src/util.c @@ -2287,17 +2287,22 @@ fmt_int(char *buf, int w, uint64_t v) return w; } -natsStatus -nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, int64_t d) +// Converts a number of nanoseconds to a go duration string. +// Note: buf must be at least 32 bytes long. +// This is the maximum size of a duration string. +// The format is: "2540400h10m10.000000000s". +// Returns the pointer to the beginning of the string (in buf). +char * +nats_formatDuration(char *buf, int64_t d) { - // Largest time is 2540400h10m10.000000000s - char buf[32]; - int w = 32; bool neg = d < 0; uint64_t u = (uint64_t) (neg ? -d : d); int prec; - natsStatus s = NATS_OK; - const char *start = (comma ? ",\"" : "\""); + + // Largest time is 2540400h10m10.000000000s + int w = 32; + w--; + buf[w] = '\0'; if (u < 1000000000) { @@ -2307,12 +2312,7 @@ nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, in buf[w] = 's'; w--; if (u == 0) - { - s = natsBuf_Append(out_buf, start, -1); - IFOK(s, natsBuf_Append(out_buf, field_name, -1)); - IFOK(s, natsBuf_Append(out_buf, "\":\"0s\"", -1)); - return NATS_UPDATE_ERR_STACK(s); - } + return buf+w; else if (u < 1000) { // print nanoseconds @@ -2373,10 +2373,23 @@ nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, in buf[w] = '-'; } + return buf+w; +} + +natsStatus +nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, int64_t d) +{ + // Largest time is 2540400h10m10.000000000s + char buf[32]; + natsStatus s = NATS_OK; + const char *start = (comma ? ",\"" : "\""); + + char *dur = nats_formatDuration(buf, d); + s = natsBuf_Append(out_buf, start, -1); IFOK(s, natsBuf_Append(out_buf, field_name, -1)); IFOK(s, natsBuf_Append(out_buf, "\":\"", -1)); - IFOK(s, natsBuf_Append(out_buf, buf + w, sizeof(buf) - w)); + IFOK(s, natsBuf_Append(out_buf, dur, -1)); IFOK(s, natsBuf_Append(out_buf, "\"", -1)); return NATS_UPDATE_ERR_STACK(s); } diff --git a/src/util.h b/src/util.h index 17e2d7bd2..abb218a77 100644 --- a/src/util.h +++ b/src/util.h @@ -240,6 +240,14 @@ nats_marshalLong(natsBuffer *buf, bool comma, const char *fieldName, int64_t lva natsStatus nats_marshalULong(natsBuffer *buf, bool comma, const char *fieldName, uint64_t uval); +char * +nats_formatDuration(char *buf, int64_t nano); + +static inline char *nats_formatDurationMilli(char *buf, int64_t milli) +{ + return nats_formatDuration(buf, milli * 1000 * 1000); +} + natsStatus nats_marshalDuration(natsBuffer *out_buf, bool comma, const char *field_name, int64_t d); diff --git a/test/list_test.txt b/test/list_test.txt index 1dfd7b42c..22a9f8eb8 100644 --- a/test/list_test.txt +++ b/test/list_test.txt @@ -90,6 +90,7 @@ _test(JetStreamInfoWithSubjects) _test(JetStreamMarshalStreamConfig) _test(JetStreamMgtConsumers) _test(JetStreamMgtStreams) +_test(JetStreamMsgDeleteMarkerMaxAge) _test(JetStreamNakWithDelay) _test(JetStreamOrderedConsSrvRestart) _test(JetStreamOrderedConsumer) @@ -98,6 +99,7 @@ _test(JetStreamOrderedConsumerWithErrors) _test(JetStreamPublish) _test(JetStreamPublishAckHandler) _test(JetStreamPublishAsync) +_test(JetStreamPublishTTL) _test(JetStreamStreamsSealAndRollup) _test(JetStreamSubscribe) _test(JetStreamSubscribeConfigCheck) diff --git a/test/test.c b/test/test.c index cf386b4e9..a0652d4e7 100644 --- a/test/test.c +++ b/test/test.c @@ -25384,6 +25384,155 @@ void test_JetStreamPublish(void) remove(confFile); } +void test_JetStreamPublishTTL(void) +{ + natsStatus s; + jsStreamConfig cfg; + jsPubOptions opts; + jsErrCode jerr = 0; + jsPubAck *pa = NULL; + natsMsg *msg = NULL; + const char *hvalue = NULL; + + JS_SETUP(2, 11, 0); + + test("Init: "); + s = jsStreamConfig_Init(&cfg); + IFOK(s, jsPubOptions_Init(&opts)); + testCond(s == NATS_OK); + + test("Disabled per-msg TTL - error: "); + cfg.Name = "DISABLED"; + cfg.Subjects = (const char*[1]){"bar"}; + cfg.SubjectsLen = 1; + cfg.AllowMsgTTL = false; + s = js_AddStream(NULL, js, &cfg, NULL, NULL); + if (s == NATS_OK) + { + opts.MsgTTL = 1000; + s = js_Publish(NULL, js, "bar", "hello", 5, &opts, &jerr); + } + testCond((s == NATS_ERR) + && (jerr == JSMessageTTLDisabledErr) + && (strstr(nats_GetLastError(NULL), "per-message TTL is disabled") != NULL)); + + test("Add enabled stream: "); + cfg.Name = "TEST"; + cfg.Subjects = (const char*[1]){"foo"}; + cfg.SubjectsLen = 1; + cfg.AllowMsgTTL = true; + + s = js_AddStream(NULL, js, &cfg, NULL, NULL); + testCond(s == NATS_OK); + + test("Publish message with a negative TTL - error: "); + opts.MsgTTL = -1000; + s = js_Publish(&pa, js, "foo", "hello", 5, &opts, &jerr); + testCond(s == NATS_INVALID_ARG); + + test("Publish message with a TTL too small - error: "); + opts.MsgTTL = 1; + s = js_Publish(&pa, js, "foo", "hello", 5, &opts, &jerr); + testCond((s == NATS_ERR) + && (jerr == JSMessageTTLInvalidErr) + && (strstr(nats_GetLastError(NULL), "invalid per-message TTL") != NULL)); + + test("Publish message with a 1s TTL: "); + opts.MsgTTL = 1000; // the server would not allow less than 1s + s = js_Publish(&pa, js, "foo", "hello", 5, &opts, &jerr); + testCond((s == NATS_OK) && (jerr == 0) && (pa != NULL)); + + test("Get the message back immediately: "); + s = js_GetMsg(&msg, js, "TEST", pa->Sequence, NULL, &jerr); + + testCond((s == NATS_OK) && (jerr == 0) && (msg != NULL) + && (natsMsgHeader_Get(msg, "Nats-TTL", &hvalue) == NATS_OK) + && (hvalue != NULL) + && (strcmp(hvalue, "1s") == 0) + && (strcmp(natsMsg_GetData(msg), "hello") == 0) + ); + natsMsg_Destroy(msg); + msg = NULL; + + test("Publish an Async message with a 1s TTL: "); + s = js_PublishAsync(js, "foo", "goodbye", 7, &opts); + IFOK(s, js_PublishAsyncComplete(js, &opts)); + + test("Get the async (last) message back immediately: "); + s = js_GetLastMsg(&msg, js, "TEST", "foo", NULL, &jerr); + testCond((s == NATS_OK) && (jerr == 0) && (msg != NULL) + && (natsMsgHeader_Get(msg, "Nats-TTL", &hvalue) == NATS_OK) + && (hvalue != NULL) + && (strcmp(hvalue, "1s") == 0) + && (strcmp(natsMsg_GetData(msg), "goodbye") == 0) + ); + natsMsg_Destroy(msg); + msg = NULL; + + test("Sleep 1.5 seconds to exceed the TTL and try again: "); + nats_Sleep(1500); + s = js_GetMsg(&msg, js, "TEST", pa->Sequence, NULL, &jerr); + testCond((s = NATS_NOT_FOUND) && (jerr == JSNoMessageFoundErr)); + + test("Get the async (last) message back immediately: "); + s = js_GetLastMsg(&msg, js, "TEST", "foo", NULL, &jerr); + testCond((s = NATS_NOT_FOUND) && (jerr == JSNoMessageFoundErr)); + + jsPubAck_Destroy(pa); + pa = NULL; + JS_TEARDOWN; +} + +void test_JetStreamMsgDeleteMarkerMaxAge(void) +{ + natsStatus s; + jsStreamConfig cfg; + jsErrCode jerr = 0; + natsMsg *msg = NULL; + const char *hvalue = NULL; + + JS_SETUP(2, 11, 0); + + test("Stream config init: "); + s = jsStreamConfig_Init(&cfg); + testCond(s == NATS_OK); + + test("Add stream: "); + cfg.Name = "TEST"; + cfg.Subjects = (const char*[2]){"foo", "bar"}; + cfg.SubjectsLen = 2; + cfg.AllowMsgTTL = true; + cfg.SubjectDeleteMarkerTTL = NATS_SECONDS_TO_NANOS(50); + cfg.MaxAge = NATS_SECONDS_TO_NANOS(1); + + s = js_AddStream(NULL, js, &cfg, NULL, NULL); + testCond(s == NATS_OK); + + test("Publish message with a TTL: "); + s = js_Publish(NULL, js, "foo", "hello", 5, NULL, &jerr); + testCond((s == NATS_OK) && (jerr == 0)); + + test("Sleep and check last message: "); + nats_Sleep(1500); + s = js_GetLastMsg(&msg, js, "TEST", "foo", NULL, &jerr); + + testCond((s == NATS_OK) + && (jerr == 0) + && (msg != NULL) + && (natsMsg_GetDataLength(msg) == 0) + && (natsMsgHeader_Get(msg, "Nats-Marker-Reason", &hvalue) == NATS_OK) + && (hvalue != NULL) + && (strcmp(hvalue, "MaxAge") == 0) + && (natsMsgHeader_Get(msg, "Nats-TTL", &hvalue) == NATS_OK) + && (hvalue != NULL) + && (strcmp(hvalue, "50s") == 0)); + + natsMsg_Destroy(msg); + + JS_TEARDOWN; +} + + static void _jsPubAckErrHandler(jsCtx *js, jsPubAckErr *pae, void *closure) { From 1247d7cb63ee35e623c0f968bb331d6ecbb2b207 Mon Sep 17 00:00:00 2001 From: Lev Brouk Date: Tue, 8 Apr 2025 06:01:14 -0700 Subject: [PATCH 2/4] Updated (c) dates --- src/js.c | 2 +- src/js.h | 2 +- src/jsm.c | 2 +- src/msg.c | 2 +- src/nats.h | 2 +- src/status.h | 2 +- src/util.c | 2 +- src/util.h | 2 +- test/test.c | 2 +- 9 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/js.c b/src/js.c index ef9b8f567..fcc1f634d 100644 --- a/src/js.c +++ b/src/js.c @@ -1,4 +1,4 @@ -// Copyright 2021-2024 The NATS Authors +// Copyright 2021-2025 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 diff --git a/src/js.h b/src/js.h index 3f0cc8b7f..d5f5d88fc 100644 --- a/src/js.h +++ b/src/js.h @@ -1,4 +1,4 @@ -// Copyright 2021-2024 The NATS Authors +// Copyright 2021-2025 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 diff --git a/src/jsm.c b/src/jsm.c index 7cc49ac1c..112d990f7 100644 --- a/src/jsm.c +++ b/src/jsm.c @@ -1,4 +1,4 @@ -// Copyright 2021-2022 The NATS Authors +// Copyright 2021-2025 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 diff --git a/src/msg.c b/src/msg.c index 8d90dd6a2..0d3ee5b1f 100644 --- a/src/msg.c +++ b/src/msg.c @@ -1,4 +1,4 @@ -// Copyright 2015-2021 The NATS Authors +// Copyright 2015-2025 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 diff --git a/src/nats.h b/src/nats.h index 4aebba810..21eec50f1 100644 --- a/src/nats.h +++ b/src/nats.h @@ -1,4 +1,4 @@ -// Copyright 2015-2024 The NATS Authors +// Copyright 2015-2025 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 diff --git a/src/status.h b/src/status.h index 47cf33287..032298896 100644 --- a/src/status.h +++ b/src/status.h @@ -1,4 +1,4 @@ -// Copyright 2015-2022 The NATS Authors +// Copyright 2015-2025 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 diff --git a/src/util.c b/src/util.c index 24ac60ec1..233f914b8 100644 --- a/src/util.c +++ b/src/util.c @@ -1,4 +1,4 @@ -// Copyright 2015-2023 The NATS Authors +// Copyright 2015-2025 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 diff --git a/src/util.h b/src/util.h index abb218a77..5f26fdce2 100644 --- a/src/util.h +++ b/src/util.h @@ -1,4 +1,4 @@ -// Copyright 2015-2023 The NATS Authors +// Copyright 2015-2025 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 diff --git a/test/test.c b/test/test.c index a0652d4e7..eeea6780f 100644 --- a/test/test.c +++ b/test/test.c @@ -1,4 +1,4 @@ -// Copyright 2015-2024 The NATS Authors +// Copyright 2015-2025 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 From 15a60128d5c3b16915a225649ab9cbdaec8a92a5 Mon Sep 17 00:00:00 2001 From: Lev Brouk Date: Tue, 8 Apr 2025 07:15:04 -0700 Subject: [PATCH 3/4] PR feedback --- src/js.c | 4 +--- src/msg.c | 7 +++---- src/nats.h | 4 ++-- 3 files changed, 6 insertions(+), 9 deletions(-) diff --git a/src/js.c b/src/js.c index fcc1f634d..41256933b 100644 --- a/src/js.c +++ b/src/js.c @@ -480,10 +480,8 @@ _setHeadersFromOptions(natsMsg *msg, jsPubOptions *opts) { if (opts->MsgTTL < 0) s = nats_setError(NATS_INVALID_ARG, "option 'MsgTTL' (%" PRId64 ") cannot be negative", opts->MsgTTL); - if (s == NATS_OK) - { + else s = natsMsgHeader_Set(msg, jsMsgTTLHdr, nats_formatDurationMilli(temp, opts->MsgTTL)); - } } return NATS_UPDATE_ERR_STACK(s); diff --git a/src/msg.c b/src/msg.c index 0d3ee5b1f..b2b0b1015 100644 --- a/src/msg.c +++ b/src/msg.c @@ -213,10 +213,9 @@ _processKeyValue(int line, natsMsg *msg, char *endPtr, char **pPtr, char **lastK start = ptr; if (*ptr == '\r') { - if ((++ptr == endPtr) || - ((*ptr == '\n') && (++ptr == endPtr)) || - ((*ptr == '\r') && (++ptr == endPtr)) || - ((*ptr == '\n') && (++ptr == endPtr))) + if ((++ptr == endPtr) || // case: "\r" + ((*ptr == '\n') && ((++ptr == endPtr) || // case: "\r\n" + ((*ptr == '\r') && (++ptr < endPtr) && (*ptr == '\n') && (++ptr == endPtr))))) // case: "\r\n\r\n" { *pPtr = ptr; return NATS_OK; diff --git a/src/nats.h b/src/nats.h index 21eec50f1..4a72345f3 100644 --- a/src/nats.h +++ b/src/nats.h @@ -556,7 +556,7 @@ typedef struct jsStreamConfig { int64_t MaxConsumers; int64_t MaxMsgs; int64_t MaxBytes; - int64_t MaxAge; + int64_t MaxAge; ///< Max age of messages in nanoseconds. int64_t MaxMsgsPerSubject; int32_t MaxMsgSize; jsDiscardPolicy Discard; @@ -617,7 +617,7 @@ typedef struct jsStreamConfig { bool AllowMsgTTL; /// @brief Enables and sets a duration for adding server markers for - /// delete, purge and max age limits. + /// delete, purge and max age limits. In nanoseconds. int64_t SubjectDeleteMarkerTTL; } jsStreamConfig; From ce905041149e0f09e122ed3113adeb3f119c3f9a Mon Sep 17 00:00:00 2001 From: Lev Brouk Date: Tue, 8 Apr 2025 07:26:55 -0700 Subject: [PATCH 4/4] PR feedback: comments --- src/nats.h | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/nats.h b/src/nats.h index 4a72345f3..0aac0644b 100644 --- a/src/nats.h +++ b/src/nats.h @@ -295,7 +295,7 @@ typedef struct jsPubOptions uint64_t ExpectLastSeq; ///< Expected last message sequence in the stream. uint64_t ExpectLastSubjectSeq; ///< Expected last message sequence for the subject in the stream. bool ExpectNoMessage; ///< Expected no message (that is, sequence == 0) for the subject in the stream. - int64_t MsgTTL; ///< Message time to live (TTL) in milliseconds, used by the server to expire the message. + int64_t MsgTTL; ///< Message time to live (TTL) in milliseconds, used by the server to expire the message. Requires nats-server v2.11.0 or later. } jsPubOptions; @@ -614,10 +614,12 @@ typedef struct jsStreamConfig { jsStreamConsumerLimits ConsumerLimits; /// @brief Allow the message to be sent with a time to live (TTL) value. + /// Requires nats-server v2.11.0 or later. bool AllowMsgTTL; /// @brief Enables and sets a duration for adding server markers for - /// delete, purge and max age limits. In nanoseconds. + /// delete, purge and max age limits. In nanoseconds. Requires + /// nats-server v2.11.0 or later. int64_t SubjectDeleteMarkerTTL; } jsStreamConfig;