MCPcopy Create free account
hub / github.com/F-Stack/f-stack / streamPropagateConsumerCreation

Function streamPropagateConsumerCreation

app/redis-6.2.6/src/t_stream.c:1433–1447  ·  view source on GitHub ↗

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

Source from the content-addressed store, hash-verified

1431 * XGROUP CREATECONSUMER <key> <groupname> <consumername>
1432 */
1433void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds consumername) {
1434 robj *argv[5];
1435 argv[0] = shared.xgroup;
1436 argv[1] = shared.createconsumer;
1437 argv[2] = key;
1438 argv[3] = groupname;
1439 argv[4] = createObject(OBJ_STRING,sdsdup(consumername));
1440
1441 /* We use progagate() because this code path is not always called from
1442 * the command execution context. Moreover this will just alter the
1443 * consumer group state, and we don't need MULTI/EXEC wrapping because
1444 * there is no message state cross-message atomicity required. */
1445 propagate(server.xgroupCommand,c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL);
1446 decrRefCount(argv[4]);
1447}
1448
1449/* Send the stream items in the specified range to the client 'c'. The range
1450 * the client will receive is between start and end inclusive, if 'count' is

Callers 2

xreadCommandFunction · 0.85

Calls 4

createObjectFunction · 0.85
sdsdupFunction · 0.85
propagateFunction · 0.85
decrRefCountFunction · 0.85

Tested by

no test coverage detected