| 1495 | boundaries, just the entries. */ |
| 1496 | #define STREAM_RWR_HISTORY (1<<2) /* Only serve consumer local PEL. */ |
| 1497 | size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count, int rev, streamCG *group, streamConsumer *consumer, int flags, streamPropInfo *spi) { |
| 1498 | void *arraylen_ptr = NULL; |
| 1499 | size_t arraylen = 0; |
| 1500 | streamIterator si; |
| 1501 | int64_t numfields; |
| 1502 | streamID id; |
| 1503 | int propagate_last_id = 0; |
| 1504 | int noack = flags & STREAM_RWR_NOACK; |
| 1505 | |
| 1506 | /* If the client is asking for some history, we serve it using a |
| 1507 | * different function, so that we return entries *solely* from its |
| 1508 | * own PEL. This ensures each consumer will always and only see |
| 1509 | * the history of messages delivered to it and not yet confirmed |
| 1510 | * as delivered. */ |
| 1511 | if (group && (flags & STREAM_RWR_HISTORY)) { |
| 1512 | return streamReplyWithRangeFromConsumerPEL(c,s,start,end,count, |
| 1513 | consumer); |
| 1514 | } |
| 1515 | |
| 1516 | if (!(flags & STREAM_RWR_RAWENTRIES)) |
| 1517 | arraylen_ptr = addReplyDeferredLen(c); |
| 1518 | streamIteratorStart(&si,s,start,end,rev); |
| 1519 | while(streamIteratorGetID(&si,&id,&numfields)) { |
| 1520 | /* Update the group last_id if needed. */ |
| 1521 | if (group && streamCompareID(&id,&group->last_id) > 0) { |
| 1522 | group->last_id = id; |
| 1523 | /* Group last ID should be propagated only if NOACK was |
| 1524 | * specified, otherwise the last id will be included |
| 1525 | * in the propagation of XCLAIM itself. */ |
| 1526 | if (noack) propagate_last_id = 1; |
| 1527 | } |
| 1528 | |
| 1529 | /* Emit a two elements array for each item. The first is |
| 1530 | * the ID, the second is an array of field-value pairs. */ |
| 1531 | addReplyArrayLen(c,2); |
| 1532 | addReplyStreamID(c,&id); |
| 1533 | |
| 1534 | addReplyArrayLen(c,numfields*2); |
| 1535 | |
| 1536 | /* Emit the field-value pairs. */ |
| 1537 | while(numfields--) { |
| 1538 | unsigned char *key, *value; |
| 1539 | int64_t key_len, value_len; |
| 1540 | streamIteratorGetField(&si,&key,&value,&key_len,&value_len); |
| 1541 | addReplyBulkCBuffer(c,key,key_len); |
| 1542 | addReplyBulkCBuffer(c,value,value_len); |
| 1543 | } |
| 1544 | |
| 1545 | /* If a group is passed, we need to create an entry in the |
| 1546 | * PEL (pending entries list) of this group *and* this consumer. |
| 1547 | * |
| 1548 | * Note that we cannot be sure about the fact the message is not |
| 1549 | * already owned by another consumer, because the admin is able |
| 1550 | * to change the consumer group last delivered ID using the |
| 1551 | * XGROUP SETID command. So if we find that there is already |
| 1552 | * a NACK for the entry, we need to associate it to the new |
| 1553 | * consumer. */ |
| 1554 | if (group && !noack) { |
no test coverage detected