Function RM_StreamAdd

app/redis-6.2.6/src/module.c:3324–3381

Adds an entry to a stream. Like XADD without trimming. * * - `key`: The key where the stream is (or will be) stored * - `flags`: A bit field of * - `REDISMODULE_STREAM_ADD_AUTOID`: Assign a stream ID automatically, like * `*` in the XADD command. * - `id`: If the `AUTOID` flag is set, this is where the assigned ID is * returned. Can be NULL if `AUTOID` is set, if you don't care to r

 * - ERANGE if the elements are too large to be stored.
 */
int RM_StreamAdd(RedisModuleKey *key, int flags, RedisModuleStreamID *id, RedisModuleString **argv, long numfields) {
    /* Validate args */
    if (!key || (numfields != 0 && !argv) || /* invalid key or argv */
        (flags & ~(REDISMODULE_STREAM_ADD_AUTOID)) || /* invalid flags */
        (!(flags & REDISMODULE_STREAM_ADD_AUTOID) && !id)) { /* id required */
        errno = EINVAL;
        return REDISMODULE_ERR;
    } else if (key->value && key->value->type != OBJ_STREAM) {
        errno = ENOTSUP; /* wrong type */
        return REDISMODULE_ERR;
    } else if (!(key->mode & REDISMODULE_WRITE)) {
        errno = EBADF; /* key not open for writing */
        return REDISMODULE_ERR;
    } else if (!(flags & REDISMODULE_STREAM_ADD_AUTOID) &&
               id->ms == 0 && id->seq == 0) {
        errno = EDOM; /* ID out of range */
        return REDISMODULE_ERR;
    }

    /* Create key if necessery */
    int created = 0;
    if (key->value == NULL) {
        moduleCreateEmptyKey(key, REDISMODULE_KEYTYPE_STREAM);
        created = 1;
    }

    stream *s = key->value->ptr;
    if (s->last_id.ms == UINT64_MAX && s->last_id.seq == UINT64_MAX) {
        /* The stream has reached the last possible ID */
        errno = EFBIG;
        return REDISMODULE_ERR;
    }

    streamID added_id;
    streamID use_id;
    streamID *use_id_ptr = NULL;
    if (!(flags & REDISMODULE_STREAM_ADD_AUTOID)) {
        use_id.ms = id->ms;
        use_id.seq = id->seq;
        use_id_ptr = &use_id;
    }
    if (streamAppendItem(s, argv, numfields, &added_id, use_id_ptr) == C_ERR) {
        /* Either the ID not greater than all existing IDs in the stream, or
         * the elements are too large to be stored. either way, errno is already
         * set by streamAppendItem. */
        return REDISMODULE_ERR;
    }
    /* Postponed signalKeyAsReady(). Done implicitly by moduleCreateEmptyKey()
     * so not needed if the stream has just been created. */
    if (!created) key->u.stream.signalready = 1;

    if (id != NULL) {
        id->ms = added_id.ms;
        id->seq = added_id.seq;
    }

    return REDISMODULE_OK;
}