Function streamDelConsumer

src/t_stream.cpp:2324–2347

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. */

 * 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,SLC_NOCREAT|SLC_NOREFRESH,NULL);
    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 = (streamNACK*)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