Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

KeyValue updates #485

Merged
merged 3 commits into from
Nov 9, 2021
Merged

KeyValue updates #485

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ea73fd5
KeyValue updates
kozlovic Nov 9, 2021
fe2e861
Update docs related to KV
kozlovic Nov 9, 2021
14a19f3
Add test to demonstrate kvWatcher_Next() can be stopped
kozlovic Nov 9, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions src/kv.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -366,12 +366,12 @@ _getEntry(kvEntry **new_entry, bool *deleted, kvStore *kv, const char *key)
kvEntry *e = NULL;
DEFINE_BUF_FOR_SUBJECT;

if (!validKey(key))
return nats_setError(NATS_INVALID_ARG, "%s", kvErrInvalidKey);

*new_entry = NULL;
*deleted = false;

if (!validKey(key))
return nats_setError(NATS_INVALID_ARG, "%s", kvErrInvalidKey);

BUILD_SUBJECT;
IFOK(s, js_GetLastMsg(&msg, kv->js, kv->stream, natsBuf_Data(&buf), NULL, NULL));
IFOK(s, _createEntry(&e, kv, &msg));
Expand Down Expand Up @@ -402,22 +402,20 @@ natsStatus
kvStore_Get(kvEntry **new_entry, kvStore *kv, const char *key)
{
natsStatus s;
kvEntry *e = NULL;
bool deleted = false;

if ((new_entry == NULL) || (kv == NULL))
return nats_setDefaultError(NATS_INVALID_ARG);

s = _getEntry(&e, &deleted, kv, key);
s = _getEntry(new_entry, &deleted, kv, key);
if (s == NATS_OK)
{
if (deleted)
{
kvEntry_Destroy(e);
kvEntry_Destroy(*new_entry);
*new_entry = NULL;
return NATS_NOT_FOUND;
}
// Valid entry, will return it.
*new_entry = e;
}
else if (s == NATS_NOT_FOUND)
return s;
Expand Down Expand Up @@ -1088,6 +1086,12 @@ kvStatus_TTL(kvStatus *sts)
return (sts == NULL || sts->si->Config == NULL ? 0 : sts->si->Config->MaxAge);
}

int64_t
kvStatus_Replicas(kvStatus *sts)
{
return (sts == NULL || sts->si->Config == NULL ? 0 : sts->si->Config->Replicas);
}

void
kvStatus_Destroy(kvStatus *sts)
{
Expand Down
9 changes: 9 additions & 0 deletions src/nats.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6371,6 +6371,15 @@ kvStatus_History(kvStatus *sts);
NATS_EXTERN int64_t
kvStatus_TTL(kvStatus *sts);

/** \brief Returns the number of replicas to keep for a bucket.
*
* Returns the number of replicas to keep for a bucket, or `0` if `sts` itself is `NULL`.
*
* @param sts the pointer to the #kvStatus object.
*/
NATS_EXTERN int64_t
kvStatus_Replicas(kvStatus *sts);

/** \brief Destroys the KeyValue status object.
*
* Releases memory allocated for this #kvStatus object.
Expand Down
21 changes: 17 additions & 4 deletions test/test.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26684,6 +26684,7 @@ test_KeyValueBasics(void)
kvc.Bucket = "TEST";
kvc.History = 5;
kvc.TTL = 3600*(int64_t)1E9;
kvc.Replicas = 1;
s = js_CreateKeyValue(&kv, js, &kvc);
testCond((s == NATS_OK) && (kv != NULL));

Expand Down Expand Up @@ -26899,6 +26900,10 @@ test_KeyValueBasics(void)
s = (kvStatus_Values(sts) == 7 ? NATS_OK : NATS_ERR);
testCond(s == NATS_OK);

test("Check replicas: ");
s = (kvStatus_Replicas(sts) == 1 ? NATS_OK : NATS_ERR);
testCond(s == NATS_OK);

test("Check status with NULL: ");
if ((kvStatus_History(NULL) != 0) || (kvStatus_Bucket(NULL) != NULL)
|| (kvStatus_TTL(NULL) != 0) || (kvStatus_Values(NULL) != 0))
Expand Down Expand Up @@ -26943,9 +26948,11 @@ _expectUpdate(kvWatcher *w, const char *key, const char *val, uint64_t rev)
if ((s != NATS_OK) || (e == NULL))
return false;

if ((strcmp(kvEntry_Key(e), key) != 0)
if ((strcmp(kvEntry_Bucket(e), "WATCH") != 0)
|| (strcmp(kvEntry_Key(e), key) != 0)
|| (strcmp(kvEntry_ValueString(e), val) != 0)
|| (kvEntry_Revision(e) != rev))
|| (kvEntry_Revision(e) != rev)
|| (kvEntry_Created(e) == 0))
{
return false;
}
Expand Down Expand Up @@ -27196,6 +27203,7 @@ test_KeyValueKeys(void)
bool countryOK = false;
char *k = NULL;
kvConfig kvc;
kvWatchOptions o;
int i;

JS_SETUP(2, 6, 2);
Expand Down Expand Up @@ -27227,7 +27235,9 @@ test_KeyValueKeys(void)
nats_clearLastError();

test("Get keys (timeout): ");
s = kvStore_Keys(&l, kv, NULL);
kvWatchOptions_Init(&o);
o.Timeout = 1;
s = kvStore_Keys(&l, kv, &o);
testCond((((s == NATS_OK) && (l.Keys != NULL) && (l.Count == 3)))
|| ((s == NATS_TIMEOUT) && (l.Keys == NULL) && (l.Count == 0)));
nats_clearLastError();
Expand Down Expand Up @@ -27324,7 +27334,10 @@ test_KeyValueDeleteVsPurge(void)
test("Check: ");
for (i=0;(s == NATS_OK) && (i<4); i++)
{
if (l.Entries[i] == NULL)
e = l.Entries[i];
if (e == NULL)
s = NATS_ERR;
else if ((int) kvEntry_Delta(e) != (3-i))
s = NATS_ERR;
}
testCond(s == NATS_OK);
Expand Down