MCPcopy Create free account
hub / github.com/F-Stack/f-stack / rewriteStreamObject

Function rewriteStreamObject

app/redis-6.2.6/src/aof.c:1286–1418  ·  view source on GitHub ↗

Emit the commands needed to rebuild a stream object. * The function returns 0 on error, 1 on success. */

Source from the content-addressed store, hash-verified

1284/* Emit the commands needed to rebuild a stream object.
1285 * The function returns 0 on error, 1 on success. */
1286int 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) ||

Callers 1

rewriteAppendOnlyFileRioFunction · 0.85

Calls 15

streamIteratorStartFunction · 0.85
streamIteratorGetIDFunction · 0.85
rioWriteBulkCountFunction · 0.85
rioWriteBulkStringFunction · 0.85
rioWriteBulkObjectFunction · 0.85
rioWriteBulkStreamIDFunction · 0.85
streamIteratorStopFunction · 0.85
streamIteratorGetFieldFunction · 0.85
raxStartFunction · 0.85
raxSeekFunction · 0.85
raxNextFunction · 0.85
raxStopFunction · 0.85

Tested by

no test coverage detected