Commit b3e80d2f authored by zhaozhao.zz's avatar zhaozhao.zz
Browse files

Stream & AOF: rewrite stream in correct way

parent 5f3adbee
...@@ -1121,23 +1121,39 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) { ...@@ -1121,23 +1121,39 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) {
streamID id; streamID id;
int64_t numfields; int64_t numfields;
/* Reconstruct the stream data using XADD commands. */ if (s->length) {
while(streamIteratorGetID(&si,&id,&numfields)) { /* Reconstruct the stream data using XADD commands. */
/* Emit a two elements array for each item. The first is while(streamIteratorGetID(&si,&id,&numfields)) {
* the ID, the second is an array of field-value pairs. */ /* Emit a two elements array for each item. The first is
* the ID, the second is an array of field-value pairs. */
/* Emit the XADD <key> <id> ...fields... command. */
if (rioWriteBulkCount(r,'*',3+numfields*2) == 0) return 0; /* Emit the XADD <key> <id> ...fields... command. */
if (rioWriteBulkString(r,"XADD",4) == 0) return 0; if (rioWriteBulkCount(r,'*',3+numfields*2) == 0) return 0;
if (rioWriteBulkObject(r,key) == 0) return 0; if (rioWriteBulkString(r,"XADD",4) == 0) return 0;
if (rioWriteBulkStreamID(r,&id) == 0) return 0; if (rioWriteBulkObject(r,key) == 0) return 0;
while(numfields--) { if (rioWriteBulkStreamID(r,&id) == 0) return 0;
unsigned char *field, *value; while(numfields--) {
int64_t field_len, value_len; unsigned char *field, *value;
streamIteratorGetField(&si,&field,&value,&field_len,&value_len); int64_t field_len, value_len;
if (rioWriteBulkString(r,(char*)field,field_len) == 0) return 0; streamIteratorGetField(&si,&field,&value,&field_len,&value_len);
if (rioWriteBulkString(r,(char*)value,value_len) == 0) return 0; if (rioWriteBulkString(r,(char*)field,field_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 {
/* Using XSTREAM CREATE if the stream is empty. */
if (rioWriteBulkCount(r,'*',4) == 0) return 0;
if (rioWriteBulkString(r,"XSTREAM",7) == 0) return 0;
if (rioWriteBulkString(r,"CREATE",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. */
......
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