Helper for rewriteStreamObject(): emit the XCLAIM needed in order to * add the message described by 'nack' having the id 'rawid', into the pending * list of the specified consumer. All this in the context of the specified * key and group. */
| 1346 | * list of the specified consumer. All this in the context of the specified |
| 1347 | * key and group. */ |
| 1348 | int rioWriteStreamPendingEntry(rio *r, robj *key, const char *groupname, size_t groupname_len, streamConsumer *consumer, unsigned char *rawid, streamNACK *nack) { |
| 1349 | /* XCLAIM <key> <group> <consumer> 0 <id> TIME <milliseconds-unix-time> |
| 1350 | RETRYCOUNT <count> JUSTID FORCE. */ |
| 1351 | streamID id; |
| 1352 | streamDecodeID(rawid,&id); |
| 1353 | if (rioWriteBulkCount(r,'*',12) == 0) return 0; |
| 1354 | if (rioWriteBulkString(r,"XCLAIM",6) == 0) return 0; |
| 1355 | if (rioWriteBulkObject(r,key) == 0) return 0; |
| 1356 | if (rioWriteBulkString(r,groupname,groupname_len) == 0) return 0; |
| 1357 | if (rioWriteBulkString(r,consumer->name,sdslen(consumer->name)) == 0) return 0; |
| 1358 | if (rioWriteBulkString(r,"0",1) == 0) return 0; |
| 1359 | if (rioWriteBulkStreamID(r,&id) == 0) return 0; |
| 1360 | if (rioWriteBulkString(r,"TIME",4) == 0) return 0; |
| 1361 | if (rioWriteBulkLongLong(r,nack->delivery_time) == 0) return 0; |
| 1362 | if (rioWriteBulkString(r,"RETRYCOUNT",10) == 0) return 0; |
| 1363 | if (rioWriteBulkLongLong(r,nack->delivery_count) == 0) return 0; |
| 1364 | if (rioWriteBulkString(r,"JUSTID",6) == 0) return 0; |
| 1365 | if (rioWriteBulkString(r,"FORCE",5) == 0) return 0; |
| 1366 | return 1; |
| 1367 | } |
| 1368 | |
| 1369 | /* Helper for rewriteStreamObject(): emit the XGROUP CREATECONSUMER is |
| 1370 | * needed in order to create consumers that do not have any pending entries. |
no test coverage detected