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. */
| 1247 | * list of the specified consumer. All this in the context of the specified |
| 1248 | * key and group. */ |
| 1249 | int rioWriteStreamPendingEntry(rio *r, robj *key, const char *groupname, size_t groupname_len, streamConsumer *consumer, unsigned char *rawid, streamNACK *nack) { |
| 1250 | /* XCLAIM <key> <group> <consumer> 0 <id> TIME <milliseconds-unix-time> |
| 1251 | RETRYCOUNT <count> JUSTID FORCE. */ |
| 1252 | streamID id; |
| 1253 | streamDecodeID(rawid,&id); |
| 1254 | if (rioWriteBulkCount(r,'*',12) == 0) return 0; |
| 1255 | if (rioWriteBulkString(r,"XCLAIM",6) == 0) return 0; |
| 1256 | if (rioWriteBulkObject(r,key) == 0) return 0; |
| 1257 | if (rioWriteBulkString(r,groupname,groupname_len) == 0) return 0; |
| 1258 | if (rioWriteBulkString(r,consumer->name,sdslen(consumer->name)) == 0) return 0; |
| 1259 | if (rioWriteBulkString(r,"0",1) == 0) return 0; |
| 1260 | if (rioWriteBulkStreamID(r,&id) == 0) return 0; |
| 1261 | if (rioWriteBulkString(r,"TIME",4) == 0) return 0; |
| 1262 | if (rioWriteBulkLongLong(r,nack->delivery_time) == 0) return 0; |
| 1263 | if (rioWriteBulkString(r,"RETRYCOUNT",10) == 0) return 0; |
| 1264 | if (rioWriteBulkLongLong(r,nack->delivery_count) == 0) return 0; |
| 1265 | if (rioWriteBulkString(r,"JUSTID",6) == 0) return 0; |
| 1266 | if (rioWriteBulkString(r,"FORCE",5) == 0) return 0; |
| 1267 | return 1; |
| 1268 | } |
| 1269 | |
| 1270 | /* Helper for rewriteStreamObject(): emit the XGROUP CREATECONSUMER is |
| 1271 | * needed in order to create consumers that do not have any pending entries. |
no test coverage detected