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

Function streamDelConsumer

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

Delete the consumer specified in the consumer group 'cg'. The consumer * may have pending messages: they are removed from the PEL, and the number * of pending messages "lost" is returned. */

Source from the content-addressed store, hash-verified

2320 * may have pending messages: they are removed from the PEL, and the number
2321 * of pending messages "lost" is returned. */
2322uint64_t streamDelConsumer(streamCG *cg, sds name) {
2323 streamConsumer *consumer =
2324 streamLookupConsumer(cg,name,SLC_NOCREAT|SLC_NOREFRESH,NULL);
2325 if (consumer == NULL) return 0;
2326
2327 uint64_t retval = raxSize(consumer->pel);
2328
2329 /* Iterate all the consumer pending messages, deleting every corresponding
2330 * entry from the global entry. */
2331 raxIterator ri;
2332 raxStart(&ri,consumer->pel);
2333 raxSeek(&ri,"^",NULL,0);
2334 while(raxNext(&ri)) {
2335 streamNACK *nack = ri.data;
2336 raxRemove(cg->pel,ri.key,ri.key_len,NULL);
2337 streamFreeNACK(nack);
2338 }
2339 raxStop(&ri);
2340
2341 /* Deallocate the consumer. */
2342 raxRemove(cg->consumers,(unsigned char*)name,sdslen(name),NULL);
2343 streamFreeConsumer(consumer);
2344 return retval;
2345}
2346
2347/* -----------------------------------------------------------------------
2348 * Consumer groups commands

Callers 1

xgroupCommandFunction · 0.85

Calls 10

streamLookupConsumerFunction · 0.85
raxSizeFunction · 0.85
raxStartFunction · 0.85
raxSeekFunction · 0.85
raxNextFunction · 0.85
raxRemoveFunction · 0.85
streamFreeNACKFunction · 0.85
raxStopFunction · 0.85
sdslenFunction · 0.85
streamFreeConsumerFunction · 0.85

Tested by

no test coverage detected