Function xinfoCommand

src/t_stream.cpp:3490–3583

XINFO CONSUMERS * XINFO GROUPS * XINFO STREAM [FULL [COUNT ]] * XINFO HELP. */

 * XINFO STREAM <key> [FULL [COUNT <count>]]
 * XINFO HELP. */
void xinfoCommand(client *c) {
    stream *s = NULL;
    char *opt;
    robj *key;

    /* HELP is special. Handle it ASAP. */
    if (!strcasecmp(szFromObj(c->argv[1]),"HELP")) {
        const char *help[] = {
"CONSUMERS <key> <groupname>",
"    Show consumers of <groupname>.",
"GROUPS <key>",
"    Show the stream consumer groups.",
"STREAM <key> [FULL [COUNT <count>]",
"    Show information about the stream.",
NULL
        };
        addReplyHelp(c, help);
        return;
    } else if (c->argc < 3) {
        addReplySubcommandSyntaxError(c);
        return;
    }

    /* With the exception of HELP handled before any other sub commands, all
     * the ones are in the form of "<subcommand> <key>". */
    opt = szFromObj(c->argv[1]);
    key = c->argv[2];

    /* Lookup the key now, this is common for all the subcommands but HELP. */
    robj_roptr o = lookupKeyReadOrReply(c,key,shared.nokeyerr);
    if (o == nullptr || checkType(c,o,OBJ_STREAM)) return;
    s = (stream*)ptrFromObj(o);

    /* Dispatch the different subcommands. */
    if (!strcasecmp(opt,"CONSUMERS") && c->argc == 4) {
        /* XINFO CONSUMERS <key> <group>. */
        streamCG *cg = streamLookupCG(s,szFromObj(c->argv[3]));
        if (cg == NULL) {
            addReplyErrorFormat(c, "-NOGROUP No such consumer group '%s' "
                                   "for key name '%s'",
                                   (char*)ptrFromObj(c->argv[3]), (char*)ptrFromObj(key));
            return;
        }

        addReplyArrayLen(c,raxSize(cg->consumers));
        raxIterator ri;
        raxStart(&ri,cg->consumers);
        raxSeek(&ri,"^",NULL,0);
        mstime_t now = mstime();
        while(raxNext(&ri)) {
            streamConsumer *consumer = (streamConsumer*)ri.data;
            mstime_t idle = now - consumer->seen_time;
            if (idle < 0) idle = 0;

            addReplyMapLen(c,3);
            addReplyBulkCString(c,"name");
            addReplyBulkCBuffer(c,consumer->name,sdslen(consumer->name));
            addReplyBulkCString(c,"pending");