Commit d0854927 authored by Oran Agra's avatar Oran Agra
Browse files

Merge branch 'unstable' into 6.2

parents ec2d1807 b57d0eb4
...@@ -459,12 +459,12 @@ struct commandHelp { ...@@ -459,12 +459,12 @@ struct commandHelp {
0, 0,
"1.2.0" }, "1.2.0" },
{ "FLUSHALL", { "FLUSHALL",
"[ASYNC]", "[ASYNC|SYNC]",
"Remove all keys from all databases", "Remove all keys from all databases",
9, 9,
"1.0.0" }, "1.0.0" },
{ "FLUSHDB", { "FLUSHDB",
"[ASYNC]", "[ASYNC|SYNC]",
"Remove all keys from the current database", "Remove all keys from the current database",
9, 9,
"1.0.0" }, "1.0.0" },
...@@ -518,6 +518,16 @@ struct commandHelp { ...@@ -518,6 +518,16 @@ struct commandHelp {
"Returns the bit value at offset in the string value stored at key", "Returns the bit value at offset in the string value stored at key",
1, 1,
"2.2.0" }, "2.2.0" },
{ "GETDEL",
"key",
"Get the value of a key and delete the key",
1,
"6.2.0" },
{ "GETEX",
"key [EX seconds|PX milliseconds|EXAT timestamp|PXAT milliseconds-timestamp|PERSIST]",
"Get the value of a key and optionally set its expiration",
1,
"6.2.0" },
{ "GETRANGE", { "GETRANGE",
"key start end", "key start end",
"Get a substring of the string stored at a key", "Get a substring of the string stored at a key",
...@@ -583,6 +593,11 @@ struct commandHelp { ...@@ -583,6 +593,11 @@ struct commandHelp {
"Set multiple hash fields to multiple values", "Set multiple hash fields to multiple values",
5, 5,
"2.0.0" }, "2.0.0" },
{ "HRANDFIELD",
"key [count [WITHVALUES]]",
"Get one or multiple random fields from a hash",
5,
"6.2.0" },
{ "HSCAN", { "HSCAN",
"key cursor [MATCH pattern] [COUNT count]", "key cursor [MATCH pattern] [COUNT count]",
"Incrementally iterate hash fields and associated values", "Incrementally iterate hash fields and associated values",
...@@ -989,7 +1004,7 @@ struct commandHelp { ...@@ -989,7 +1004,7 @@ struct commandHelp {
10, 10,
"2.6.0" }, "2.6.0" },
{ "SCRIPT FLUSH", { "SCRIPT FLUSH",
"-", "[ASYNC|SYNC]",
"Remove all the scripts from the script cache.", "Remove all the scripts from the script cache.",
10, 10,
"2.6.0" }, "2.6.0" },
...@@ -1019,7 +1034,7 @@ struct commandHelp { ...@@ -1019,7 +1034,7 @@ struct commandHelp {
8, 8,
"1.0.0" }, "1.0.0" },
{ "SET", { "SET",
"key value [EX seconds|PX milliseconds|KEEPTTL] [NX|XX] [GET]", "key value [EX seconds|PX milliseconds|EXAT timestamp|PXAT milliseconds-timestamp|KEEPTTL] [NX|XX] [GET]",
"Set the string value of a key", "Set the string value of a key",
1, 1,
"1.0.0" }, "1.0.0" },
...@@ -1323,6 +1338,11 @@ struct commandHelp { ...@@ -1323,6 +1338,11 @@ struct commandHelp {
"Remove and return members with the lowest scores in a sorted set", "Remove and return members with the lowest scores in a sorted set",
4, 4,
"5.0.0" }, "5.0.0" },
{ "ZRANDMEMBER",
"key [count [WITHSCORES]]",
"Get one or multiple random elements from a sorted set",
4,
"6.2.0" },
{ "ZRANGE", { "ZRANGE",
"key min max [BYSCORE|BYLEX] [REV] [LIMIT offset count] [WITHSCORES]", "key min max [BYSCORE|BYLEX] [REV] [LIMIT offset count] [WITHSCORES]",
"Return a range of members in a sorted set", "Return a range of members in a sorted set",
......
...@@ -49,6 +49,14 @@ void lazyFreeTrackingTable(void *args[]) { ...@@ -49,6 +49,14 @@ void lazyFreeTrackingTable(void *args[]) {
atomicIncr(lazyfreed_objects,len); atomicIncr(lazyfreed_objects,len);
} }
void lazyFreeLuaScripts(void *args[]) {
dict *lua_scripts = args[0];
long long len = dictSize(lua_scripts);
dictRelease(lua_scripts);
atomicDecr(lazyfree_objects,len);
atomicIncr(lazyfreed_objects,len);
}
/* Return the number of currently pending objects to free. */ /* Return the number of currently pending objects to free. */
size_t lazyfreeGetPendingObjectsCount(void) { size_t lazyfreeGetPendingObjectsCount(void) {
size_t aux; size_t aux;
...@@ -212,3 +220,13 @@ void freeTrackingRadixTreeAsync(rax *tracking) { ...@@ -212,3 +220,13 @@ void freeTrackingRadixTreeAsync(rax *tracking) {
atomicIncr(lazyfree_objects,tracking->numele); atomicIncr(lazyfree_objects,tracking->numele);
bioCreateLazyFreeJob(lazyFreeTrackingTable,1,tracking); bioCreateLazyFreeJob(lazyFreeTrackingTable,1,tracking);
} }
/* Free lua_scripts dict, if the dict is huge enough, free it in async way. */
void freeLuaScriptsAsync(dict *lua_scripts) {
if (dictSize(lua_scripts) > LAZYFREE_THRESHOLD) {
atomicIncr(lazyfree_objects,dictSize(lua_scripts));
bioCreateLazyFreeJob(lazyFreeLuaScripts,1,lua_scripts);
} else {
dictRelease(lua_scripts);
}
}
This diff is collapsed.
...@@ -4,16 +4,26 @@ ...@@ -4,16 +4,26 @@
# Convert the C comment to markdown # Convert the C comment to markdown
def markdown(s) def markdown(s)
s = s.gsub(/\*\/$/,"") s = s.gsub(/\*\/$/,"")
s = s.gsub(/^ \* {0,1}/,"") s = s.gsub(/^ ?\* ?/,"")
s = s.gsub(/^\/\* /,"") s = s.gsub(/^\/\*\*? ?/,"")
s.chop! while s[-1] == "\n" || s[-1] == " " s.chop! while s[-1] == "\n" || s[-1] == " "
lines = s.split("\n") lines = s.split("\n")
newlines = [] newlines = []
# Fix some markdown, except in code blocks indented by 4 spaces.
lines.each{|l| lines.each{|l|
if l[0] != ' ' if not l.start_with?(' ')
l = l.gsub(/RM_[A-z()]+/){|x| "`#{x}`"} # Rewrite RM_Xyz() to `RedisModule_Xyz()`. The () suffix is
l = l.gsub(/RedisModule_[A-z()]+/){|x| "`#{x}`"} # optional. Even RM_Xyz*() with * as wildcard is handled.
l = l.gsub(/REDISMODULE_[A-z]+/){|x| "`#{x}`"} l = l.gsub(/(?<!`)RM_([A-z]+(?:\*?\(\))?)/, '`RedisModule_\1`')
# Add backquotes around RedisModule functions and type where missing.
l = l.gsub(/(?<!`)RedisModule[A-z]+(?:\*?\(\))?/){|x| "`#{x}`"}
# Add backquotes around c functions like malloc() where missing.
l = l.gsub(/(?<![`A-z])[a-z_]+\(\)/, '`\0`')
# Add backquotes around macro and var names containing underscores.
l = l.gsub(/(?<![`A-z\*])[A-Za-z]+_[A-Za-z0-9_]+/){|x| "`#{x}`"}
# Link URLs preceded by space (i.e. when not already linked)
l = l.gsub(/ (https?:\/\/[A-Za-z0-9_\/\.\-]+[A-Za-z0-9\/])/,
' [\1](\1)')
end end
newlines << l newlines << l
} }
...@@ -41,6 +51,7 @@ def docufy(src,i) ...@@ -41,6 +51,7 @@ def docufy(src,i)
end end
puts "# Modules API reference\n\n" puts "# Modules API reference\n\n"
puts "<!-- This file is generated from module.c using gendoc.rb -->\n\n"
src = File.open("../module.c").to_a src = File.open("../module.c").to_a
src.each_with_index{|line,i| src.each_with_index{|line,i|
if line =~ /RM_/ && line[0] != ' ' && line[0] != '#' && line[0] != '/' if line =~ /RM_/ && line[0] != ' ' && line[0] != '#' && line[0] != '/'
......
...@@ -1104,6 +1104,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1104,6 +1104,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
"Accepting client connection: %s", server.neterr); "Accepting client connection: %s", server.neterr);
return; return;
} }
anetCloexec(cfd);
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport); serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip); acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
} }
...@@ -1124,6 +1125,7 @@ void acceptTLSHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1124,6 +1125,7 @@ void acceptTLSHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
"Accepting client connection: %s", server.neterr); "Accepting client connection: %s", server.neterr);
return; return;
} }
anetCloexec(cfd);
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport); serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
acceptCommonHandler(connCreateAcceptedTLS(cfd, server.tls_auth_clients),0,cip); acceptCommonHandler(connCreateAcceptedTLS(cfd, server.tls_auth_clients),0,cip);
} }
...@@ -1143,6 +1145,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1143,6 +1145,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
"Accepting client connection: %s", server.neterr); "Accepting client connection: %s", server.neterr);
return; return;
} }
anetCloexec(cfd);
serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket); serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket);
acceptCommonHandler(connCreateAcceptedSocket(cfd),CLIENT_UNIX_SOCKET,NULL); acceptCommonHandler(connCreateAcceptedSocket(cfd),CLIENT_UNIX_SOCKET,NULL);
} }
...@@ -1707,7 +1710,7 @@ int processInlineBuffer(client *c) { ...@@ -1707,7 +1710,7 @@ int processInlineBuffer(client *c) {
} }
/* Handle the \r\n case. */ /* Handle the \r\n case. */
if (newline && newline != c->querybuf+c->qb_pos && *(newline-1) == '\r') if (newline != c->querybuf+c->qb_pos && *(newline-1) == '\r')
newline--, linefeed_chars++; newline--, linefeed_chars++;
/* Split the input buffer up to the \r\n */ /* Split the input buffer up to the \r\n */
...@@ -2436,8 +2439,10 @@ void clientCommand(client *c) { ...@@ -2436,8 +2439,10 @@ void clientCommand(client *c) {
" Kill connection made from <ip:port>.", " Kill connection made from <ip:port>.",
"KILL <option> <value> [<option> <value> [...]]", "KILL <option> <value> [<option> <value> [...]]",
" Kill connections. Options are:", " Kill connections. Options are:",
" * ADDR <ip:port>", " * ADDR (<ip:port>|<unixsocket>:0)",
" Kill connection made from <ip:port>", " Kill connections made from the specified address",
" * LADDR (<ip:port>|<unixsocket>:0)",
" Kill connections made to specified local address",
" * TYPE (normal|master|replica|pubsub)", " * TYPE (normal|master|replica|pubsub)",
" Kill connections by type.", " Kill connections by type.",
" * USER <username>", " * USER <username>",
...@@ -2675,7 +2680,7 @@ NULL ...@@ -2675,7 +2680,7 @@ NULL
c->argc == 4)) c->argc == 4))
{ {
/* CLIENT PAUSE TIMEOUT [WRITE|ALL] */ /* CLIENT PAUSE TIMEOUT [WRITE|ALL] */
long long duration; mstime_t end;
int type = CLIENT_PAUSE_ALL; int type = CLIENT_PAUSE_ALL;
if (c->argc == 4) { if (c->argc == 4) {
if (!strcasecmp(c->argv[3]->ptr,"write")) { if (!strcasecmp(c->argv[3]->ptr,"write")) {
...@@ -2689,9 +2694,9 @@ NULL ...@@ -2689,9 +2694,9 @@ NULL
} }
} }
if (getTimeoutFromObjectOrReply(c,c->argv[2],&duration, if (getTimeoutFromObjectOrReply(c,c->argv[2],&end,
UNIT_MILLISECONDS) != C_OK) return; UNIT_MILLISECONDS) != C_OK) return;
pauseClients(duration, type); pauseClients(end, type);
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"tracking") && c->argc >= 3) { } else if (!strcasecmp(c->argv[1]->ptr,"tracking") && c->argc >= 3) {
/* CLIENT TRACKING (on|off) [REDIRECT <id>] [BCAST] [PREFIX first] /* CLIENT TRACKING (on|off) [REDIRECT <id>] [BCAST] [PREFIX first]
...@@ -3355,8 +3360,6 @@ void processEventsWhileBlocked(void) { ...@@ -3355,8 +3360,6 @@ void processEventsWhileBlocked(void) {
* Threaded I/O * Threaded I/O
* ========================================================================== */ * ========================================================================== */
int tio_debug = 0;
#define IO_THREADS_MAX_NUM 128 #define IO_THREADS_MAX_NUM 128
#define IO_THREADS_OP_READ 0 #define IO_THREADS_OP_READ 0
#define IO_THREADS_OP_WRITE 1 #define IO_THREADS_OP_WRITE 1
...@@ -3407,8 +3410,6 @@ void *IOThreadMain(void *myid) { ...@@ -3407,8 +3410,6 @@ void *IOThreadMain(void *myid) {
serverAssert(getIOPendingCount(id) != 0); serverAssert(getIOPendingCount(id) != 0);
if (tio_debug) printf("[%ld] %d to handle\n", id, (int)listLength(io_threads_list[id]));
/* Process: note that the main thread will never touch our list /* Process: note that the main thread will never touch our list
* before we drop the pending count to 0. */ * before we drop the pending count to 0. */
listIter li; listIter li;
...@@ -3426,8 +3427,6 @@ void *IOThreadMain(void *myid) { ...@@ -3426,8 +3427,6 @@ void *IOThreadMain(void *myid) {
} }
listEmpty(io_threads_list[id]); listEmpty(io_threads_list[id]);
setIOPendingCount(id, 0); setIOPendingCount(id, 0);
if (tio_debug) printf("[%ld] Done\n", id);
} }
} }
...@@ -3482,8 +3481,6 @@ void killIOThreads(void) { ...@@ -3482,8 +3481,6 @@ void killIOThreads(void) {
} }
void startThreadedIO(void) { void startThreadedIO(void) {
if (tio_debug) { printf("S"); fflush(stdout); }
if (tio_debug) printf("--- STARTING THREADED IO ---\n");
serverAssert(server.io_threads_active == 0); serverAssert(server.io_threads_active == 0);
for (int j = 1; j < server.io_threads_num; j++) for (int j = 1; j < server.io_threads_num; j++)
pthread_mutex_unlock(&io_threads_mutex[j]); pthread_mutex_unlock(&io_threads_mutex[j]);
...@@ -3494,10 +3491,6 @@ void stopThreadedIO(void) { ...@@ -3494,10 +3491,6 @@ void stopThreadedIO(void) {
/* We may have still clients with pending reads when this function /* We may have still clients with pending reads when this function
* is called: handle them before stopping the threads. */ * is called: handle them before stopping the threads. */
handleClientsWithPendingReadsUsingThreads(); handleClientsWithPendingReadsUsingThreads();
if (tio_debug) { printf("E"); fflush(stdout); }
if (tio_debug) printf("--- STOPPING THREADED IO [R%d] [W%d] ---\n",
(int) listLength(server.clients_pending_read),
(int) listLength(server.clients_pending_write));
serverAssert(server.io_threads_active == 1); serverAssert(server.io_threads_active == 1);
for (int j = 1; j < server.io_threads_num; j++) for (int j = 1; j < server.io_threads_num; j++)
pthread_mutex_lock(&io_threads_mutex[j]); pthread_mutex_lock(&io_threads_mutex[j]);
...@@ -3540,8 +3533,6 @@ int handleClientsWithPendingWritesUsingThreads(void) { ...@@ -3540,8 +3533,6 @@ int handleClientsWithPendingWritesUsingThreads(void) {
/* Start threads if needed. */ /* Start threads if needed. */
if (!server.io_threads_active) startThreadedIO(); if (!server.io_threads_active) startThreadedIO();
if (tio_debug) printf("%d TOTAL WRITE pending clients\n", processed);
/* Distribute the clients across N different lists. */ /* Distribute the clients across N different lists. */
listIter li; listIter li;
listNode *ln; listNode *ln;
...@@ -3586,7 +3577,6 @@ int handleClientsWithPendingWritesUsingThreads(void) { ...@@ -3586,7 +3577,6 @@ int handleClientsWithPendingWritesUsingThreads(void) {
pending += getIOPendingCount(j); pending += getIOPendingCount(j);
if (pending == 0) break; if (pending == 0) break;
} }
if (tio_debug) printf("I/O WRITE All threads finshed\n");
/* Run the list of clients again to install the write handler where /* Run the list of clients again to install the write handler where
* needed. */ * needed. */
...@@ -3639,8 +3629,6 @@ int handleClientsWithPendingReadsUsingThreads(void) { ...@@ -3639,8 +3629,6 @@ int handleClientsWithPendingReadsUsingThreads(void) {
int processed = listLength(server.clients_pending_read); int processed = listLength(server.clients_pending_read);
if (processed == 0) return 0; if (processed == 0) return 0;
if (tio_debug) printf("%d TOTAL READ pending clients\n", processed);
/* Distribute the clients across N different lists. */ /* Distribute the clients across N different lists. */
listIter li; listIter li;
listNode *ln; listNode *ln;
...@@ -3676,7 +3664,6 @@ int handleClientsWithPendingReadsUsingThreads(void) { ...@@ -3676,7 +3664,6 @@ int handleClientsWithPendingReadsUsingThreads(void) {
pending += getIOPendingCount(j); pending += getIOPendingCount(j);
if (pending == 0) break; if (pending == 0) break;
} }
if (tio_debug) printf("I/O READ All threads finshed\n");
/* Run the list of clients again to process the new buffers. */ /* Run the list of clients again to process the new buffers. */
while(listLength(server.clients_pending_read)) { while(listLength(server.clients_pending_read)) {
......
...@@ -5301,7 +5301,7 @@ static clusterManagerNode *clusterNodeForResharding(char *id, ...@@ -5301,7 +5301,7 @@ static clusterManagerNode *clusterNodeForResharding(char *id,
clusterManagerLogErr(invalid_node_msg, id); clusterManagerLogErr(invalid_node_msg, id);
*raise_err = 1; *raise_err = 1;
return NULL; return NULL;
} else if (node != NULL && target != NULL) { } else if (target != NULL) {
if (!strcmp(node->name, target->name)) { if (!strcmp(node->name, target->name)) {
clusterManagerLogErr( "*** It is not possible to use " clusterManagerLogErr( "*** It is not possible to use "
"the target node as " "the target node as "
...@@ -6940,6 +6940,10 @@ void sendCapa() { ...@@ -6940,6 +6940,10 @@ void sendCapa() {
sendReplconf("capa", "eof"); sendReplconf("capa", "eof");
} }
void sendRdbOnly(void) {
sendReplconf("rdb-only", "1");
}
/* Read raw bytes through a redisContext. The read operation is not greedy /* Read raw bytes through a redisContext. The read operation is not greedy
* and may not fill the buffer entirely. * and may not fill the buffer entirely.
*/ */
...@@ -7137,7 +7141,6 @@ static void getRDB(clusterManagerNode *node) { ...@@ -7137,7 +7141,6 @@ static void getRDB(clusterManagerNode *node) {
node->context = NULL; node->context = NULL;
fsync(fd); fsync(fd);
close(fd); close(fd);
fprintf(stderr,"Transfer finished with success.\n");
if (node) { if (node) {
sdsfree(filename); sdsfree(filename);
return; return;
...@@ -8258,6 +8261,7 @@ int main(int argc, char **argv) { ...@@ -8258,6 +8261,7 @@ int main(int argc, char **argv) {
if (config.getrdb_mode) { if (config.getrdb_mode) {
if (cliConnect(0) == REDIS_ERR) exit(1); if (cliConnect(0) == REDIS_ERR) exit(1);
sendCapa(); sendCapa();
sendRdbOnly();
getRDB(NULL); getRDB(NULL);
} }
......
...@@ -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(). */
...@@ -216,9 +230,8 @@ typedef uint64_t RedisModuleTimerID; ...@@ -216,9 +230,8 @@ typedef uint64_t RedisModuleTimerID;
#define REDISMODULE_EVENT_LOADING_PROGRESS 10 #define REDISMODULE_EVENT_LOADING_PROGRESS 10
#define REDISMODULE_EVENT_SWAPDB 11 #define REDISMODULE_EVENT_SWAPDB 11
#define REDISMODULE_EVENT_REPL_BACKUP 12 #define REDISMODULE_EVENT_REPL_BACKUP 12
#define REDISMODULE_EVENT_FORK_CHILD 13
/* Next event flag, should be updated if a new event added. */ #define _REDISMODULE_EVENT_NEXT 14 /* Next event flag, should be updated if a new event added. */
#define _REDISMODULE_EVENT_NEXT 13
typedef struct RedisModuleEvent { typedef struct RedisModuleEvent {
uint64_t id; /* REDISMODULE_EVENT_... defines. */ uint64_t id; /* REDISMODULE_EVENT_... defines. */
...@@ -281,6 +294,10 @@ static const RedisModuleEvent ...@@ -281,6 +294,10 @@ static const RedisModuleEvent
RedisModuleEvent_ReplBackup = { RedisModuleEvent_ReplBackup = {
REDISMODULE_EVENT_REPL_BACKUP, REDISMODULE_EVENT_REPL_BACKUP,
1 1
},
RedisModuleEvent_ForkChild = {
REDISMODULE_EVENT_FORK_CHILD,
1
}; };
/* Those are values that are used for the 'subevent' callback argument. */ /* Those are values that are used for the 'subevent' callback argument. */
...@@ -331,6 +348,10 @@ static const RedisModuleEvent ...@@ -331,6 +348,10 @@ static const RedisModuleEvent
#define REDISMODULE_SUBEVENT_REPL_BACKUP_DISCARD 2 #define REDISMODULE_SUBEVENT_REPL_BACKUP_DISCARD 2
#define _REDISMODULE_SUBEVENT_REPL_BACKUP_NEXT 3 #define _REDISMODULE_SUBEVENT_REPL_BACKUP_NEXT 3
#define REDISMODULE_SUBEVENT_FORK_CHILD_BORN 0
#define REDISMODULE_SUBEVENT_FORK_CHILD_DIED 1
#define _REDISMODULE_SUBEVENT_FORK_CHILD_NEXT 2
#define _REDISMODULE_SUBEVENT_SHUTDOWN_NEXT 0 #define _REDISMODULE_SUBEVENT_SHUTDOWN_NEXT 0
#define _REDISMODULE_SUBEVENT_CRON_LOOP_NEXT 0 #define _REDISMODULE_SUBEVENT_CRON_LOOP_NEXT 0
#define _REDISMODULE_SUBEVENT_SWAPDB_NEXT 0 #define _REDISMODULE_SUBEVENT_SWAPDB_NEXT 0
...@@ -578,6 +599,7 @@ REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongLong)(Redi ...@@ -578,6 +599,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 +621,7 @@ REDISMODULE_API int (*RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, Redis ...@@ -599,6 +621,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 +652,15 @@ REDISMODULE_API int (*RedisModule_ZsetRangePrev)(RedisModuleKey *key) REDISMODUL ...@@ -629,6 +652,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;
...@@ -744,6 +776,8 @@ REDISMODULE_API int (*RedisModule_IsBlockedTimeoutRequest)(RedisModuleCtx *ctx) ...@@ -744,6 +776,8 @@ REDISMODULE_API int (*RedisModule_IsBlockedTimeoutRequest)(RedisModuleCtx *ctx)
REDISMODULE_API void * (*RedisModule_GetBlockedClientPrivateData)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API void * (*RedisModule_GetBlockedClientPrivateData)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleBlockedClient * (*RedisModule_GetBlockedClientHandle)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API RedisModuleBlockedClient * (*RedisModule_GetBlockedClientHandle)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_AbortBlock)(RedisModuleBlockedClient *bc) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_AbortBlock)(RedisModuleBlockedClient *bc) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_BlockedClientMeasureTimeStart)(RedisModuleBlockedClient *bc) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_BlockedClientMeasureTimeEnd)(RedisModuleBlockedClient *bc) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleCtx * (*RedisModule_GetThreadSafeContext)(RedisModuleBlockedClient *bc) REDISMODULE_ATTR; REDISMODULE_API RedisModuleCtx * (*RedisModule_GetThreadSafeContext)(RedisModuleBlockedClient *bc) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleCtx * (*RedisModule_GetDetachedThreadSafeContext)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API RedisModuleCtx * (*RedisModule_GetDetachedThreadSafeContext)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_FreeThreadSafeContext)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_FreeThreadSafeContext)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
...@@ -842,6 +876,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -842,6 +876,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 +891,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -856,6 +891,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 +923,15 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -887,6 +923,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);
...@@ -1006,6 +1051,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1006,6 +1051,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(GetBlockedClientPrivateData); REDISMODULE_GET_API(GetBlockedClientPrivateData);
REDISMODULE_GET_API(GetBlockedClientHandle); REDISMODULE_GET_API(GetBlockedClientHandle);
REDISMODULE_GET_API(AbortBlock); REDISMODULE_GET_API(AbortBlock);
REDISMODULE_GET_API(BlockedClientMeasureTimeStart);
REDISMODULE_GET_API(BlockedClientMeasureTimeEnd);
REDISMODULE_GET_API(SetDisconnectCallback); REDISMODULE_GET_API(SetDisconnectCallback);
REDISMODULE_GET_API(SubscribeToKeyspaceEvents); REDISMODULE_GET_API(SubscribeToKeyspaceEvents);
REDISMODULE_GET_API(NotifyKeyspaceEvent); REDISMODULE_GET_API(NotifyKeyspaceEvent);
......
This diff is collapsed.
...@@ -1282,14 +1282,17 @@ void scriptingInit(int setup) { ...@@ -1282,14 +1282,17 @@ void scriptingInit(int setup) {
/* Release resources related to Lua scripting. /* Release resources related to Lua scripting.
* This function is used in order to reset the scripting environment. */ * This function is used in order to reset the scripting environment. */
void scriptingRelease(void) { void scriptingRelease(int async) {
dictRelease(server.lua_scripts); if (async)
freeLuaScriptsAsync(server.lua_scripts);
else
dictRelease(server.lua_scripts);
server.lua_scripts_mem = 0; server.lua_scripts_mem = 0;
lua_close(server.lua); lua_close(server.lua);
} }
void scriptingReset(void) { void scriptingReset(int async) {
scriptingRelease(); scriptingRelease(async);
scriptingInit(0); scriptingInit(0);
} }
...@@ -1711,8 +1714,12 @@ void scriptCommand(client *c) { ...@@ -1711,8 +1714,12 @@ void scriptCommand(client *c) {
" Set the debug mode for subsequent scripts executed.", " Set the debug mode for subsequent scripts executed.",
"EXISTS <sha1> [<sha1> ...]", "EXISTS <sha1> [<sha1> ...]",
" Return information about the existence of the scripts in the script cache.", " Return information about the existence of the scripts in the script cache.",
"FLUSH", "FLUSH [ASYNC|SYNC]",
" Flush the Lua scripts cache. Very dangerous on replicas.", " Flush the Lua scripts cache. Very dangerous on replicas.",
" When called without the optional mode argument, the behavior is determined by the",
" lazyfree-lazy-user-flush configuration directive. Valid modes are:",
" * ASYNC: Asynchronously flush the scripts cache.",
" * SYNC: Synchronously flush the scripts cache.",
"KILL", "KILL",
" Kill the currently executing Lua script.", " Kill the currently executing Lua script.",
"LOAD <script>", "LOAD <script>",
...@@ -1720,8 +1727,19 @@ void scriptCommand(client *c) { ...@@ -1720,8 +1727,19 @@ void scriptCommand(client *c) {
NULL NULL
}; };
addReplyHelp(c, help); addReplyHelp(c, help);
} else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"flush")) { } else if (c->argc >= 2 && !strcasecmp(c->argv[1]->ptr,"flush")) {
scriptingReset(); int async = 0;
if (c->argc == 3 && !strcasecmp(c->argv[2]->ptr,"sync")) {
async = 0;
} else if (c->argc == 3 && !strcasecmp(c->argv[2]->ptr,"async")) {
async = 1;
} else if (c->argc == 2) {
async = server.lazyfree_lazy_user_flush ? 1 : 0;
} else {
addReplyError(c,"SCRIPT FLUSH only support SYNC|ASYNC option");
return;
}
scriptingReset(async);
addReply(c,shared.ok); addReply(c,shared.ok);
replicationScriptCacheFlush(); replicationScriptCacheFlush();
server.dirty++; /* Propagating this command is a good idea. */ server.dirty++; /* Propagating this command is a good idea. */
......
...@@ -1157,12 +1157,80 @@ void *sds_malloc(size_t size) { return s_malloc(size); } ...@@ -1157,12 +1157,80 @@ void *sds_malloc(size_t size) { return s_malloc(size); }
void *sds_realloc(void *ptr, size_t size) { return s_realloc(ptr,size); } void *sds_realloc(void *ptr, size_t size) { return s_realloc(ptr,size); }
void sds_free(void *ptr) { s_free(ptr); } void sds_free(void *ptr) { s_free(ptr); }
/* Perform expansion of a template string and return the result as a newly
* allocated sds.
*
* Template variables are specified using curly brackets, e.g. {variable}.
* An opening bracket can be quoted by repeating it twice.
*/
sds sdstemplate(const char *template, sdstemplate_callback_t cb_func, void *cb_arg)
{
sds res = sdsempty();
const char *p = template;
while (*p) {
/* Find next variable, copy everything until there */
const char *sv = strchr(p, '{');
if (!sv) {
/* Not found: copy till rest of template and stop */
res = sdscat(res, p);
break;
} else if (sv > p) {
/* Found: copy anything up to the begining of the variable */
res = sdscatlen(res, p, sv - p);
}
/* Skip into variable name, handle premature end or quoting */
sv++;
if (!*sv) goto error; /* Premature end of template */
if (*sv == '{') {
/* Quoted '{' */
p = sv + 1;
res = sdscat(res, "{");
continue;
}
/* Find end of variable name, handle premature end of template */
const char *ev = strchr(sv, '}');
if (!ev) goto error;
/* Pass variable name to callback and obtain value. If callback failed,
* abort. */
sds varname = sdsnewlen(sv, ev - sv);
sds value = cb_func(varname, cb_arg);
sdsfree(varname);
if (!value) goto error;
/* Append value to result and continue */
res = sdscat(res, value);
sdsfree(value);
p = ev + 1;
}
return res;
error:
sdsfree(res);
return NULL;
}
#ifdef REDIS_TEST #ifdef REDIS_TEST
#include <stdio.h> #include <stdio.h>
#include <limits.h> #include <limits.h>
#include "testhelp.h" #include "testhelp.h"
#define UNUSED(x) (void)(x) #define UNUSED(x) (void)(x)
static sds sdsTestTemplateCallback(sds varname, void *arg) {
UNUSED(arg);
static const char *_var1 = "variable1";
static const char *_var2 = "variable2";
if (!strcmp(varname, _var1)) return sdsnew("value1");
else if (!strcmp(varname, _var2)) return sdsnew("value2");
else return NULL;
}
int sdsTest(int argc, char **argv) { int sdsTest(int argc, char **argv) {
UNUSED(argc); UNUSED(argc);
UNUSED(argv); UNUSED(argv);
...@@ -1342,6 +1410,30 @@ int sdsTest(int argc, char **argv) { ...@@ -1342,6 +1410,30 @@ int sdsTest(int argc, char **argv) {
sdsfree(x); sdsfree(x);
} }
/* Simple template */
x = sdstemplate("v1={variable1} v2={variable2}", sdsTestTemplateCallback, NULL);
test_cond("sdstemplate() normal flow",
memcmp(x,"v1=value1 v2=value2",19) == 0);
sdsfree(x);
/* Template with callback error */
x = sdstemplate("v1={variable1} v3={doesnotexist}", sdsTestTemplateCallback, NULL);
test_cond("sdstemplate() with callback error", x == NULL);
/* Template with empty var name */
x = sdstemplate("v1={", sdsTestTemplateCallback, NULL);
test_cond("sdstemplate() with empty var name", x == NULL);
/* Template with truncated var name */
x = sdstemplate("v1={start", sdsTestTemplateCallback, NULL);
test_cond("sdstemplate() with truncated var name", x == NULL);
/* Template with quoting */
x = sdstemplate("v1={{{variable1}} {{} v2={variable2}", sdsTestTemplateCallback, NULL);
test_cond("sdstemplate() with quoting",
memcmp(x,"v1={value1} {} v2=value2",24) == 0);
sdsfree(x);
} }
test_report(); test_report();
return 0; return 0;
......
...@@ -253,6 +253,14 @@ sds sdsmapchars(sds s, const char *from, const char *to, size_t setlen); ...@@ -253,6 +253,14 @@ sds sdsmapchars(sds s, const char *from, const char *to, size_t setlen);
sds sdsjoin(char **argv, int argc, char *sep); sds sdsjoin(char **argv, int argc, char *sep);
sds sdsjoinsds(sds *argv, int argc, const char *sep, size_t seplen); sds sdsjoinsds(sds *argv, int argc, const char *sep, size_t seplen);
/* Callback for sdstemplate. The function gets called by sdstemplate
* every time a variable needs to be expanded. The variable name is
* provided as variable, and the callback is expected to return a
* substitution value. Returning a NULL indicates an error.
*/
typedef sds (*sdstemplate_callback_t)(const sds variable, void *arg);
sds sdstemplate(const char *template, sdstemplate_callback_t cb_func, void *cb_arg);
/* Low level functions exposed to the user API */ /* Low level functions exposed to the user API */
sds sdsMakeRoomFor(sds s, size_t addlen); sds sdsMakeRoomFor(sds s, size_t addlen);
void sdsIncrLen(sds s, ssize_t incr); void sdsIncrLen(sds s, ssize_t incr);
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -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
This diff is collapsed.
This diff is collapsed.
...@@ -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) {
......
This diff is collapsed.
This diff is collapsed.
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