Unverified Commit 4355145a authored by Viktor Söderqvist's avatar Viktor Söderqvist Committed by GitHub
Browse files

Add modules API for streams (#8288)

APIs added for these stream operations: add, delete, iterate and
trim (by ID or maxlength). The functions are prefixed by RM_Stream.

* RM_StreamAdd
* RM_StreamDelete
* RM_StreamIteratorStart
* RM_StreamIteratorStop
* RM_StreamIteratorNextID
* RM_StreamIteratorNextField
* RM_StreamIteratorDelete
* RM_StreamTrimByLength
* RM_StreamTrimByID

The type RedisModuleStreamID is added and functions for converting
from and to RedisModuleString.

* RM_CreateStringFromStreamID
* RM_StringToStreamID

Whenever the stream functions return REDISMODULE_ERR, errno is set to
provide additional error information.

Refactoring: The zset iterator fields in the RedisModuleKey struct
are wrapped in a union, to allow the same space to be used for type-
specific info for streams and allow future use for other key types.
parent bb7cd974
...@@ -31,4 +31,5 @@ $TCLSH tests/test_helper.tcl \ ...@@ -31,4 +31,5 @@ $TCLSH tests/test_helper.tcl \
--single unit/moduleapi/getkeys \ --single unit/moduleapi/getkeys \
--single unit/moduleapi/test_lazyfree \ --single unit/moduleapi/test_lazyfree \
--single unit/moduleapi/defrag \ --single unit/moduleapi/defrag \
--single unit/moduleapi/stream \
"${@}" "${@}"
This diff is collapsed.
...@@ -69,6 +69,20 @@ ...@@ -69,6 +69,20 @@
#define REDISMODULE_HASH_CFIELDS (1<<2) #define REDISMODULE_HASH_CFIELDS (1<<2)
#define REDISMODULE_HASH_EXISTS (1<<3) #define REDISMODULE_HASH_EXISTS (1<<3)
/* StreamID type. */
typedef struct RedisModuleStreamID {
uint64_t ms;
uint64_t seq;
} RedisModuleStreamID;
/* StreamAdd() flags. */
#define REDISMODULE_STREAM_ADD_AUTOID (1<<0)
/* StreamIteratorStart() flags. */
#define REDISMODULE_STREAM_ITERATOR_EXCLUSIVE (1<<0)
#define REDISMODULE_STREAM_ITERATOR_REVERSE (1<<1)
/* StreamIteratorTrim*() flags. */
#define REDISMODULE_STREAM_TRIM_APPROX (1<<0)
/* Context Flags: Info about the current context returned by /* Context Flags: Info about the current context returned by
* RM_GetContextFlags(). */ * RM_GetContextFlags(). */
...@@ -578,6 +592,7 @@ REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongLong)(Redi ...@@ -578,6 +592,7 @@ REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongLong)(Redi
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromDouble)(RedisModuleCtx *ctx, double d) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromDouble)(RedisModuleCtx *ctx, double d) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongDouble)(RedisModuleCtx *ctx, long double ld, int humanfriendly) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongDouble)(RedisModuleCtx *ctx, long double ld, int humanfriendly) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromStreamID)(RedisModuleCtx *ctx, const RedisModuleStreamID *id) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringPrintf)(RedisModuleCtx *ctx, const char *fmt, ...) REDISMODULE_ATTR_PRINTF(2,3) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringPrintf)(RedisModuleCtx *ctx, const char *fmt, ...) REDISMODULE_ATTR_PRINTF(2,3) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_FreeString)(RedisModuleCtx *ctx, RedisModuleString *str) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_FreeString)(RedisModuleCtx *ctx, RedisModuleString *str) REDISMODULE_ATTR;
REDISMODULE_API const char * (*RedisModule_StringPtrLen)(const RedisModuleString *str, size_t *len) REDISMODULE_ATTR; REDISMODULE_API const char * (*RedisModule_StringPtrLen)(const RedisModuleString *str, size_t *len) REDISMODULE_ATTR;
...@@ -599,6 +614,7 @@ REDISMODULE_API int (*RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, Redis ...@@ -599,6 +614,7 @@ REDISMODULE_API int (*RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, Redis
REDISMODULE_API int (*RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToDouble)(const RedisModuleString *str, double *d) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringToDouble)(const RedisModuleString *str, double *d) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToLongDouble)(const RedisModuleString *str, long double *d) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringToLongDouble)(const RedisModuleString *str, long double *d) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToStreamID)(const RedisModuleString *str, RedisModuleStreamID *id) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_AutoMemory)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_AutoMemory)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_Replicate)(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_Replicate)(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_ReplicateVerbatim)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_ReplicateVerbatim)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
...@@ -629,6 +645,15 @@ REDISMODULE_API int (*RedisModule_ZsetRangePrev)(RedisModuleKey *key) REDISMODUL ...@@ -629,6 +645,15 @@ REDISMODULE_API int (*RedisModule_ZsetRangePrev)(RedisModuleKey *key) REDISMODUL
REDISMODULE_API int (*RedisModule_ZsetRangeEndReached)(RedisModuleKey *key) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_ZsetRangeEndReached)(RedisModuleKey *key) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_HashSet)(RedisModuleKey *key, int flags, ...) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_HashSet)(RedisModuleKey *key, int flags, ...) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_HashGet)(RedisModuleKey *key, int flags, ...) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_HashGet)(RedisModuleKey *key, int flags, ...) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StreamAdd)(RedisModuleKey *key, int flags, RedisModuleStreamID *id, RedisModuleString **argv, int64_t numfields) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StreamDelete)(RedisModuleKey *key, RedisModuleStreamID *id) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StreamIteratorStart)(RedisModuleKey *key, int flags, RedisModuleStreamID *startid, RedisModuleStreamID *endid) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StreamIteratorStop)(RedisModuleKey *key) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StreamIteratorNextID)(RedisModuleKey *key, RedisModuleStreamID *id, long *numfields) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StreamIteratorNextField)(RedisModuleKey *key, RedisModuleString **field_ptr, RedisModuleString **value_ptr) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StreamIteratorDelete)(RedisModuleKey *key) REDISMODULE_ATTR;
REDISMODULE_API long long (*RedisModule_StreamTrimByLength)(RedisModuleKey *key, int flags, long long length) REDISMODULE_ATTR;
REDISMODULE_API long long (*RedisModule_StreamTrimByID)(RedisModuleKey *key, int flags, RedisModuleStreamID *id) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_IsKeysPositionRequest)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_IsKeysPositionRequest)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_KeyAtPos)(RedisModuleCtx *ctx, int pos) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_KeyAtPos)(RedisModuleCtx *ctx, int pos) REDISMODULE_ATTR;
REDISMODULE_API unsigned long long (*RedisModule_GetClientId)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API unsigned long long (*RedisModule_GetClientId)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
...@@ -842,6 +867,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -842,6 +867,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(StringToLongLong); REDISMODULE_GET_API(StringToLongLong);
REDISMODULE_GET_API(StringToDouble); REDISMODULE_GET_API(StringToDouble);
REDISMODULE_GET_API(StringToLongDouble); REDISMODULE_GET_API(StringToLongDouble);
REDISMODULE_GET_API(StringToStreamID);
REDISMODULE_GET_API(Call); REDISMODULE_GET_API(Call);
REDISMODULE_GET_API(CallReplyProto); REDISMODULE_GET_API(CallReplyProto);
REDISMODULE_GET_API(FreeCallReply); REDISMODULE_GET_API(FreeCallReply);
...@@ -856,6 +882,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -856,6 +882,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(CreateStringFromDouble); REDISMODULE_GET_API(CreateStringFromDouble);
REDISMODULE_GET_API(CreateStringFromLongDouble); REDISMODULE_GET_API(CreateStringFromLongDouble);
REDISMODULE_GET_API(CreateStringFromString); REDISMODULE_GET_API(CreateStringFromString);
REDISMODULE_GET_API(CreateStringFromStreamID);
REDISMODULE_GET_API(CreateStringPrintf); REDISMODULE_GET_API(CreateStringPrintf);
REDISMODULE_GET_API(FreeString); REDISMODULE_GET_API(FreeString);
REDISMODULE_GET_API(StringPtrLen); REDISMODULE_GET_API(StringPtrLen);
...@@ -887,6 +914,15 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -887,6 +914,15 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ZsetRangeEndReached); REDISMODULE_GET_API(ZsetRangeEndReached);
REDISMODULE_GET_API(HashSet); REDISMODULE_GET_API(HashSet);
REDISMODULE_GET_API(HashGet); REDISMODULE_GET_API(HashGet);
REDISMODULE_GET_API(StreamAdd);
REDISMODULE_GET_API(StreamDelete);
REDISMODULE_GET_API(StreamIteratorStart);
REDISMODULE_GET_API(StreamIteratorStop);
REDISMODULE_GET_API(StreamIteratorNextID);
REDISMODULE_GET_API(StreamIteratorNextField);
REDISMODULE_GET_API(StreamIteratorDelete);
REDISMODULE_GET_API(StreamTrimByLength);
REDISMODULE_GET_API(StreamTrimByID);
REDISMODULE_GET_API(IsKeysPositionRequest); REDISMODULE_GET_API(IsKeysPositionRequest);
REDISMODULE_GET_API(KeyAtPos); REDISMODULE_GET_API(KeyAtPos);
REDISMODULE_GET_API(GetClientId); REDISMODULE_GET_API(GetClientId);
......
...@@ -108,6 +108,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end ...@@ -108,6 +108,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev); void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev);
int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields); int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields);
void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsigned char **valueptr, int64_t *fieldlen, int64_t *valuelen); void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsigned char **valueptr, int64_t *fieldlen, int64_t *valuelen);
void streamIteratorRemoveEntry(streamIterator *si, streamID *current);
void streamIteratorStop(streamIterator *si); void streamIteratorStop(streamIterator *si);
streamCG *streamLookupCG(stream *s, sds groupname); streamCG *streamLookupCG(stream *s, sds groupname);
streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int flags, int *created); streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int flags, int *created);
...@@ -121,5 +122,11 @@ int streamDecrID(streamID *id); ...@@ -121,5 +122,11 @@ int streamDecrID(streamID *id);
void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds consumername); void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds consumername);
robj *streamDup(robj *o); robj *streamDup(robj *o);
int streamValidateListpackIntegrity(unsigned char *lp, size_t size, int deep); int streamValidateListpackIntegrity(unsigned char *lp, size_t size, int deep);
int streamParseID(const robj *o, streamID *id);
robj *createObjectFromStreamID(streamID *id);
int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id);
int streamDeleteItem(stream *s, streamID *id);
int64_t streamTrimByLength(stream *s, long long maxlen, int approx);
int64_t streamTrimByID(stream *s, streamID minid, int approx);
#endif #endif
...@@ -818,6 +818,28 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) { ...@@ -818,6 +818,28 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
return deleted; return deleted;
} }
/* Trims a stream by length. Returns the number of deleted items. */
int64_t streamTrimByLength(stream *s, long long maxlen, int approx) {
streamAddTrimArgs args = {
.trim_strategy = TRIM_STRATEGY_MAXLEN,
.approx_trim = approx,
.limit = approx ? 100 * server.stream_node_max_entries : 0,
.maxlen = maxlen
};
return streamTrim(s, &args);
}
/* Trims a stream by minimum ID. Returns the number of deleted items. */
int64_t streamTrimByID(stream *s, streamID minid, int approx) {
streamAddTrimArgs args = {
.trim_strategy = TRIM_STRATEGY_MINID,
.approx_trim = approx,
.limit = approx ? 100 * server.stream_node_max_entries : 0,
.minid = minid
};
return streamTrim(s, &args);
}
/* Parse the arguements of XADD/XTRIM. /* Parse the arguements of XADD/XTRIM.
* *
* See streamAddTrimArgs for more details about the arguments handled. * See streamAddTrimArgs for more details about the arguments handled.
...@@ -1625,7 +1647,7 @@ robj *streamTypeLookupWriteOrCreate(client *c, robj *key, int no_create) { ...@@ -1625,7 +1647,7 @@ robj *streamTypeLookupWriteOrCreate(client *c, robj *key, int no_create) {
* treated as an invalid ID. * treated as an invalid ID.
* *
* If 'c' is set to NULL, no reply is sent to the client. */ * If 'c' is set to NULL, no reply is sent to the client. */
int streamGenericParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int strict) { int streamGenericParseIDOrReply(client *c, const robj *o, streamID *id, uint64_t missing_seq, int strict) {
char buf[128]; char buf[128];
if (sdslen(o->ptr) > sizeof(buf)-1) goto invalid; if (sdslen(o->ptr) > sizeof(buf)-1) goto invalid;
memcpy(buf,o->ptr,sdslen(o->ptr)+1); memcpy(buf,o->ptr,sdslen(o->ptr)+1);
...@@ -1661,6 +1683,11 @@ invalid: ...@@ -1661,6 +1683,11 @@ invalid:
return C_ERR; return C_ERR;
} }
/* Wrapper for streamGenericParseIDOrReply() used by module API. */
int streamParseID(const robj *o, streamID *id) {
return streamGenericParseIDOrReply(NULL, o, id, 0, 0);
}
/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to /* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
* 0, to be used when - and + are acceptable IDs. */ * 0, to be used when - and + are acceptable IDs. */
int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) { int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
......
...@@ -27,7 +27,8 @@ TEST_MODULES = \ ...@@ -27,7 +27,8 @@ TEST_MODULES = \
getkeys.so \ getkeys.so \
test_lazyfree.so \ test_lazyfree.so \
timer.so \ timer.so \
defragtest.so defragtest.so \
stream.so
.PHONY: all .PHONY: all
......
#include "redismodule.h"
#include <string.h>
#include <strings.h>
#include <assert.h>
#include <unistd.h>
#include <errno.h>
/* Command which adds a stream entry with automatic ID, like XADD *.
*
* Syntax: STREAM.ADD key field1 value1 [ field2 value2 ... ]
*
* The response is the ID of the added stream entry or an error message.
*/
int stream_add(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 2 || argc % 2 != 0) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
RedisModuleStreamID id;
if (RedisModule_StreamAdd(key, REDISMODULE_STREAM_ADD_AUTOID, &id,
&argv[2], (argc-2)/2) == REDISMODULE_OK) {
RedisModuleString *id_str = RedisModule_CreateStringFromStreamID(ctx, &id);
RedisModule_ReplyWithString(ctx, id_str);
RedisModule_FreeString(ctx, id_str);
} else {
RedisModule_ReplyWithError(ctx, "ERR StreamAdd failed");
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* Command which adds a stream entry N times.
*
* Syntax: STREAM.ADD key N field1 value1 [ field2 value2 ... ]
*
* Returns the number of successfully added entries.
*/
int stream_addn(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 3 || argc % 2 == 0) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
long long n, i;
if (RedisModule_StringToLongLong(argv[2], &n) == REDISMODULE_ERR) {
RedisModule_ReplyWithError(ctx, "N must be a number");
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
for (i = 0; i < n; i++) {
if (RedisModule_StreamAdd(key, REDISMODULE_STREAM_ADD_AUTOID, NULL,
&argv[3], (argc-3)/2) == REDISMODULE_ERR)
break;
}
RedisModule_ReplyWithLongLong(ctx, i);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* STREAM.DELETE key stream-id */
int stream_delete(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 3) return RedisModule_WrongArity(ctx);
RedisModuleStreamID id;
if (RedisModule_StringToStreamID(argv[2], &id) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx, "Invalid stream ID");
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
if (RedisModule_StreamDelete(key, &id) == REDISMODULE_OK) {
RedisModule_ReplyWithSimpleString(ctx, "OK");
} else {
RedisModule_ReplyWithError(ctx, "ERR StreamDelete failed");
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* STREAM.RANGE key start-id end-id
*
* Returns an array of stream items. Each item is an array on the form
* [stream-id, [field1, value1, field2, value2, ...]].
*
* A funny side-effect used for testing RM_StreamIteratorDelete() is that if any
* entry has a field named "selfdestruct", the stream entry is deleted. It is
* however included in the results of this command.
*/
int stream_range(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 4) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleStreamID startid, endid;
if (RedisModule_StringToStreamID(argv[2], &startid) != REDISMODULE_OK ||
RedisModule_StringToStreamID(argv[3], &endid) != REDISMODULE_OK) {
RedisModule_ReplyWithError(ctx, "Invalid stream ID");
return REDISMODULE_OK;
}
/* If startid > endid, we swap and set the reverse flag. */
int flags = 0;
if (startid.ms > endid.ms ||
(startid.ms == endid.ms && startid.seq > endid.seq)) {
RedisModuleStreamID tmp = startid;
startid = endid;
endid = tmp;
flags |= REDISMODULE_STREAM_ITERATOR_REVERSE;
}
/* Open key and start iterator. */
int openflags = REDISMODULE_READ | REDISMODULE_WRITE;
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], openflags);
if (RedisModule_StreamIteratorStart(key, flags,
&startid, &endid) != REDISMODULE_OK) {
/* Key is not a stream, etc. */
RedisModule_ReplyWithError(ctx, "ERR StreamIteratorStart failed");
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* Check error handling: Delete current entry when no current entry. */
assert(RedisModule_StreamIteratorDelete(key) ==
REDISMODULE_ERR);
assert(errno == ENOENT);
/* Check error handling: Fetch fields when no current entry. */
assert(RedisModule_StreamIteratorNextField(key, NULL, NULL) ==
REDISMODULE_ERR);
assert(errno == ENOENT);
/* Return array. */
RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
RedisModule_AutoMemory(ctx);
RedisModuleStreamID id;
long numfields;
long len = 0;
while (RedisModule_StreamIteratorNextID(key, &id,
&numfields) == REDISMODULE_OK) {
RedisModule_ReplyWithArray(ctx, 2);
RedisModuleString *id_str = RedisModule_CreateStringFromStreamID(ctx, &id);
RedisModule_ReplyWithString(ctx, id_str);
RedisModule_ReplyWithArray(ctx, numfields * 2);
int delete = 0;
RedisModuleString *field, *value;
for (long i = 0; i < numfields; i++) {
assert(RedisModule_StreamIteratorNextField(key, &field, &value) ==
REDISMODULE_OK);
RedisModule_ReplyWithString(ctx, field);
RedisModule_ReplyWithString(ctx, value);
/* check if this is a "selfdestruct" field */
size_t field_len;
const char *field_str = RedisModule_StringPtrLen(field, &field_len);
if (!strncmp(field_str, "selfdestruct", field_len)) delete = 1;
}
if (delete) {
assert(RedisModule_StreamIteratorDelete(key) == REDISMODULE_OK);
}
/* check error handling: no more fields to fetch */
assert(RedisModule_StreamIteratorNextField(key, &field, &value) ==
REDISMODULE_ERR);
assert(errno == ENOENT);
len++;
}
RedisModule_ReplySetArrayLength(ctx, len);
RedisModule_StreamIteratorStop(key);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/*
* STREAM.TRIM key (MAXLEN (=|~) length | MINID (=|~) id)
*/
int stream_trim(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 5) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
/* Parse args */
int trim_by_id = 0; /* 0 = maxlen, 1 = minid */
long long maxlen;
RedisModuleStreamID minid;
size_t arg_len;
const char *arg = RedisModule_StringPtrLen(argv[2], &arg_len);
if (!strcasecmp(arg, "minid")) {
trim_by_id = 1;
if (RedisModule_StringToStreamID(argv[4], &minid) != REDISMODULE_OK) {
RedisModule_ReplyWithError(ctx, "ERR Invalid stream ID");
return REDISMODULE_OK;
}
} else if (!strcasecmp(arg, "maxlen")) {
if (RedisModule_StringToLongLong(argv[4], &maxlen) == REDISMODULE_ERR) {
RedisModule_ReplyWithError(ctx, "ERR Maxlen must be a number");
return REDISMODULE_OK;
}
} else {
RedisModule_ReplyWithError(ctx, "ERR Invalid arguments");
return REDISMODULE_OK;
}
/* Approx or exact */
int flags;
arg = RedisModule_StringPtrLen(argv[3], &arg_len);
if (arg_len == 1 && arg[0] == '~') {
flags = REDISMODULE_STREAM_TRIM_APPROX;
} else if (arg_len == 1 && arg[0] == '=') {
flags = 0;
} else {
RedisModule_ReplyWithError(ctx, "ERR Invalid approx-or-exact mark");
return REDISMODULE_OK;
}
/* Trim */
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
long long trimmed;
if (trim_by_id) {
trimmed = RedisModule_StreamTrimByID(key, flags, &minid);
} else {
trimmed = RedisModule_StreamTrimByLength(key, flags, maxlen);
}
/* Return result */
if (trimmed < 0) {
RedisModule_ReplyWithError(ctx, "ERR Trimming failed");
} else {
RedisModule_ReplyWithLongLong(ctx, trimmed);
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx, "stream", 1, REDISMODULE_APIVER_1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.add", stream_add, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.addn", stream_addn, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.delete", stream_delete, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.range", stream_range, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.trim", stream_trim, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
set testmodule [file normalize tests/modules/stream.so]
start_server {tags {"modules"}} {
r module load $testmodule
test {Module stream add and delete} {
r del mystream
# add to empty key
set streamid1 [r stream.add mystream item 1 value a]
# add to existing stream
set streamid2 [r stream.add mystream item 2 value b]
# check result
assert { [string match "*-*" $streamid1] }
set items [r XRANGE mystream - +]
assert_equal $items \
"{$streamid1 {item 1 value a}} {$streamid2 {item 2 value b}}"
# delete one of them and try deleting non-existing ID
assert_equal OK [r stream.delete mystream $streamid1]
assert_error "ERR StreamDelete*" {r stream.delete mystream 123-456}
assert_error "Invalid stream ID*" {r stream.delete mystream foo}
assert_equal "{$streamid2 {item 2 value b}}" [r XRANGE mystream - +]
# check error condition: wrong type
r del mystream
r set mystream mystring
assert_error "ERR StreamAdd*" {r stream.add mystream item 1 value a}
assert_error "ERR StreamDelete*" {r stream.delete mystream 123-456}
}
test {Module stream add unblocks blocking xread} {
r del mystream
# Blocking XREAD on an empty key
set rd1 [redis_deferring_client]
$rd1 XREAD BLOCK 3000 STREAMS mystream $
# wait until client is actually blocked
wait_for_condition 50 100 {
[s 0 blocked_clients] eq {1}
} else {
fail "Client is not blocked"
}
set id [r stream.add mystream field 1 value a]
assert_equal "{mystream {{$id {field 1 value a}}}}" [$rd1 read]
# Blocking XREAD on an existing stream
set rd2 [redis_deferring_client]
$rd2 XREAD BLOCK 3000 STREAMS mystream $
# wait until client is actually blocked
wait_for_condition 50 100 {
[s 0 blocked_clients] eq {1}
} else {
fail "Client is not blocked"
}
set id [r stream.add mystream field 2 value b]
assert_equal "{mystream {{$id {field 2 value b}}}}" [$rd2 read]
}
test {Module stream add benchmark (1M stream add)} {
set n 1000000
r del mystream
set result [r stream.addn mystream $n field value]
assert_equal $result $n
}
test {Module stream iterator} {
r del mystream
set streamid1 [r xadd mystream * item 1 value a]
set streamid2 [r xadd mystream * item 2 value b]
# range result
set result1 [r stream.range mystream "-" "+"]
set expect1 [r xrange mystream "-" "+"]
assert_equal $result1 $expect1
# reverse range
set result_rev [r stream.range mystream "+" "-"]
set expect_rev [r xrevrange mystream "+" "-"]
assert_equal $result_rev $expect_rev
# only one item: range with startid = endid
set result2 [r stream.range mystream "-" $streamid1]
assert_equal $result2 "{$streamid1 {item 1 value a}}"
assert_equal $result2 [list [list $streamid1 {item 1 value a}]]
# only one item: range with startid = endid
set result3 [r stream.range mystream $streamid2 $streamid2]
assert_equal $result3 "{$streamid2 {item 2 value b}}"
assert_equal $result3 [list [list $streamid2 {item 2 value b}]]
}
test {Module stream iterator delete} {
r del mystream
set id1 [r xadd mystream * normal item]
set id2 [r xadd mystream * selfdestruct yes]
set id3 [r xadd mystream * another item]
# stream.range deletes the "selfdestruct" item after returning it
assert_equal \
"{$id1 {normal item}} {$id2 {selfdestruct yes}} {$id3 {another item}}" \
[r stream.range mystream - +]
# now, the "selfdestruct" item is gone
assert_equal \
"{$id1 {normal item}} {$id3 {another item}}" \
[r stream.range mystream - +]
}
test {Module stream trim by length} {
r del mystream
# exact maxlen
r xadd mystream * item 1 value a
r xadd mystream * item 2 value b
r xadd mystream * item 3 value c
assert_equal 3 [r xlen mystream]
assert_equal 0 [r stream.trim mystream maxlen = 5]
assert_equal 3 [r xlen mystream]
assert_equal 2 [r stream.trim mystream maxlen = 1]
assert_equal 1 [r xlen mystream]
assert_equal 1 [r stream.trim mystream maxlen = 0]
# check that there is no limit for exact maxlen
r stream.addn mystream 20000 item x value y
assert_equal 20000 [r stream.trim mystream maxlen = 0]
# approx maxlen (100 items per node implies default limit 10K items)
r stream.addn mystream 20000 item x value y
assert_equal 20000 [r xlen mystream]
assert_equal 10000 [r stream.trim mystream maxlen ~ 2]
assert_equal 9900 [r stream.trim mystream maxlen ~ 2]
assert_equal 0 [r stream.trim mystream maxlen ~ 2]
assert_equal 100 [r xlen mystream]
assert_equal 100 [r stream.trim mystream maxlen ~ 0]
assert_equal 0 [r xlen mystream]
}
test {Module stream trim by ID} {
r del mystream
# exact minid
r xadd mystream * item 1 value a
r xadd mystream * item 2 value b
set minid [r xadd mystream * item 3 value c]
assert_equal 3 [r xlen mystream]
assert_equal 0 [r stream.trim mystream minid = -]
assert_equal 3 [r xlen mystream]
assert_equal 2 [r stream.trim mystream minid = $minid]
assert_equal 1 [r xlen mystream]
assert_equal 1 [r stream.trim mystream minid = +]
# check that there is no limit for exact minid
r stream.addn mystream 20000 item x value y
assert_equal 20000 [r stream.trim mystream minid = +]
# approx minid (100 items per node implies default limit 10K items)
r stream.addn mystream 19980 item x value y
set minid [r xadd mystream * item x value y]
r stream.addn mystream 19 item x value y
assert_equal 20000 [r xlen mystream]
assert_equal 10000 [r stream.trim mystream minid ~ $minid]
assert_equal 9900 [r stream.trim mystream minid ~ $minid]
assert_equal 0 [r stream.trim mystream minid ~ $minid]
assert_equal 100 [r xlen mystream]
assert_equal 100 [r stream.trim mystream minid ~ +]
assert_equal 0 [r xlen mystream]
}
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment