Function streamPropagateConsumerCreation

src/t_stream.cpp:1434–1448

We need this when we want to propagate creation of consumer that was created * by XREADGROUP with the NOACK option. In that case, the only way to create * the consumer at the replica is by using XGROUP CREATECONSUMER (see issue #7140) * * XGROUP CREATECONSUMER */

 *  XGROUP CREATECONSUMER <key> <groupname> <consumername>
 */
void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds consumername) {
    robj *argv[5];
    argv[0] = shared.xgroup;
    argv[1] = shared.createconsumer;
    argv[2] = key;
    argv[3] = groupname;
    argv[4] = createObject(OBJ_STRING,sdsdup(consumername));

    /* We use progagate() because this code path is not always called from
     * the command execution context. Moreover this will just alter the
     * consumer group state, and we don't need MULTI/EXEC wrapping because
     * there is no message state cross-message atomicity required. */
    propagate(cserver.xgroupCommand,c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL);
    decrRefCount(argv[4]);
}

/* Send the stream items in the specified range to the client 'c'. The range
 * the client will receive is between start and end inclusive, if 'count' is