Commit c1689166 authored by antirez's avatar antirez
Browse files

Streams: rewrite empty streams with XADD MAXLEN 0. Use XSETID.

Related to #5426.
parent ea78a1db
...@@ -1158,22 +1158,28 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) { ...@@ -1158,22 +1158,28 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) {
if (rioWriteBulkString(r,(char*)value,value_len) == 0) return 0; if (rioWriteBulkString(r,(char*)value,value_len) == 0) return 0;
} }
} }
/* Append XSTREAM SETID after XADD, make sure lastid is correct,
* in case of XDEL lastid. */
if (rioWriteBulkCount(r,'*',4) == 0) return 0;
if (rioWriteBulkString(r,"XSTREAM",7) == 0) return 0;
if (rioWriteBulkString(r,"SETID",5) == 0) return 0;
if (rioWriteBulkObject(r,key) == 0) return 0;
if (rioWriteBulkStreamID(r,&s->last_id) == 0) return 0;
} else { } else {
/* Using XSTREAM CREATE if the stream is empty. */ /* Use the XADD MAXLEN 0 trick to generate an empty stream if
if (rioWriteBulkCount(r,'*',4) == 0) return 0; * the key we are serializing is an empty string, which is possible
if (rioWriteBulkString(r,"XSTREAM",7) == 0) return 0; * for the Stream type. */
if (rioWriteBulkString(r,"CREATE",6) == 0) return 0; if (rioWriteBulkCount(r,'*',7) == 0) return 0;
if (rioWriteBulkString(r,"XADD",4) == 0) return 0;
if (rioWriteBulkObject(r,key) == 0) return 0; if (rioWriteBulkObject(r,key) == 0) return 0;
if (rioWriteBulkStreamID(r,&s->last_id) == 0) return 0; if (rioWriteBulkString(r,"MAXLEN",6) == 0) return 0;
if (rioWriteBulkString(r,"0",1) == 0) return 0;
if (rioWriteBulkString(r,"*",1) == 0) return 0;
if (rioWriteBulkString(r,"x",1) == 0) return 0;
if (rioWriteBulkString(r,"y",1) == 0) return 0;
} }
/* Append XSETID after XADD, make sure lastid is correct,
* in case of XDEL lastid. */
if (rioWriteBulkCount(r,'*',3) == 0) return 0;
if (rioWriteBulkString(r,"XSETID",6) == 0) return 0;
if (rioWriteBulkObject(r,key) == 0) return 0;
if (rioWriteBulkStreamID(r,&s->last_id) == 0) return 0;
/* Create all the stream consumer groups. */ /* Create all the stream consumer groups. */
if (s->cgroups) { if (s->cgroups) {
raxIterator ri; raxIterator ri;
......
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