Function streamLookupConsumer

src/t_stream.cpp:2301–2319

Lookup the consumer with the specified name in the group 'cg': if the * consumer does not exist it is created unless SLC_NOCREAT flag was specified. * Its last seen time is updated unless SLC_NOREFRESH flag was specified. */

 * consumer does not exist it is created unless SLC_NOCREAT flag was specified.
 * Its last seen time is updated unless SLC_NOREFRESH flag was specified. */
streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int flags, int *created) {
    if (created) *created = 0;
    int create = !(flags & SLC_NOCREAT);
    int refresh = !(flags & SLC_NOREFRESH);
    streamConsumer *consumer = (streamConsumer*)raxFind(cg->consumers,(unsigned char*)name,
                               sdslen(name));
    if (consumer == raxNotFound) {
        if (!create) return NULL;
        consumer = (streamConsumer*)zmalloc(sizeof(*consumer), MALLOC_SHARED);
        consumer->name = sdsdup(name);
        consumer->pel = raxNew();
        raxInsert(cg->consumers,(unsigned char*)name,sdslen(name),
                  consumer,NULL);
        consumer->seen_time = mstime();
        if (created) *created = 1;
    } else if (refresh)
        consumer->seen_time = mstime();
    return consumer;
}

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