Function xpendingCommand

src/t_stream.cpp:2604–2780

XPENDING [[IDLE ] [ ]] * * If start and stop are omitted, the command just outputs information about * the amount of pending messages for the key/group pair, together with * the minimum and maximum ID of pending messages. * * If start and stop are provided instead, the pending messages are returned * with information about the current owne

 * with information about the current owner, number of deliveries and last
 * delivery time and so forth. */
void xpendingCommand(client *c) {
    int justinfo = c->argc == 3; /* Without the range just outputs general
                                    informations about the PEL. */
    robj *key = c->argv[1];
    robj *groupname = c->argv[2];
    robj *consumername = NULL;
    streamID startid, endid;
    long long count = 0;
    long long minidle = 0;
    int startex = 0, endex = 0;

    /* Start and stop, and the consumer, can be omitted. Also the IDLE modifier. */
    if (c->argc != 3 && (c->argc < 6 || c->argc > 9)) {
        addReplyErrorObject(c,shared.syntaxerr);
        return;
    }

    /* Parse start/end/count arguments ASAP if needed, in order to report
     * syntax errors before any other error. */
    if (c->argc >= 6) {
        int startidx = 3; /* Without IDLE */

        if (!strcasecmp(szFromObj(c->argv[3]), "IDLE")) {
            if (getLongLongFromObjectOrReply(c, c->argv[4], &minidle, NULL) == C_ERR)
                return;
            if (c->argc < 8) {
                /* If IDLE was provided we must have at least 'start end count' */
                addReplyErrorObject(c,shared.syntaxerr);
                return;
            }
            /* Search for rest of arguments after 'IDLE <idle>' */
            startidx += 2;
        }

        /* count argument. */
        if (getLongLongFromObjectOrReply(c,c->argv[startidx+2],&count,NULL) == C_ERR)
            return;
        if (count < 0) count = 0;

        /* start and end arguments. */
        if (streamParseIntervalIDOrReply(c,c->argv[startidx],&startid,&startex,0) != C_OK)
            return;
        if (startex && streamIncrID(&startid) != C_OK) {
            addReplyError(c,"invalid start ID for the interval");
            return;
        }
        if (streamParseIntervalIDOrReply(c,c->argv[startidx+1],&endid,&endex,UINT64_MAX) != C_OK)
            return;
        if (endex && streamDecrID(&endid) != C_OK) {
            addReplyError(c,"invalid end ID for the interval");
            return;
        }

        if (startidx+3 < c->argc) {
            /* 'consumer' was provided */
            consumername = c->argv[startidx+3];
        }
    }