Function defragStreamConsumer

src/defrag.cpp:770–786

}

void* defragStreamConsumer(raxIterator *ri, void *privdata, long *defragged) {
    streamConsumer *c = (streamConsumer*)ri->data;
    streamCG *cg = (streamCG*)privdata;
    void *newc = activeDefragAlloc(c);
    if (newc) {
        /* note: we don't increment 'defragged' that's done by the caller */
        c = (streamConsumer*)newc;
    }
    sds newsds = activeDefragSds(c->name);
    if (newsds)
        (*defragged)++, c->name = newsds;
    if (c->pel) {
        PendingEntryContext pel_ctx = {cg, c};
        *defragged += defragRadixTree(&c->pel, 0, defragStreamConsumerPendingEntry, &pel_ctx);
    }
    return newc; /* returns NULL if c was not defragged */
}

void* defragStreamConsumerGroup(raxIterator *ri, void *privdata, long *defragged) {
    streamCG *cg = (streamCG*)ri->data;