Function streamDup

src/t_stream.cpp:151–248

This is a helper function for the COPY command. * Duplicate a Stream object, with the guarantee that the returned object * has the same encoding as the original one. * * The resulting object always has refcount set to 1 */

 *
 * The resulting object always has refcount set to 1 */
robj *streamDup(robj *o) {
    robj *sobj;

    serverAssert(o->type == OBJ_STREAM);

    switch (o->encoding) {
        case OBJ_ENCODING_STREAM:
            sobj = createStreamObject();
            break;
        default:
            serverPanic("Wrong encoding.");
            break;
    }

    stream *s;
    stream *new_s;
    s = (stream*)ptrFromObj(o);
    new_s = (stream*)ptrFromObj(sobj);

    raxIterator ri;
    uint64_t rax_key[2];
    raxStart(&ri, s->rax);
    raxSeek(&ri, "^", NULL, 0);
    size_t lp_bytes = 0;      /* Total bytes in the listpack. */
    unsigned char *lp = NULL; /* listpack pointer. */
    /* Get a reference to the listpack node. */
    while (raxNext(&ri)) {
        lp = (unsigned char*)ri.data;
        lp_bytes = lpBytes(lp);
        unsigned char *new_lp = (unsigned char*)zmalloc(lp_bytes);
        memcpy(new_lp, lp, lp_bytes);
        memcpy(rax_key, ri.key, sizeof(rax_key));
        raxInsert(new_s->rax, (unsigned char *)&rax_key, sizeof(rax_key),
                  new_lp, NULL);
    }
    new_s->length = s->length;
    new_s->last_id = s->last_id;
    raxStop(&ri);

    if (s->cgroups == NULL) return sobj;

    /* Consumer Groups */
    raxIterator ri_cgroups;
    raxStart(&ri_cgroups, s->cgroups);
    raxSeek(&ri_cgroups, "^", NULL, 0);
    while (raxNext(&ri_cgroups)) {
        streamCG *cg = (streamCG*)ri_cgroups.data;
        streamCG *new_cg = streamCreateCG(new_s, (char *)ri_cgroups.key,
                                          ri_cgroups.key_len, &cg->last_id);

        serverAssert(new_cg != NULL);

        /* Consumer Group PEL */
        raxIterator ri_cg_pel;
        raxStart(&ri_cg_pel,cg->pel);
        raxSeek(&ri_cg_pel,"^",NULL,0);
        while(raxNext(&ri_cg_pel)){
            streamNACK *nack = (streamNACK*)ri_cg_pel.data;