Function xgroupCommand

src/t_stream.cpp:2358–2504

XGROUP CREATE [MKSTREAM] * XGROUP SETID * XGROUP DESTROY * XGROUP CREATECONSUMER * XGROUP DELCONSUMER */

 * XGROUP CREATECONSUMER <key> <groupname> <consumer>
 * XGROUP DELCONSUMER <key> <groupname> <consumername> */
void xgroupCommand(client *c) {
    stream *s = NULL;
    sds grpname = NULL;
    streamCG *cg = NULL;
    char *opt = szFromObj(c->argv[1]); /* Subcommand name. */
    int mkstream = 0;
    robj *o;

    /* CREATE has an MKSTREAM option that creates the stream if it
     * does not exist. */
    if (c->argc == 6 && !strcasecmp(opt,"CREATE")) {
        if (strcasecmp(szFromObj(c->argv[5]),"MKSTREAM")) {
            addReplySubcommandSyntaxError(c);
            return;
        }
        mkstream = 1;
        grpname = szFromObj(c->argv[3]);
    }

    /* Everything but the "HELP" option requires a key and group name. */
    if (c->argc >= 4) {
        o = lookupKeyWrite(c->db,c->argv[2]);
        if (o) {
            if (checkType(c,o,OBJ_STREAM)) return;
            s = (stream*)ptrFromObj(o);
        }
        grpname = szFromObj(c->argv[3]);
    }

    /* Check for missing key/group. */
    if (c->argc >= 4 && !mkstream) {
        /* At this point key must exist, or there is an error. */
        if (s == NULL) {
            addReplyError(c,
                "The XGROUP subcommand requires the key to exist. "
                "Note that for CREATE you may want to use the MKSTREAM "
                "option to create an empty stream automatically.");
            return;
        }

        /* Certain subcommands require the group to exist. */
        if ((cg = streamLookupCG(s,grpname)) == NULL &&
            (!strcasecmp(opt,"SETID") ||
             !strcasecmp(opt,"CREATECONSUMER") ||
             !strcasecmp(opt,"DELCONSUMER")))
        {
            addReplyErrorFormat(c, "-NOGROUP No such consumer group '%s' "
                                   "for key name '%s'",
                                   (char*)grpname, (char*)ptrFromObj(c->argv[2]));
            return;
        }
    }

    /* Dispatch the different subcommands. */
    if (c->argc == 2 && !strcasecmp(opt,"HELP")) {
        const char *help[] = {
"CREATE <key> <groupname> <id|$> [option]",
"    Create a new consumer group. Options are:",