Emit the commands needed to rebuild a stream object. * The function returns 0 on error, 1 on success. */
| 1284 | /* Emit the commands needed to rebuild a stream object. |
| 1285 | * The function returns 0 on error, 1 on success. */ |
| 1286 | int rewriteStreamObject(rio *r, robj *key, robj *o) { |
| 1287 | stream *s = o->ptr; |
| 1288 | streamIterator si; |
| 1289 | streamIteratorStart(&si,s,NULL,NULL,0); |
| 1290 | streamID id; |
| 1291 | int64_t numfields; |
| 1292 | |
| 1293 | if (s->length) { |
| 1294 | /* Reconstruct the stream data using XADD commands. */ |
| 1295 | while(streamIteratorGetID(&si,&id,&numfields)) { |
| 1296 | /* Emit a two elements array for each item. The first is |
| 1297 | * the ID, the second is an array of field-value pairs. */ |
| 1298 | |
| 1299 | /* Emit the XADD <key> <id> ...fields... command. */ |
| 1300 | if (!rioWriteBulkCount(r,'*',3+numfields*2) || |
| 1301 | !rioWriteBulkString(r,"XADD",4) || |
| 1302 | !rioWriteBulkObject(r,key) || |
| 1303 | !rioWriteBulkStreamID(r,&id)) |
| 1304 | { |
| 1305 | streamIteratorStop(&si); |
| 1306 | return 0; |
| 1307 | } |
| 1308 | while(numfields--) { |
| 1309 | unsigned char *field, *value; |
| 1310 | int64_t field_len, value_len; |
| 1311 | streamIteratorGetField(&si,&field,&value,&field_len,&value_len); |
| 1312 | if (!rioWriteBulkString(r,(char*)field,field_len) || |
| 1313 | !rioWriteBulkString(r,(char*)value,value_len)) |
| 1314 | { |
| 1315 | streamIteratorStop(&si); |
| 1316 | return 0; |
| 1317 | } |
| 1318 | } |
| 1319 | } |
| 1320 | } else { |
| 1321 | /* Use the XADD MAXLEN 0 trick to generate an empty stream if |
| 1322 | * the key we are serializing is an empty string, which is possible |
| 1323 | * for the Stream type. */ |
| 1324 | id.ms = 0; id.seq = 1; |
| 1325 | if (!rioWriteBulkCount(r,'*',7) || |
| 1326 | !rioWriteBulkString(r,"XADD",4) || |
| 1327 | !rioWriteBulkObject(r,key) || |
| 1328 | !rioWriteBulkString(r,"MAXLEN",6) || |
| 1329 | !rioWriteBulkString(r,"0",1) || |
| 1330 | !rioWriteBulkStreamID(r,&id) || |
| 1331 | !rioWriteBulkString(r,"x",1) || |
| 1332 | !rioWriteBulkString(r,"y",1)) |
| 1333 | { |
| 1334 | streamIteratorStop(&si); |
| 1335 | return 0; |
| 1336 | } |
| 1337 | } |
| 1338 | |
| 1339 | /* Append XSETID after XADD, make sure lastid is correct, |
| 1340 | * in case of XDEL lastid. */ |
| 1341 | if (!rioWriteBulkCount(r,'*',3) || |
| 1342 | !rioWriteBulkString(r,"XSETID",6) || |
| 1343 | !rioWriteBulkObject(r,key) || |
no test coverage detected