fluencelabs/redis
1
0
Fork 0
mirror of https://github.com/fluencelabs/redis synced 2025-03-20 09:30:55 +00:00
Code Issues Projects Releases Wiki Activity

CG: XGROUP DELCONSUMER implemented.

Browse Source
This commit is contained in:
antirez 2018-03-02 17:24:00 +01:00
parent 9b423ae237
commit d4f81ebdba
1 changed files with 46 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
src/t_stream.c
View File

@ -1351,6 +1351,33 @@ streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int create) {
return consumer;
}
/* Delete the consumer specified in the consumer group 'cg'. The consumer
* may have pending messages: they are removed from the PEL, and the number
* of pending messages "lost" is returned. */
uint64_t streamDelConsumer(streamCG *cg, sds name) {
streamConsumer *consumer = streamLookupConsumer(cg,name,0);
if (consumer == NULL) return 0;
uint64_t retval = raxSize(consumer->pel);
/* Iterate all the consumer pending messages, deleting every corresponding
* entry from the global entry. */
raxIterator ri;
raxStart(&ri,consumer->pel);
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
streamNACK *nack = ri.data;
raxRemove(cg->pel,ri.key,ri.key_len,NULL);
streamFreeNACK(nack);
}
raxStop(&ri);
/* Deallocate the consumer. */
raxRemove(cg->consumers,(unsigned char*)name,sdslen(name),NULL);
streamFreeConsumer(consumer);
return retval;
}
/* -----------------------------------------------------------------------
* Consumer groups commands
* ----------------------------------------------------------------------- */
@ -1370,6 +1397,8 @@ NULL
};
stream *s = NULL;
sds grpname = NULL;
streamCG *cg = NULL;
char *opt = c->argv[1]->ptr; /* Subcommand name. */
/* Lookup the key now, this is common for all the subcommands but HELP. */
if (c->argc >= 4) {
@ -1377,9 +1406,21 @@ NULL
if (o == NULL) return;
s = o->ptr;
grpname = c->argv[3]->ptr;
/* Certain subcommands require the group to exist. */
if ((cg = streamLookupCG(s,grpname)) == NULL &&
(!strcasecmp(opt,"SETID") ||
!strcasecmp(opt,"DELGROUP") ||
!strcasecmp(opt,"DELCONSUMER")))
{
addReplyErrorFormat(c, "-NOGROUP No such consumer group '%s' "
"for key name '%s'",
grpname, c->argv[2]->ptr);
return;
}
}
char *opt = c->argv[1]->ptr;
/* Dispatch the different subcommands. */
if (!strcasecmp(opt,"CREATE") && c->argc == 5) {
streamID id;
if (!strcmp(c->argv[4]->ptr,"$")) {
@ -1397,6 +1438,10 @@ NULL
} else if (!strcasecmp(opt,"SETID") && c->argc == 5) {
} else if (!strcasecmp(opt,"DELGROUP") && c->argc == 4) {
} else if (!strcasecmp(opt,"DELCONSUMER") && c->argc == 5) {
/* Delete the consumer and returns the number of pending messages
* that were yet associated with such a consumer. */
long long pending = streamDelConsumer(cg,c->argv[4]->ptr);
addReplyLongLong(c,pending);
} else if (!strcasecmp(opt,"HELP")) {
addReplyHelp(c, help);
} else {