Function defragStreamConsumerPendingEntry

src/defrag.cpp:754–768

} PendingEntryContext;

void* defragStreamConsumerPendingEntry(raxIterator *ri, void *privdata, long *defragged) {
    UNUSED(defragged);
    PendingEntryContext *ctx = (PendingEntryContext*)privdata;
    streamNACK *nack = (streamNACK*)ri->data, *newnack;
    nack->consumer = ctx->c; /* update nack pointer to consumer */
    newnack = (streamNACK*)activeDefragAlloc(nack);
    if (newnack) {
        /* update consumer group pointer to the nack */
        void *prev;
        raxInsert(ctx->cg->pel, ri->key, ri->key_len, newnack, &prev);
        serverAssert(prev==nack);
        /* note: we don't increment 'defragged' that's done by the caller */
    }
    return newnack;
}

void* defragStreamConsumer(raxIterator *ri, void *privdata, long *defragged) {
    streamConsumer *c = (streamConsumer*)ri->data;