Adding upstream version 5:7.2.4.upstream/5%7.2.4upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 4038 insertions, 0 deletions

diff --git a/src/t_stream.c b/src/t_stream.c
new file mode 100644
index 0000000..5fcb631
--- /dev/null
+++ b/src/t_stream.c
@@ -0,0 +1,4038 @@
+/*
+ * Copyright (c) 2017, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "server.h"
+#include "endianconv.h"
+#include "stream.h"
+
+/* Every stream item inside the listpack, has a flags field that is used to
+ * mark the entry as deleted, or having the same field as the "master"
+ * entry at the start of the listpack> */
+#define STREAM_ITEM_FLAG_NONE 0             /* No special flags. */
+#define STREAM_ITEM_FLAG_DELETED (1<<0)     /* Entry is deleted. Skip it. */
+#define STREAM_ITEM_FLAG_SAMEFIELDS (1<<1)  /* Same fields as master entry. */
+
+/* For stream commands that require multiple IDs
+ * when the number of IDs is less than 'STREAMID_STATIC_VECTOR_LEN',
+ * avoid malloc allocation.*/
+#define STREAMID_STATIC_VECTOR_LEN 8
+
+/* Max pre-allocation for listpack. This is done to avoid abuse of a user
+ * setting stream_node_max_bytes to a huge number. */
+#define STREAM_LISTPACK_MAX_PRE_ALLOCATE 4096
+
+/* Don't let listpacks grow too big, even if the user config allows it.
+ * doing so can lead to an overflow (trying to store more than 32bit length
+ * into the listpack header), or actually an assertion since lpInsert
+ * will return NULL. */
+#define STREAM_LISTPACK_MAX_SIZE (1<<30)
+
+void streamFreeCG(streamCG *cg);
+void streamFreeNACK(streamNACK *na);
+size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamConsumer *consumer);
+int streamParseStrictIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int *seq_given);
+int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq);
+
+/* -----------------------------------------------------------------------
+ * Low level stream encoding: a radix tree of listpacks.
+ * ----------------------------------------------------------------------- */
+
+/* Create a new stream data structure. */
+stream *streamNew(void) {
+    stream *s = zmalloc(sizeof(*s));
+    s->rax = raxNew();
+    s->length = 0;
+    s->first_id.ms = 0;
+    s->first_id.seq = 0;
+    s->last_id.ms = 0;
+    s->last_id.seq = 0;
+    s->max_deleted_entry_id.seq = 0;
+    s->max_deleted_entry_id.ms = 0;
+    s->entries_added = 0;
+    s->cgroups = NULL; /* Created on demand to save memory when not used. */
+    return s;
+}
+
+/* Free a stream, including the listpacks stored inside the radix tree. */
+void freeStream(stream *s) {
+    raxFreeWithCallback(s->rax,(void(*)(void*))lpFree);
+    if (s->cgroups)
+        raxFreeWithCallback(s->cgroups,(void(*)(void*))streamFreeCG);
+    zfree(s);
+}
+
+/* Return the length of a stream. */
+unsigned long streamLength(const robj *subject) {
+    stream *s = subject->ptr;
+    return s->length;
+}
+
+/* Set 'id' to be its successor stream ID.
+ * If 'id' is the maximal possible id, it is wrapped around to 0-0 and a
+ * C_ERR is returned. */
+int streamIncrID(streamID *id) {
+    int ret = C_OK;
+    if (id->seq == UINT64_MAX) {
+        if (id->ms == UINT64_MAX) {
+            /* Special case where 'id' is the last possible streamID... */
+            id->ms = id->seq = 0;
+            ret = C_ERR;
+        } else {
+            id->ms++;
+            id->seq = 0;
+        }
+    } else {
+        id->seq++;
+    }
+    return ret;
+}
+
+/* Set 'id' to be its predecessor stream ID.
+ * If 'id' is the minimal possible id, it remains 0-0 and a C_ERR is
+ * returned. */
+int streamDecrID(streamID *id) {
+    int ret = C_OK;
+    if (id->seq == 0) {
+        if (id->ms == 0) {
+            /* Special case where 'id' is the first possible streamID... */
+            id->ms = id->seq = UINT64_MAX;
+            ret = C_ERR;
+        } else {
+            id->ms--;
+            id->seq = UINT64_MAX;
+        }
+    } else {
+        id->seq--;
+    }
+    return ret;
+}
+
+/* Generate the next stream item ID given the previous one. If the current
+ * milliseconds Unix time is greater than the previous one, just use this
+ * as time part and start with sequence part of zero. Otherwise we use the
+ * previous time (and never go backward) and increment the sequence. */
+void streamNextID(streamID *last_id, streamID *new_id) {
+    uint64_t ms = commandTimeSnapshot();
+    if (ms > last_id->ms) {
+        new_id->ms = ms;
+        new_id->seq = 0;
+    } else {
+        *new_id = *last_id;
+        streamIncrID(new_id);
+    }
+}
+
+/* This is a helper function for the COPY command.
+ * Duplicate a Stream object, with the guarantee that the returned object
+ * has the same encoding as the original one.
+ *
+ * The resulting object always has refcount set to 1 */
+robj *streamDup(robj *o) {
+    robj *sobj;
+
+    serverAssert(o->type == OBJ_STREAM);
+
+    switch (o->encoding) {
+        case OBJ_ENCODING_STREAM:
+            sobj = createStreamObject();
+            break;
+        default:
+            serverPanic("Wrong encoding.");
+            break;
+    }
+
+    stream *s;
+    stream *new_s;
+    s = o->ptr;
+    new_s = sobj->ptr;
+
+    raxIterator ri;
+    uint64_t rax_key[2];
+    raxStart(&ri, s->rax);
+    raxSeek(&ri, "^", NULL, 0);
+    size_t lp_bytes = 0;      /* Total bytes in the listpack. */
+    unsigned char *lp = NULL; /* listpack pointer. */
+    /* Get a reference to the listpack node. */
+    while (raxNext(&ri)) {
+        lp = ri.data;
+        lp_bytes = lpBytes(lp);
+        unsigned char *new_lp = zmalloc(lp_bytes);
+        memcpy(new_lp, lp, lp_bytes);
+        memcpy(rax_key, ri.key, sizeof(rax_key));
+        raxInsert(new_s->rax, (unsigned char *)&rax_key, sizeof(rax_key),
+                  new_lp, NULL);
+    }
+    new_s->length = s->length;
+    new_s->first_id = s->first_id;
+    new_s->last_id = s->last_id;
+    new_s->max_deleted_entry_id = s->max_deleted_entry_id;
+    new_s->entries_added = s->entries_added;
+    raxStop(&ri);
+
+    if (s->cgroups == NULL) return sobj;
+
+    /* Consumer Groups */
+    raxIterator ri_cgroups;
+    raxStart(&ri_cgroups, s->cgroups);
+    raxSeek(&ri_cgroups, "^", NULL, 0);
+    while (raxNext(&ri_cgroups)) {
+        streamCG *cg = ri_cgroups.data;
+        streamCG *new_cg = streamCreateCG(new_s, (char *)ri_cgroups.key,
+                                          ri_cgroups.key_len, &cg->last_id,
+                                          cg->entries_read);
+
+        serverAssert(new_cg != NULL);
+
+        /* Consumer Group PEL */
+        raxIterator ri_cg_pel;
+        raxStart(&ri_cg_pel,cg->pel);
+        raxSeek(&ri_cg_pel,"^",NULL,0);
+        while(raxNext(&ri_cg_pel)){
+            streamNACK *nack = ri_cg_pel.data;
+            streamNACK *new_nack = streamCreateNACK(NULL);
+            new_nack->delivery_time = nack->delivery_time;
+            new_nack->delivery_count = nack->delivery_count;
+            raxInsert(new_cg->pel, ri_cg_pel.key, sizeof(streamID), new_nack, NULL);
+        }
+        raxStop(&ri_cg_pel);
+
+        /* Consumers */
+        raxIterator ri_consumers;
+        raxStart(&ri_consumers, cg->consumers);
+        raxSeek(&ri_consumers, "^", NULL, 0);
+        while (raxNext(&ri_consumers)) {
+            streamConsumer *consumer = ri_consumers.data;
+            streamConsumer *new_consumer;
+            new_consumer = zmalloc(sizeof(*new_consumer));
+            new_consumer->name = sdsdup(consumer->name);
+            new_consumer->pel = raxNew();
+            raxInsert(new_cg->consumers,(unsigned char *)new_consumer->name,
+                        sdslen(new_consumer->name), new_consumer, NULL);
+            new_consumer->seen_time = consumer->seen_time;
+            new_consumer->active_time = consumer->active_time;
+
+            /* Consumer PEL */
+            raxIterator ri_cpel;
+            raxStart(&ri_cpel, consumer->pel);
+            raxSeek(&ri_cpel, "^", NULL, 0);
+            while (raxNext(&ri_cpel)) {
+                streamNACK *new_nack = raxFind(new_cg->pel,ri_cpel.key,sizeof(streamID));
+
+                serverAssert(new_nack != raxNotFound);
+
+                new_nack->consumer = new_consumer;
+                raxInsert(new_consumer->pel,ri_cpel.key,sizeof(streamID),new_nack,NULL);
+            }
+            raxStop(&ri_cpel);
+        }
+        raxStop(&ri_consumers);
+    }
+    raxStop(&ri_cgroups);
+    return sobj;
+}
+
+/* This is a wrapper function for lpGet() to directly get an integer value
+ * from the listpack (that may store numbers as a string), converting
+ * the string if needed.
+ * The 'valid" argument is an optional output parameter to get an indication
+ * if the record was valid, when this parameter is NULL, the function will
+ * fail with an assertion. */
+static inline int64_t lpGetIntegerIfValid(unsigned char *ele, int *valid) {
+    int64_t v;
+    unsigned char *e = lpGet(ele,&v,NULL);
+    if (e == NULL) {
+        if (valid)
+            *valid = 1;
+        return v;
+    }
+    /* The following code path should never be used for how listpacks work:
+     * they should always be able to store an int64_t value in integer
+     * encoded form. However the implementation may change. */
+    long long ll;
+    int ret = string2ll((char*)e,v,&ll);
+    if (valid)
+        *valid = ret;
+    else
+        serverAssert(ret != 0);
+    v = ll;
+    return v;
+}
+
+#define lpGetInteger(ele) lpGetIntegerIfValid(ele, NULL)
+
+/* Get an edge streamID of a given listpack.
+ * 'master_id' is an input param, used to build the 'edge_id' output param */
+int lpGetEdgeStreamID(unsigned char *lp, int first, streamID *master_id, streamID *edge_id)
+{
+   if (lp == NULL)
+       return 0;
+
+   unsigned char *lp_ele;
+
+   /* We need to seek either the first or the last entry depending
+    * on the direction of the iteration. */
+   if (first) {
+       /* Get the master fields count. */
+       lp_ele = lpFirst(lp);        /* Seek items count */
+       lp_ele = lpNext(lp, lp_ele); /* Seek deleted count. */
+       lp_ele = lpNext(lp, lp_ele); /* Seek num fields. */
+       int64_t master_fields_count = lpGetInteger(lp_ele);
+       lp_ele = lpNext(lp, lp_ele); /* Seek first field. */
+
+       /* If we are iterating in normal order, skip the master fields
+        * to seek the first actual entry. */
+       for (int64_t i = 0; i < master_fields_count; i++)
+           lp_ele = lpNext(lp, lp_ele);
+
+       /* If we are going forward, skip the previous entry's
+        * lp-count field (or in case of the master entry, the zero
+        * term field) */
+       lp_ele = lpNext(lp, lp_ele);
+       if (lp_ele == NULL)
+           return 0;
+   } else {
+       /* If we are iterating in reverse direction, just seek the
+        * last part of the last entry in the listpack (that is, the
+        * fields count). */
+       lp_ele = lpLast(lp);
+
+       /* If we are going backward, read the number of elements this
+        * entry is composed of, and jump backward N times to seek
+        * its start. */
+       int64_t lp_count = lpGetInteger(lp_ele);
+       if (lp_count == 0) /* We reached the master entry. */
+           return 0;
+
+       while (lp_count--)
+           lp_ele = lpPrev(lp, lp_ele);
+   }
+
+   lp_ele = lpNext(lp, lp_ele); /* Seek ID (lp_ele currently points to 'flags'). */
+
+   /* Get the ID: it is encoded as difference between the master
+    * ID and this entry ID. */
+   streamID id = *master_id;
+   id.ms += lpGetInteger(lp_ele);
+   lp_ele = lpNext(lp, lp_ele);
+   id.seq += lpGetInteger(lp_ele);
+   *edge_id = id;
+   return 1;
+}
+
+/* Debugging function to log the full content of a listpack. Useful
+ * for development and debugging. */
+void streamLogListpackContent(unsigned char *lp) {
+    unsigned char *p = lpFirst(lp);
+    while(p) {
+        unsigned char buf[LP_INTBUF_SIZE];
+        int64_t v;
+        unsigned char *ele = lpGet(p,&v,buf);
+        serverLog(LL_WARNING,"- [%d] '%.*s'", (int)v, (int)v, ele);
+        p = lpNext(lp,p);
+    }
+}
+
+/* Convert the specified stream entry ID as a 128 bit big endian number, so
+ * that the IDs can be sorted lexicographically. */
+void streamEncodeID(void *buf, streamID *id) {
+    uint64_t e[2];
+    e[0] = htonu64(id->ms);
+    e[1] = htonu64(id->seq);
+    memcpy(buf,e,sizeof(e));
+}
+
+/* This is the reverse of streamEncodeID(): the decoded ID will be stored
+ * in the 'id' structure passed by reference. The buffer 'buf' must point
+ * to a 128 bit big-endian encoded ID. */
+void streamDecodeID(void *buf, streamID *id) {
+    uint64_t e[2];
+    memcpy(e,buf,sizeof(e));
+    id->ms = ntohu64(e[0]);
+    id->seq = ntohu64(e[1]);
+}
+
+/* Compare two stream IDs. Return -1 if a < b, 0 if a == b, 1 if a > b. */
+int streamCompareID(streamID *a, streamID *b) {
+    if (a->ms > b->ms) return 1;
+    else if (a->ms < b->ms) return -1;
+    /* The ms part is the same. Check the sequence part. */
+    else if (a->seq > b->seq) return 1;
+    else if (a->seq < b->seq) return -1;
+    /* Everything is the same: IDs are equal. */
+    return 0;
+}
+
+/* Retrieves the ID of the stream edge entry. An edge is either the first or
+ * the last ID in the stream, and may be a tombstone. To filter out tombstones,
+ * set the'skip_tombstones' argument to 1. */
+void streamGetEdgeID(stream *s, int first, int skip_tombstones, streamID *edge_id)
+{
+    streamIterator si;
+    int64_t numfields;
+    streamIteratorStart(&si,s,NULL,NULL,!first);
+    si.skip_tombstones = skip_tombstones;
+    int found = streamIteratorGetID(&si,edge_id,&numfields);
+    if (!found) {
+        streamID min_id = {0, 0}, max_id = {UINT64_MAX, UINT64_MAX};
+        *edge_id = first ? max_id : min_id;
+    }
+    streamIteratorStop(&si);
+}
+
+/* Adds a new item into the stream 's' having the specified number of
+ * field-value pairs as specified in 'numfields' and stored into 'argv'.
+ * Returns the new entry ID populating the 'added_id' structure.
+ *
+ * If 'use_id' is not NULL, the ID is not auto-generated by the function,
+ * but instead the passed ID is used to add the new entry. In this case
+ * adding the entry may fail as specified later in this comment.
+ * 
+ * When 'use_id' is used alongside with a zero 'seq-given', the sequence
+ * part of the passed ID is ignored and the function will attempt to use an
+ * auto-generated sequence.
+ *
+ * The function returns C_OK if the item was added, this is always true
+ * if the ID was generated by the function. However the function may return
+ * C_ERR in several cases:
+ * 1. If an ID was given via 'use_id', but adding it failed since the
+ *    current top ID is greater or equal. errno will be set to EDOM.
+ * 2. If a size of a single element or the sum of the elements is too big to
+ *    be stored into the stream. errno will be set to ERANGE. */
+int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id, int seq_given) {
+
+    /* Generate the new entry ID. */
+    streamID id;
+    if (use_id) {
+        if (seq_given) {
+            id = *use_id;
+        } else {
+            /* The automatically generated sequence can be either zero (new
+             * timestamps) or the incremented sequence of the last ID. In the
+             * latter case, we need to prevent an overflow/advancing forward
+             * in time. */
+            if (s->last_id.ms == use_id->ms) {
+                if (s->last_id.seq == UINT64_MAX) {
+                    errno = EDOM;
+                    return C_ERR;
+                }
+                id = s->last_id;
+                id.seq++;
+            } else {
+                id = *use_id;
+            }
+        }
+    } else {
+        streamNextID(&s->last_id,&id);
+    }
+
+    /* Check that the new ID is greater than the last entry ID
+     * or return an error. Automatically generated IDs might
+     * overflow (and wrap-around) when incrementing the sequence
+       part. */
+    if (streamCompareID(&id,&s->last_id) <= 0) {
+        errno = EDOM;
+        return C_ERR;
+    }
+
+    /* Avoid overflow when trying to add an element to the stream (listpack
+     * can only host up to 32bit length strings, and also a total listpack size
+     * can't be bigger than 32bit length. */
+    size_t totelelen = 0;
+    for (int64_t i = 0; i < numfields*2; i++) {
+        sds ele = argv[i]->ptr;
+        totelelen += sdslen(ele);
+    }
+    if (totelelen > STREAM_LISTPACK_MAX_SIZE) {
+        errno = ERANGE;
+        return C_ERR;
+    }
+
+    /* Add the new entry. */
+    raxIterator ri;
+    raxStart(&ri,s->rax);
+    raxSeek(&ri,"$",NULL,0);
+
+    size_t lp_bytes = 0;        /* Total bytes in the tail listpack. */
+    unsigned char *lp = NULL;   /* Tail listpack pointer. */
+
+    if (!raxEOF(&ri)) {
+        /* Get a reference to the tail node listpack. */
+        lp = ri.data;
+        lp_bytes = lpBytes(lp);
+    }
+    raxStop(&ri);
+
+    /* We have to add the key into the radix tree in lexicographic order,
+     * to do so we consider the ID as a single 128 bit number written in
+     * big endian, so that the most significant bytes are the first ones. */
+    uint64_t rax_key[2];    /* Key in the radix tree containing the listpack.*/
+    streamID master_id;     /* ID of the master entry in the listpack. */
+
+    /* Create a new listpack and radix tree node if needed. Note that when
+     * a new listpack is created, we populate it with a "master entry". This
+     * is just a set of fields that is taken as references in order to compress
+     * the stream entries that we'll add inside the listpack.
+     *
+     * Note that while we use the first added entry fields to create
+     * the master entry, the first added entry is NOT represented in the master
+     * entry, which is a stand alone object. But of course, the first entry
+     * will compress well because it's used as reference.
+     *
+     * The master entry is composed like in the following example:
+     *
+     * +-------+---------+------------+---------+--/--+---------+---------+-+
+     * | count | deleted | num-fields | field_1 | field_2 | ... | field_N |0|
+     * +-------+---------+------------+---------+--/--+---------+---------+-+
+     *
+     * count and deleted just represent respectively the total number of
+     * entries inside the listpack that are valid, and marked as deleted
+     * (deleted flag in the entry flags set). So the total number of items
+     * actually inside the listpack (both deleted and not) is count+deleted.
+     *
+     * The real entries will be encoded with an ID that is just the
+     * millisecond and sequence difference compared to the key stored at
+     * the radix tree node containing the listpack (delta encoding), and
+     * if the fields of the entry are the same as the master entry fields, the
+     * entry flags will specify this fact and the entry fields and number
+     * of fields will be omitted (see later in the code of this function).
+     *
+     * The "0" entry at the end is the same as the 'lp-count' entry in the
+     * regular stream entries (see below), and marks the fact that there are
+     * no more entries, when we scan the stream from right to left. */
+
+    /* First of all, check if we can append to the current macro node or
+     * if we need to switch to the next one. 'lp' will be set to NULL if
+     * the current node is full. */
+    if (lp != NULL) {
+        int new_node = 0;
+        size_t node_max_bytes = server.stream_node_max_bytes;
+        if (node_max_bytes == 0 || node_max_bytes > STREAM_LISTPACK_MAX_SIZE)
+            node_max_bytes = STREAM_LISTPACK_MAX_SIZE;
+        if (lp_bytes + totelelen >= node_max_bytes) {
+            new_node = 1;
+        } else if (server.stream_node_max_entries) {
+            unsigned char *lp_ele = lpFirst(lp);
+            /* Count both live entries and deleted ones. */
+            int64_t count = lpGetInteger(lp_ele) + lpGetInteger(lpNext(lp,lp_ele));
+            if (count >= server.stream_node_max_entries) new_node = 1;
+        }
+
+        if (new_node) {
+            /* Shrink extra pre-allocated memory */
+            lp = lpShrinkToFit(lp);
+            if (ri.data != lp)
+                raxInsert(s->rax,ri.key,ri.key_len,lp,NULL);
+            lp = NULL;
+        }
+    }
+
+    int flags = STREAM_ITEM_FLAG_NONE;
+    if (lp == NULL) {
+        master_id = id;
+        streamEncodeID(rax_key,&id);
+        /* Create the listpack having the master entry ID and fields.
+         * Pre-allocate some bytes when creating listpack to avoid realloc on
+         * every XADD. Since listpack.c uses malloc_size, it'll grow in steps,
+         * and won't realloc on every XADD.
+         * When listpack reaches max number of entries, we'll shrink the
+         * allocation to fit the data. */
+        size_t prealloc = STREAM_LISTPACK_MAX_PRE_ALLOCATE;
+        if (server.stream_node_max_bytes > 0 && server.stream_node_max_bytes < prealloc) {
+            prealloc = server.stream_node_max_bytes;
+        }
+        lp = lpNew(prealloc);
+        lp = lpAppendInteger(lp,1); /* One item, the one we are adding. */
+        lp = lpAppendInteger(lp,0); /* Zero deleted so far. */
+        lp = lpAppendInteger(lp,numfields);
+        for (int64_t i = 0; i < numfields; i++) {
+            sds field = argv[i*2]->ptr;
+            lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
+        }
+        lp = lpAppendInteger(lp,0); /* Master entry zero terminator. */
+        raxInsert(s->rax,(unsigned char*)&rax_key,sizeof(rax_key),lp,NULL);
+        /* The first entry we insert, has obviously the same fields of the
+         * master entry. */
+        flags |= STREAM_ITEM_FLAG_SAMEFIELDS;
+    } else {
+        serverAssert(ri.key_len == sizeof(rax_key));
+        memcpy(rax_key,ri.key,sizeof(rax_key));
+
+        /* Read the master ID from the radix tree key. */
+        streamDecodeID(rax_key,&master_id);
+        unsigned char *lp_ele = lpFirst(lp);
+
+        /* Update count and skip the deleted fields. */
+        int64_t count = lpGetInteger(lp_ele);
+        lp = lpReplaceInteger(lp,&lp_ele,count+1);
+        lp_ele = lpNext(lp,lp_ele); /* seek deleted. */
+        lp_ele = lpNext(lp,lp_ele); /* seek master entry num fields. */
+
+        /* Check if the entry we are adding, have the same fields
+         * as the master entry. */
+        int64_t master_fields_count = lpGetInteger(lp_ele);
+        lp_ele = lpNext(lp,lp_ele);
+        if (numfields == master_fields_count) {
+            int64_t i;
+            for (i = 0; i < master_fields_count; i++) {
+                sds field = argv[i*2]->ptr;
+                int64_t e_len;
+                unsigned char buf[LP_INTBUF_SIZE];
+                unsigned char *e = lpGet(lp_ele,&e_len,buf);
+                /* Stop if there is a mismatch. */
+                if (sdslen(field) != (size_t)e_len ||
+                    memcmp(e,field,e_len) != 0) break;
+                lp_ele = lpNext(lp,lp_ele);
+            }
+            /* All fields are the same! We can compress the field names
+             * setting a single bit in the flags. */
+            if (i == master_fields_count) flags |= STREAM_ITEM_FLAG_SAMEFIELDS;
+        }
+    }
+
+    /* Populate the listpack with the new entry. We use the following
+     * encoding:
+     *
+     * +-----+--------+----------+-------+-------+-/-+-------+-------+--------+
+     * |flags|entry-id|num-fields|field-1|value-1|...|field-N|value-N|lp-count|
+     * +-----+--------+----------+-------+-------+-/-+-------+-------+--------+
+     *
+     * However if the SAMEFIELD flag is set, we have just to populate
+     * the entry with the values, so it becomes:
+     *
+     * +-----+--------+-------+-/-+-------+--------+
+     * |flags|entry-id|value-1|...|value-N|lp-count|
+     * +-----+--------+-------+-/-+-------+--------+
+     *
+     * The entry-id field is actually two separated fields: the ms
+     * and seq difference compared to the master entry.
+     *
+     * The lp-count field is a number that states the number of listpack pieces
+     * that compose the entry, so that it's possible to travel the entry
+     * in reverse order: we can just start from the end of the listpack, read
+     * the entry, and jump back N times to seek the "flags" field to read
+     * the stream full entry. */
+    lp = lpAppendInteger(lp,flags);
+    lp = lpAppendInteger(lp,id.ms - master_id.ms);
+    lp = lpAppendInteger(lp,id.seq - master_id.seq);
+    if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
+        lp = lpAppendInteger(lp,numfields);
+    for (int64_t i = 0; i < numfields; i++) {
+        sds field = argv[i*2]->ptr, value = argv[i*2+1]->ptr;
+        if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
+            lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
+        lp = lpAppend(lp,(unsigned char*)value,sdslen(value));
+    }
+    /* Compute and store the lp-count field. */
+    int64_t lp_count = numfields;
+    lp_count += 3; /* Add the 3 fixed fields flags + ms-diff + seq-diff. */
+    if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) {
+        /* If the item is not compressed, it also has the fields other than
+         * the values, and an additional num-fields field. */
+        lp_count += numfields+1;
+    }
+    lp = lpAppendInteger(lp,lp_count);
+
+    /* Insert back into the tree in order to update the listpack pointer. */
+    if (ri.data != lp)
+        raxInsert(s->rax,(unsigned char*)&rax_key,sizeof(rax_key),lp,NULL);
+    s->length++;
+    s->entries_added++;
+    s->last_id = id;
+    if (s->length == 1) s->first_id = id;
+    if (added_id) *added_id = id;
+    return C_OK;
+}
+
+typedef struct {
+    /* XADD options */
+    streamID id; /* User-provided ID, for XADD only. */
+    int id_given; /* Was an ID different than "*" specified? for XADD only. */
+    int seq_given; /* Was an ID different than "ms-*" specified? for XADD only. */
+    int no_mkstream; /* if set to 1 do not create new stream */
+
+    /* XADD + XTRIM common options */
+    int trim_strategy; /* TRIM_STRATEGY_* */
+    int trim_strategy_arg_idx; /* Index of the count in MAXLEN/MINID, for rewriting. */
+    int approx_trim; /* If 1 only delete whole radix tree nodes, so
+                      * the trim argument is not applied verbatim. */
+    long long limit; /* Maximum amount of entries to trim. If 0, no limitation
+                      * on the amount of trimming work is enforced. */
+    /* TRIM_STRATEGY_MAXLEN options */
+    long long maxlen; /* After trimming, leave stream at this length . */
+    /* TRIM_STRATEGY_MINID options */
+    streamID minid; /* Trim by ID (No stream entries with ID < 'minid' will remain) */
+} streamAddTrimArgs;
+
+#define TRIM_STRATEGY_NONE 0
+#define TRIM_STRATEGY_MAXLEN 1
+#define TRIM_STRATEGY_MINID 2
+
+/* Trim the stream 's' according to args->trim_strategy, and return the
+ * number of elements removed from the stream. The 'approx' option, if non-zero,
+ * specifies that the trimming must be performed in a approximated way in
+ * order to maximize performances. This means that the stream may contain
+ * entries with IDs < 'id' in case of MINID (or more elements than 'maxlen'
+ * in case of MAXLEN), and elements are only removed if we can remove
+ * a *whole* node of the radix tree. The elements are removed from the head
+ * of the stream (older elements).
+ *
+ * The function may return zero if:
+ *
+ * 1) The minimal entry ID of the stream is already < 'id' (MINID); or
+ * 2) The stream is already shorter or equal to the specified max length (MAXLEN); or
+ * 3) The 'approx' option is true and the head node did not have enough elements
+ *    to be deleted.
+ *
+ * args->limit is the maximum number of entries to delete. The purpose is to
+ * prevent this function from taking to long.
+ * If 'limit' is 0 then we do not limit the number of deleted entries.
+ * Much like the 'approx', if 'limit' is smaller than the number of entries
+ * that should be trimmed, there is a chance we will still have entries with
+ * IDs < 'id' (or number of elements >= maxlen in case of MAXLEN).
+ */
+int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
+    size_t maxlen = args->maxlen;
+    streamID *id = &args->minid;
+    int approx = args->approx_trim;
+    int64_t limit = args->limit;
+    int trim_strategy = args->trim_strategy;
+
+    if (trim_strategy == TRIM_STRATEGY_NONE)
+        return 0;
+
+    raxIterator ri;
+    raxStart(&ri,s->rax);
+    raxSeek(&ri,"^",NULL,0);
+
+    int64_t deleted = 0;
+    while (raxNext(&ri)) {
+        if (trim_strategy == TRIM_STRATEGY_MAXLEN && s->length <= maxlen)
+            break;
+
+        unsigned char *lp = ri.data, *p = lpFirst(lp);
+        int64_t entries = lpGetInteger(p);
+
+        /* Check if we exceeded the amount of work we could do */
+        if (limit && (deleted + entries) > limit)
+            break;
+
+        /* Check if we can remove the whole node. */
+        int remove_node;
+        streamID master_id = {0}; /* For MINID */
+        if (trim_strategy == TRIM_STRATEGY_MAXLEN) {
+            remove_node = s->length - entries >= maxlen;
+        } else {
+            /* Read the master ID from the radix tree key. */
+            streamDecodeID(ri.key, &master_id);
+
+            /* Read last ID. */
+            streamID last_id = {0,0};
+            lpGetEdgeStreamID(lp, 0, &master_id, &last_id);
+
+            /* We can remove the entire node id its last ID < 'id' */
+            remove_node = streamCompareID(&last_id, id) < 0;
+        }
+
+        if (remove_node) {
+            lpFree(lp);
+            raxRemove(s->rax,ri.key,ri.key_len,NULL);
+            raxSeek(&ri,">=",ri.key,ri.key_len);
+            s->length -= entries;
+            deleted += entries;
+            continue;
+        }
+
+        /* If we cannot remove a whole element, and approx is true,
+         * stop here. */
+        if (approx) break;
+
+        /* Now we have to trim entries from within 'lp' */
+        int64_t deleted_from_lp = 0;
+
+        p = lpNext(lp, p); /* Skip deleted field. */
+        p = lpNext(lp, p); /* Skip num-of-fields in the master entry. */
+
+        /* Skip all the master fields. */
+        int64_t master_fields_count = lpGetInteger(p);
+        p = lpNext(lp,p); /* Skip the first field. */
+        for (int64_t j = 0; j < master_fields_count; j++)
+            p = lpNext(lp,p); /* Skip all master fields. */
+        p = lpNext(lp,p); /* Skip the zero master entry terminator. */
+
+        /* 'p' is now pointing to the first entry inside the listpack.
+         * We have to run entry after entry, marking entries as deleted
+         * if they are already not deleted. */
+        while (p) {
+            /* We keep a copy of p (which point to flags part) in order to
+             * update it after (and if) we actually remove the entry */
+            unsigned char *pcopy = p;
+
+            int64_t flags = lpGetInteger(p);
+            p = lpNext(lp, p); /* Skip flags. */
+            int64_t to_skip;
+
+            int64_t ms_delta = lpGetInteger(p);
+            p = lpNext(lp, p); /* Skip ID ms delta */
+            int64_t seq_delta = lpGetInteger(p);
+            p = lpNext(lp, p); /* Skip ID seq delta */
+
+            streamID currid = {0}; /* For MINID */
+            if (trim_strategy == TRIM_STRATEGY_MINID) {
+                currid.ms = master_id.ms + ms_delta;
+                currid.seq = master_id.seq + seq_delta;
+            }
+
+            int stop;
+            if (trim_strategy == TRIM_STRATEGY_MAXLEN) {
+                stop = s->length <= maxlen;
+            } else {
+                /* Following IDs will definitely be greater because the rax
+                 * tree is sorted, no point of continuing. */
+                stop = streamCompareID(&currid, id) >= 0;
+            }
+            if (stop)
+                break;
+
+            if (flags & STREAM_ITEM_FLAG_SAMEFIELDS) {
+                to_skip = master_fields_count;
+            } else {
+                to_skip = lpGetInteger(p); /* Get num-fields. */
+                p = lpNext(lp,p); /* Skip num-fields. */
+                to_skip *= 2; /* Fields and values. */
+            }
+
+            while(to_skip--) p = lpNext(lp,p); /* Skip the whole entry. */
+            p = lpNext(lp,p); /* Skip the final lp-count field. */
+
+            /* Mark the entry as deleted. */
+            if (!(flags & STREAM_ITEM_FLAG_DELETED)) {
+                intptr_t delta = p - lp;
+                flags |= STREAM_ITEM_FLAG_DELETED;
+                lp = lpReplaceInteger(lp, &pcopy, flags);
+                deleted_from_lp++;
+                s->length--;
+                p = lp + delta;
+            }
+        }
+        deleted += deleted_from_lp;
+
+        /* Now we update the entries/deleted counters. */
+        p = lpFirst(lp);
+        lp = lpReplaceInteger(lp,&p,entries-deleted_from_lp);
+        p = lpNext(lp,p); /* Skip deleted field. */
+        int64_t marked_deleted = lpGetInteger(p);
+        lp = lpReplaceInteger(lp,&p,marked_deleted+deleted_from_lp);
+        p = lpNext(lp,p); /* Skip num-of-fields in the master entry. */
+
+        /* Here we should perform garbage collection in case at this point
+         * there are too many entries deleted inside the listpack. */
+        entries -= deleted_from_lp;
+        marked_deleted += deleted_from_lp;
+        if (entries + marked_deleted > 10 && marked_deleted > entries/2) {
+            /* TODO: perform a garbage collection. */
+        }
+
+        /* Update the listpack with the new pointer. */
+        raxInsert(s->rax,ri.key,ri.key_len,lp,NULL);
+
+        break; /* If we are here, there was enough to delete in the current
+                  node, so no need to go to the next node. */
+    }
+    raxStop(&ri);
+
+    /* Update the stream's first ID after the trimming. */
+    if (s->length == 0) {
+        s->first_id.ms = 0;
+        s->first_id.seq = 0;
+    } else if (deleted) {
+        streamGetEdgeID(s,1,1,&s->first_id);
+    }
+
+    return deleted;
+}
+
+/* Trims a stream by length. Returns the number of deleted items. */
+int64_t streamTrimByLength(stream *s, long long maxlen, int approx) {
+    streamAddTrimArgs args = {
+        .trim_strategy = TRIM_STRATEGY_MAXLEN,
+        .approx_trim = approx,
+        .limit = approx ? 100 * server.stream_node_max_entries : 0,
+        .maxlen = maxlen
+    };
+    return streamTrim(s, &args);
+}
+
+/* Trims a stream by minimum ID. Returns the number of deleted items. */
+int64_t streamTrimByID(stream *s, streamID minid, int approx) {
+    streamAddTrimArgs args = {
+        .trim_strategy = TRIM_STRATEGY_MINID,
+        .approx_trim = approx,
+        .limit = approx ? 100 * server.stream_node_max_entries : 0,
+        .minid = minid
+    };
+    return streamTrim(s, &args);
+}
+
+/* Parse the arguments of XADD/XTRIM.
+ *
+ * See streamAddTrimArgs for more details about the arguments handled.
+ *
+ * This function returns the position of the ID argument (relevant only to XADD).
+ * On error -1 is returned and a reply is sent. */
+static int streamParseAddOrTrimArgsOrReply(client *c, streamAddTrimArgs *args, int xadd) {
+    /* Initialize arguments to defaults */
+    memset(args, 0, sizeof(*args));
+
+    /* Parse options. */
+    int i = 2; /* This is the first argument position where we could
+                  find an option, or the ID. */
+    int limit_given = 0;
+    for (; i < c->argc; i++) {
+        int moreargs = (c->argc-1) - i; /* Number of additional arguments. */
+        char *opt = c->argv[i]->ptr;
+        if (xadd && opt[0] == '*' && opt[1] == '\0') {
+            /* This is just a fast path for the common case of auto-ID
+             * creation. */
+            break;
+        } else if (!strcasecmp(opt,"maxlen") && moreargs) {
+            if (args->trim_strategy != TRIM_STRATEGY_NONE) {
+                addReplyError(c,"syntax error, MAXLEN and MINID options at the same time are not compatible");
+                return -1;
+            }
+            args->approx_trim = 0;
+            char *next = c->argv[i+1]->ptr;
+            /* Check for the form MAXLEN ~ <count>. */
+            if (moreargs >= 2 && next[0] == '~' && next[1] == '\0') {
+                args->approx_trim = 1;
+                i++;
+            } else if (moreargs >= 2 && next[0] == '=' && next[1] == '\0') {
+                i++;
+            }
+            if (getLongLongFromObjectOrReply(c,c->argv[i+1],&args->maxlen,NULL)
+                != C_OK) return -1;
+
+            if (args->maxlen < 0) {
+                addReplyError(c,"The MAXLEN argument must be >= 0.");
+                return -1;
+            }
+            i++;
+            args->trim_strategy = TRIM_STRATEGY_MAXLEN;
+            args->trim_strategy_arg_idx = i;
+        } else if (!strcasecmp(opt,"minid") && moreargs) {
+            if (args->trim_strategy != TRIM_STRATEGY_NONE) {
+                addReplyError(c,"syntax error, MAXLEN and MINID options at the same time are not compatible");
+                return -1;
+            }
+            args->approx_trim = 0;
+            char *next = c->argv[i+1]->ptr;
+            /* Check for the form MINID ~ <id> */
+            if (moreargs >= 2 && next[0] == '~' && next[1] == '\0') {
+                args->approx_trim = 1;
+                i++;
+            } else if (moreargs >= 2 && next[0] == '=' && next[1] == '\0') {
+                i++;
+            }
+
+            if (streamParseStrictIDOrReply(c,c->argv[i+1],&args->minid,0,NULL) != C_OK)
+                return -1;
+            
+            i++;
+            args->trim_strategy = TRIM_STRATEGY_MINID;
+            args->trim_strategy_arg_idx = i;
+        } else if (!strcasecmp(opt,"limit") && moreargs) {
+            /* Note about LIMIT: If it was not provided by the caller we set
+             * it to 100*server.stream_node_max_entries, and that's to prevent the
+             * trimming from taking too long, on the expense of not deleting entries
+             * that should be trimmed.
+             * If user wanted exact trimming (i.e. no '~') we never limit the number
+             * of trimmed entries */
+            if (getLongLongFromObjectOrReply(c,c->argv[i+1],&args->limit,NULL) != C_OK)
+                return -1;
+
+            if (args->limit < 0) {
+                addReplyError(c,"The LIMIT argument must be >= 0.");
+                return -1;
+            }
+            limit_given = 1;
+            i++;
+        } else if (xadd && !strcasecmp(opt,"nomkstream")) {
+            args->no_mkstream = 1;
+        } else if (xadd) {
+            /* If we are here is a syntax error or a valid ID. */
+            if (streamParseStrictIDOrReply(c,c->argv[i],&args->id,0,&args->seq_given) != C_OK)
+                return -1;
+            args->id_given = 1;
+            break;
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return -1;
+        }
+    }
+
+    if (args->limit && args->trim_strategy == TRIM_STRATEGY_NONE) {
+        addReplyError(c,"syntax error, LIMIT cannot be used without specifying a trimming strategy");
+        return -1;
+    }
+
+    if (!xadd && args->trim_strategy == TRIM_STRATEGY_NONE) {
+        addReplyError(c,"syntax error, XTRIM must be called with a trimming strategy");
+        return -1;
+    }
+
+    if (mustObeyClient(c)) {
+        /* If command came from master or from AOF we must not enforce maxnodes
+         * (The maxlen/minid argument was re-written to make sure there's no
+         * inconsistency). */
+        args->limit = 0;
+    } else {
+        /* We need to set the limit (only if we got '~') */
+        if (limit_given) {
+            if (!args->approx_trim) {
+                /* LIMIT was provided without ~ */
+                addReplyError(c,"syntax error, LIMIT cannot be used without the special ~ option");
+                return -1;
+            }
+        } else {
+            /* User didn't provide LIMIT, we must set it. */
+            if (args->approx_trim) {
+                /* In order to prevent from trimming to do too much work and 
+                 * cause latency spikes we limit the amount of work it can do.
+                 * We have to cap args->limit from both sides in case 
+                 * stream_node_max_entries is 0 or too big (could cause overflow)
+                 */
+                args->limit = 100 * server.stream_node_max_entries; /* Maximum 100 rax nodes. */
+                if (args->limit <= 0) args->limit = 10000;
+                if (args->limit > 1000000) args->limit = 1000000;
+            } else {
+                /* No LIMIT for exact trimming */
+                args->limit = 0;
+            }
+        }
+    }
+
+    return i;
+}
+
+/* Initialize the stream iterator, so that we can call iterating functions
+ * to get the next items. This requires a corresponding streamIteratorStop()
+ * at the end. The 'rev' parameter controls the direction. If it's zero the
+ * iteration is from the start to the end element (inclusive), otherwise
+ * if rev is non-zero, the iteration is reversed.
+ *
+ * Once the iterator is initialized, we iterate like this:
+ *
+ *  streamIterator myiterator;
+ *  streamIteratorStart(&myiterator,...);
+ *  int64_t numfields;
+ *  while(streamIteratorGetID(&myiterator,&ID,&numfields)) {
+ *      while(numfields--) {
+ *          unsigned char *key, *value;
+ *          size_t key_len, value_len;
+ *          streamIteratorGetField(&myiterator,&key,&value,&key_len,&value_len);
+ *
+ *          ... do what you want with key and value ...
+ *      }
+ *  }
+ *  streamIteratorStop(&myiterator); */
+void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev) {
+    /* Initialize the iterator and translates the iteration start/stop
+     * elements into a 128 big big-endian number. */
+    if (start) {
+        streamEncodeID(si->start_key,start);
+    } else {
+        si->start_key[0] = 0;
+        si->start_key[1] = 0;
+    }
+
+    if (end) {
+        streamEncodeID(si->end_key,end);
+    } else {
+        si->end_key[0] = UINT64_MAX;
+        si->end_key[1] = UINT64_MAX;
+    }
+
+    /* Seek the correct node in the radix tree. */
+    raxStart(&si->ri,s->rax);
+    if (!rev) {
+        if (start && (start->ms || start->seq)) {
+            raxSeek(&si->ri,"<=",(unsigned char*)si->start_key,
+                    sizeof(si->start_key));
+            if (raxEOF(&si->ri)) raxSeek(&si->ri,"^",NULL,0);
+        } else {
+            raxSeek(&si->ri,"^",NULL,0);
+        }
+    } else {
+        if (end && (end->ms || end->seq)) {
+            raxSeek(&si->ri,"<=",(unsigned char*)si->end_key,
+                    sizeof(si->end_key));
+            if (raxEOF(&si->ri)) raxSeek(&si->ri,"$",NULL,0);
+        } else {
+            raxSeek(&si->ri,"$",NULL,0);
+        }
+    }
+    si->stream = s;
+    si->lp = NULL;     /* There is no current listpack right now. */
+    si->lp_ele = NULL; /* Current listpack cursor. */
+    si->rev = rev;     /* Direction, if non-zero reversed, from end to start. */
+    si->skip_tombstones = 1;    /* By default tombstones aren't emitted. */
+}
+
+/* Return 1 and store the current item ID at 'id' if there are still
+ * elements within the iteration range, otherwise return 0 in order to
+ * signal the iteration terminated. */
+int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
+    while(1) { /* Will stop when element > stop_key or end of radix tree. */
+        /* If the current listpack is set to NULL, this is the start of the
+         * iteration or the previous listpack was completely iterated.
+         * Go to the next node. */
+        if (si->lp == NULL || si->lp_ele == NULL) {
+            if (!si->rev && !raxNext(&si->ri)) return 0;
+            else if (si->rev && !raxPrev(&si->ri)) return 0;
+            serverAssert(si->ri.key_len == sizeof(streamID));
+            /* Get the master ID. */
+            streamDecodeID(si->ri.key,&si->master_id);
+            /* Get the master fields count. */
+            si->lp = si->ri.data;
+            si->lp_ele = lpFirst(si->lp);           /* Seek items count */
+            si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek deleted count. */
+            si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek num fields. */
+            si->master_fields_count = lpGetInteger(si->lp_ele);
+            si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek first field. */
+            si->master_fields_start = si->lp_ele;
+            /* We are now pointing to the first field of the master entry.
+             * We need to seek either the first or the last entry depending
+             * on the direction of the iteration. */
+            if (!si->rev) {
+                /* If we are iterating in normal order, skip the master fields
+                 * to seek the first actual entry. */
+                for (uint64_t i = 0; i < si->master_fields_count; i++)
+                    si->lp_ele = lpNext(si->lp,si->lp_ele);
+            } else {
+                /* If we are iterating in reverse direction, just seek the
+                 * last part of the last entry in the listpack (that is, the
+                 * fields count). */
+                si->lp_ele = lpLast(si->lp);
+            }
+        } else if (si->rev) {
+            /* If we are iterating in the reverse order, and this is not
+             * the first entry emitted for this listpack, then we already
+             * emitted the current entry, and have to go back to the previous
+             * one. */
+            int64_t lp_count = lpGetInteger(si->lp_ele);
+            while(lp_count--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
+            /* Seek lp-count of prev entry. */
+            si->lp_ele = lpPrev(si->lp,si->lp_ele);
+        }
+
+        /* For every radix tree node, iterate the corresponding listpack,
+         * returning elements when they are within range. */
+        while(1) {
+            if (!si->rev) {
+                /* If we are going forward, skip the previous entry
+                 * lp-count field (or in case of the master entry, the zero
+                 * term field) */
+                si->lp_ele = lpNext(si->lp,si->lp_ele);
+                if (si->lp_ele == NULL) break;
+            } else {
+                /* If we are going backward, read the number of elements this
+                 * entry is composed of, and jump backward N times to seek
+                 * its start. */
+                int64_t lp_count = lpGetInteger(si->lp_ele);
+                if (lp_count == 0) { /* We reached the master entry. */
+                    si->lp = NULL;
+                    si->lp_ele = NULL;
+                    break;
+                }
+                while(lp_count--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
+            }
+
+            /* Get the flags entry. */
+            si->lp_flags = si->lp_ele;
+            int64_t flags = lpGetInteger(si->lp_ele);
+            si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek ID. */
+
+            /* Get the ID: it is encoded as difference between the master
+             * ID and this entry ID. */
+            *id = si->master_id;
+            id->ms += lpGetInteger(si->lp_ele);
+            si->lp_ele = lpNext(si->lp,si->lp_ele);
+            id->seq += lpGetInteger(si->lp_ele);
+            si->lp_ele = lpNext(si->lp,si->lp_ele);
+            unsigned char buf[sizeof(streamID)];
+            streamEncodeID(buf,id);
+
+            /* The number of entries is here or not depending on the
+             * flags. */
+            if (flags & STREAM_ITEM_FLAG_SAMEFIELDS) {
+                *numfields = si->master_fields_count;
+            } else {
+                *numfields = lpGetInteger(si->lp_ele);
+                si->lp_ele = lpNext(si->lp,si->lp_ele);
+            }
+            serverAssert(*numfields>=0);
+
+            /* If current >= start, and the entry is not marked as
+             * deleted or tombstones are included, emit it. */
+            if (!si->rev) {
+                if (memcmp(buf,si->start_key,sizeof(streamID)) >= 0 &&
+                    (!si->skip_tombstones || !(flags & STREAM_ITEM_FLAG_DELETED)))
+                {
+                    if (memcmp(buf,si->end_key,sizeof(streamID)) > 0)
+                        return 0; /* We are already out of range. */
+                    si->entry_flags = flags;
+                    if (flags & STREAM_ITEM_FLAG_SAMEFIELDS)
+                        si->master_fields_ptr = si->master_fields_start;
+                    return 1; /* Valid item returned. */
+                }
+            } else {
+                if (memcmp(buf,si->end_key,sizeof(streamID)) <= 0 &&
+                    (!si->skip_tombstones || !(flags & STREAM_ITEM_FLAG_DELETED)))
+                {
+                    if (memcmp(buf,si->start_key,sizeof(streamID)) < 0)
+                        return 0; /* We are already out of range. */
+                    si->entry_flags = flags;
+                    if (flags & STREAM_ITEM_FLAG_SAMEFIELDS)
+                        si->master_fields_ptr = si->master_fields_start;
+                    return 1; /* Valid item returned. */
+                }
+            }
+
+            /* If we do not emit, we have to discard if we are going
+             * forward, or seek the previous entry if we are going
+             * backward. */
+            if (!si->rev) {
+                int64_t to_discard = (flags & STREAM_ITEM_FLAG_SAMEFIELDS) ?
+                                      *numfields : *numfields*2;
+                for (int64_t i = 0; i < to_discard; i++)
+                    si->lp_ele = lpNext(si->lp,si->lp_ele);
+            } else {
+                int64_t prev_times = 4; /* flag + id ms + id seq + one more to
+                                           go back to the previous entry "count"
+                                           field. */
+                /* If the entry was not flagged SAMEFIELD we also read the
+                 * number of fields, so go back one more. */
+                if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) prev_times++;
+                while(prev_times--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
+            }
+        }
+
+        /* End of listpack reached. Try the next/prev radix tree node. */
+    }
+}
+
+/* Get the field and value of the current item we are iterating. This should
+ * be called immediately after streamIteratorGetID(), and for each field
+ * according to the number of fields returned by streamIteratorGetID().
+ * The function populates the field and value pointers and the corresponding
+ * lengths by reference, that are valid until the next iterator call, assuming
+ * no one touches the stream meanwhile. */
+void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsigned char **valueptr, int64_t *fieldlen, int64_t *valuelen) {
+    if (si->entry_flags & STREAM_ITEM_FLAG_SAMEFIELDS) {
+        *fieldptr = lpGet(si->master_fields_ptr,fieldlen,si->field_buf);
+        si->master_fields_ptr = lpNext(si->lp,si->master_fields_ptr);
+    } else {
+        *fieldptr = lpGet(si->lp_ele,fieldlen,si->field_buf);
+        si->lp_ele = lpNext(si->lp,si->lp_ele);
+    }
+    *valueptr = lpGet(si->lp_ele,valuelen,si->value_buf);
+    si->lp_ele = lpNext(si->lp,si->lp_ele);
+}
+
+/* Remove the current entry from the stream: can be called after the
+ * GetID() API or after any GetField() call, however we need to iterate
+ * a valid entry while calling this function. Moreover the function
+ * requires the entry ID we are currently iterating, that was previously
+ * returned by GetID().
+ *
+ * Note that after calling this function, next calls to GetField() can't
+ * be performed: the entry is now deleted. Instead the iterator will
+ * automatically re-seek to the next entry, so the caller should continue
+ * with GetID(). */
+void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
+    unsigned char *lp = si->lp;
+    int64_t aux;
+
+    /* We do not really delete the entry here. Instead we mark it as
+     * deleted by flagging it, and also incrementing the count of the
+     * deleted entries in the listpack header.
+     *
+     * We start flagging: */
+    int64_t flags = lpGetInteger(si->lp_flags);
+    flags |= STREAM_ITEM_FLAG_DELETED;
+    lp = lpReplaceInteger(lp,&si->lp_flags,flags);
+
+    /* Change the valid/deleted entries count in the master entry. */
+    unsigned char *p = lpFirst(lp);
+    aux = lpGetInteger(p);
+
+    if (aux == 1) {
+        /* If this is the last element in the listpack, we can remove the whole
+         * node. */
+        lpFree(lp);
+        raxRemove(si->stream->rax,si->ri.key,si->ri.key_len,NULL);
+    } else {
+        /* In the base case we alter the counters of valid/deleted entries. */
+        lp = lpReplaceInteger(lp,&p,aux-1);
+        p = lpNext(lp,p); /* Seek deleted field. */
+        aux = lpGetInteger(p);
+        lp = lpReplaceInteger(lp,&p,aux+1);
+
+        /* Update the listpack with the new pointer. */
+        if (si->lp != lp)
+            raxInsert(si->stream->rax,si->ri.key,si->ri.key_len,lp,NULL);
+    }
+
+    /* Update the number of entries counter. */
+    si->stream->length--;
+
+    /* Re-seek the iterator to fix the now messed up state. */
+    streamID start, end;
+    if (si->rev) {
+        streamDecodeID(si->start_key,&start);
+        end = *current;
+    } else {
+        start = *current;
+        streamDecodeID(si->end_key,&end);
+    }
+    streamIteratorStop(si);
+    streamIteratorStart(si,si->stream,&start,&end,si->rev);
+
+    /* TODO: perform a garbage collection here if the ratio between
+     * deleted and valid goes over a certain limit. */
+}
+
+/* Stop the stream iterator. The only cleanup we need is to free the rax
+ * iterator, since the stream iterator itself is supposed to be stack
+ * allocated. */
+void streamIteratorStop(streamIterator *si) {
+    raxStop(&si->ri);
+}
+
+/* Return 1 if `id` exists in `s` (and not marked as deleted) */
+int streamEntryExists(stream *s, streamID *id) {
+    streamIterator si;
+    streamIteratorStart(&si,s,id,id,0);
+    streamID myid;
+    int64_t numfields;
+    int found = streamIteratorGetID(&si,&myid,&numfields);
+    streamIteratorStop(&si);
+    if (!found)
+        return 0;
+    serverAssert(streamCompareID(id,&myid) == 0);
+    return 1;
+}
+
+/* Delete the specified item ID from the stream, returning 1 if the item
+ * was deleted 0 otherwise (if it does not exist). */
+int streamDeleteItem(stream *s, streamID *id) {
+    int deleted = 0;
+    streamIterator si;
+    streamIteratorStart(&si,s,id,id,0);
+    streamID myid;
+    int64_t numfields;
+    if (streamIteratorGetID(&si,&myid,&numfields)) {
+        streamIteratorRemoveEntry(&si,&myid);
+        deleted = 1;
+    }
+    streamIteratorStop(&si);
+    return deleted;
+}
+
+/* Get the last valid (non-tombstone) streamID of 's'. */
+void streamLastValidID(stream *s, streamID *maxid)
+{
+    streamIterator si;
+    streamIteratorStart(&si,s,NULL,NULL,1);
+    int64_t numfields;
+    if (!streamIteratorGetID(&si,maxid,&numfields) && s->length)
+        serverPanic("Corrupt stream, length is %llu, but no max id", (unsigned long long)s->length);
+    streamIteratorStop(&si);
+}
+
+/* Maximum size for a stream ID string. In theory 20*2+1 should be enough,
+ * But to avoid chance for off by one issues and null-term, in case this will
+ * be used as parsing buffer, we use a slightly larger buffer. On the other
+ * hand considering sds header is gonna add 4 bytes, we wanna keep below the
+ * allocator's 48 bytes bin. */
+#define STREAM_ID_STR_LEN 44
+
+sds createStreamIDString(streamID *id) {
+    /* Optimization: pre-allocate a big enough buffer to avoid reallocs. */
+    sds str = sdsnewlen(SDS_NOINIT, STREAM_ID_STR_LEN);
+    sdssetlen(str, 0);
+    return sdscatfmt(str,"%U-%U", id->ms,id->seq);
+}
+
+/* Emit a reply in the client output buffer by formatting a Stream ID
+ * in the standard <ms>-<seq> format, using the simple string protocol
+ * of REPL. */
+void addReplyStreamID(client *c, streamID *id) {
+    addReplyBulkSds(c,createStreamIDString(id));
+}
+
+void setDeferredReplyStreamID(client *c, void *dr, streamID *id) {
+    setDeferredReplyBulkSds(c, dr, createStreamIDString(id));
+}
+
+/* Similar to the above function, but just creates an object, usually useful
+ * for replication purposes to create arguments. */
+robj *createObjectFromStreamID(streamID *id) {
+    return createObject(OBJ_STRING, createStreamIDString(id));
+}
+
+/* Returns non-zero if the ID is 0-0. */
+int streamIDEqZero(streamID *id) {
+    return !(id->ms || id->seq);
+}
+
+/* A helper that returns non-zero if the range from 'start' to `end`
+ * contains a tombstone.
+ *
+ * NOTE: this assumes that the caller had verified that 'start' is less than
+ * 's->last_id'. */
+int streamRangeHasTombstones(stream *s, streamID *start, streamID *end) {
+    streamID start_id, end_id;
+
+    if (!s->length || streamIDEqZero(&s->max_deleted_entry_id)) {
+        /* The stream is empty or has no tombstones. */
+        return 0;
+    }
+
+    if (streamCompareID(&s->first_id,&s->max_deleted_entry_id) > 0) {
+        /* The latest tombstone is before the first entry. */
+        return 0;
+    }
+
+    if (start) {
+        start_id = *start;
+    } else {
+        start_id.ms = 0;
+        start_id.seq = 0;
+    }
+
+    if (end) {
+        end_id = *end;
+    } else {
+        end_id.ms = UINT64_MAX;
+        end_id.seq = UINT64_MAX;
+    }
+
+    if (streamCompareID(&start_id,&s->max_deleted_entry_id) <= 0 &&
+        streamCompareID(&s->max_deleted_entry_id,&end_id) <= 0)
+    {
+        /* start_id <= max_deleted_entry_id <= end_id: The range does include a tombstone. */
+        return 1;
+    }
+
+    /* The range doesn't includes a tombstone. */
+    return 0;
+}
+
+/* Replies with a consumer group's current lag, that is the number of messages
+ * in the stream that are yet to be delivered. In case that the lag isn't
+ * available due to fragmentation, the reply to the client is a null. */
+void streamReplyWithCGLag(client *c, stream *s, streamCG *cg) {
+    int valid = 0;
+    long long lag = 0;
+
+    if (!s->entries_added) {
+        /* The lag of a newly-initialized stream is 0. */
+        lag = 0;
+        valid = 1;
+    } else if (cg->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s,&cg->last_id,NULL)) {
+        /* No fragmentation ahead means that the group's logical reads counter
+         * is valid for performing the lag calculation. */
+        lag = (long long)s->entries_added - cg->entries_read;
+        valid = 1;
+    } else {
+        /* Attempt to retrieve the group's last ID logical read counter. */
+        long long entries_read = streamEstimateDistanceFromFirstEverEntry(s,&cg->last_id);
+        if (entries_read != SCG_INVALID_ENTRIES_READ) {
+            /* A valid counter was obtained. */
+            lag = (long long)s->entries_added - entries_read;
+            valid = 1;
+        }
+    }
+
+    if (valid) {
+        addReplyLongLong(c,lag);
+    } else {
+        addReplyNull(c);
+    }
+}
+
+/* This function returns a value that is the ID's logical read counter, or its
+ * distance (the number of entries) from the first entry ever to have been added
+ * to the stream.
+ * 
+ * A counter is returned only in one of the following cases:
+ * 1. The ID is the same as the stream's last ID. In this case, the returned
+ *    is the same as the stream's entries_added counter.
+ * 2. The ID equals that of the currently first entry in the stream, and the
+ *    stream has no tombstones. The returned value, in this case, is the result
+ *    of subtracting the stream's length from its added_entries, incremented by
+ *    one.
+ * 3. The ID less than the stream's first current entry's ID, and there are no
+ *    tombstones. Here the estimated counter is the result of subtracting the
+ *    stream's length from its added_entries.
+ * 4. The stream's added_entries is zero, meaning that no entries were ever
+ *    added.
+ *
+ * The special return value of ULLONG_MAX signals that the counter's value isn't
+ * obtainable. It is returned in these cases:
+ * 1. The provided ID, if it even exists, is somewhere between the stream's
+ *    current first and last entries' IDs, or in the future.
+ * 2. The stream contains one or more tombstones. */
+long long streamEstimateDistanceFromFirstEverEntry(stream *s, streamID *id) {
+    /* The counter of any ID in an empty, never-before-used stream is 0. */
+    if (!s->entries_added) {
+        return 0;
+    }
+
+    /* In the empty stream, if the ID is smaller or equal to the last ID,
+     * it can set to the current added_entries value. */
+    if (!s->length && streamCompareID(id,&s->last_id) < 1) {
+        return s->entries_added;
+    }
+
+    int cmp_last = streamCompareID(id,&s->last_id);
+    if (cmp_last == 0) {
+        /* Return the exact counter of the last entry in the stream. */
+        return s->entries_added;
+    } else if (cmp_last > 0) {
+        /* The counter of a future ID is unknown. */
+        return SCG_INVALID_ENTRIES_READ;
+    }
+
+    int cmp_id_first = streamCompareID(id,&s->first_id);
+    int cmp_xdel_first = streamCompareID(&s->max_deleted_entry_id,&s->first_id);
+    if (streamIDEqZero(&s->max_deleted_entry_id) || cmp_xdel_first < 0) {
+        /* There's definitely no fragmentation ahead. */
+        if (cmp_id_first < 0) {
+            /* Return the estimated counter. */
+            return s->entries_added - s->length;
+        } else if (cmp_id_first == 0) {
+            /* Return the exact counter of the first entry in the stream. */
+            return s->entries_added - s->length + 1;
+        }
+    }
+
+    /* The ID is either before an XDEL that fragments the stream or an arbitrary
+     * ID. Either case, so we can't make a prediction. */
+    return SCG_INVALID_ENTRIES_READ;
+}
+
+/* As a result of an explicit XCLAIM or XREADGROUP command, new entries
+ * are created in the pending list of the stream and consumers. We need
+ * to propagate this changes in the form of XCLAIM commands. */
+void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupname, robj *id, streamNACK *nack) {
+    /* We need to generate an XCLAIM that will work in a idempotent fashion:
+     *
+     * XCLAIM <key> <group> <consumer> 0 <id> TIME <milliseconds-unix-time>
+     *        RETRYCOUNT <count> FORCE JUSTID LASTID <id>.
+     *
+     * Note that JUSTID is useful in order to avoid that XCLAIM will do
+     * useless work in the slave side, trying to fetch the stream item. */
+    robj *argv[14];
+    argv[0] = shared.xclaim;
+    argv[1] = key;
+    argv[2] = groupname;
+    argv[3] = createStringObject(nack->consumer->name,sdslen(nack->consumer->name));
+    argv[4] = shared.integers[0];
+    argv[5] = id;
+    argv[6] = shared.time;
+    argv[7] = createStringObjectFromLongLong(nack->delivery_time);
+    argv[8] = shared.retrycount;
+    argv[9] = createStringObjectFromLongLong(nack->delivery_count);
+    argv[10] = shared.force;
+    argv[11] = shared.justid;
+    argv[12] = shared.lastid;
+    argv[13] = createObjectFromStreamID(&group->last_id);
+
+    alsoPropagate(c->db->id,argv,14,PROPAGATE_AOF|PROPAGATE_REPL);
+
+    decrRefCount(argv[3]);
+    decrRefCount(argv[7]);
+    decrRefCount(argv[9]);
+    decrRefCount(argv[13]);
+}
+
+/* We need this when we want to propagate the new last-id of a consumer group
+ * that was consumed by XREADGROUP with the NOACK option: in that case we can't
+ * propagate the last ID just using the XCLAIM LASTID option, so we emit
+ *
+ *  XGROUP SETID <key> <groupname> <id> ENTRIESREAD <entries_read>
+ */
+void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupname) {
+    robj *argv[7];
+    argv[0] = shared.xgroup;
+    argv[1] = shared.setid;
+    argv[2] = key;
+    argv[3] = groupname;
+    argv[4] = createObjectFromStreamID(&group->last_id);
+    argv[5] = shared.entriesread;
+    argv[6] = createStringObjectFromLongLong(group->entries_read);
+
+    alsoPropagate(c->db->id,argv,7,PROPAGATE_AOF|PROPAGATE_REPL);
+
+    decrRefCount(argv[4]);
+    decrRefCount(argv[6]);
+}
+
+/* We need this when we want to propagate creation of consumer that was created
+ * by XREADGROUP with the NOACK option. In that case, the only way to create
+ * the consumer at the replica is by using XGROUP CREATECONSUMER (see issue #7140)
+ *
+ *  XGROUP CREATECONSUMER <key> <groupname> <consumername>
+ */
+void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds consumername) {
+    robj *argv[5];
+    argv[0] = shared.xgroup;
+    argv[1] = shared.createconsumer;
+    argv[2] = key;
+    argv[3] = groupname;
+    argv[4] = createObject(OBJ_STRING,sdsdup(consumername));
+
+    alsoPropagate(c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL);
+
+    decrRefCount(argv[4]);
+}
+
+/* Send the stream items in the specified range to the client 'c'. The range
+ * the client will receive is between start and end inclusive, if 'count' is
+ * non zero, no more than 'count' elements are sent.
+ *
+ * The 'end' pointer can be NULL to mean that we want all the elements from
+ * 'start' till the end of the stream. If 'rev' is non zero, elements are
+ * produced in reversed order from end to start.
+ *
+ * The function returns the number of entries emitted.
+ *
+ * If group and consumer are not NULL, the function performs additional work:
+ * 1. It updates the last delivered ID in the group in case we are
+ *    sending IDs greater than the current last ID.
+ * 2. If the requested IDs are already assigned to some other consumer, the
+ *    function will not return it to the client.
+ * 3. An entry in the pending list will be created for every entry delivered
+ *    for the first time to this consumer.
+ * 4. The group's read counter is incremented if it is already valid and there
+ *    are no future tombstones, or is invalidated (set to 0) otherwise. If the
+ *    counter is invalid to begin with, we try to obtain it for the last
+ *    delivered ID.
+ *
+ * The behavior may be modified passing non-zero flags:
+ *
+ * STREAM_RWR_NOACK: Do not create PEL entries, that is, the point "3" above
+ *                   is not performed.
+ * STREAM_RWR_RAWENTRIES: Do not emit array boundaries, but just the entries,
+ *                        and return the number of entries emitted as usually.
+ *                        This is used when the function is just used in order
+ *                        to emit data and there is some higher level logic.
+ *
+ * The final argument 'spi' (stream propagation info pointer) is a structure
+ * filled with information needed to propagate the command execution to AOF
+ * and slaves, in the case a consumer group was passed: we need to generate
+ * XCLAIM commands to create the pending list into AOF/slaves in that case.
+ *
+ * If 'spi' is set to NULL no propagation will happen even if the group was
+ * given, but currently such a feature is never used by the code base that
+ * will always pass 'spi' and propagate when a group is passed.
+ *
+ * Note that this function is recursive in certain cases. When it's called
+ * with a non NULL group and consumer argument, it may call
+ * streamReplyWithRangeFromConsumerPEL() in order to get entries from the
+ * consumer pending entries list. However such a function will then call
+ * streamReplyWithRange() in order to emit single entries (found in the
+ * PEL by ID) to the client. This is the use case for the STREAM_RWR_RAWENTRIES
+ * flag.
+ */
+#define STREAM_RWR_NOACK (1<<0)         /* Do not create entries in the PEL. */
+#define STREAM_RWR_RAWENTRIES (1<<1)    /* Do not emit protocol for array
+                                           boundaries, just the entries. */
+#define STREAM_RWR_HISTORY (1<<2)       /* Only serve consumer local PEL. */
+size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count, int rev, streamCG *group, streamConsumer *consumer, int flags, streamPropInfo *spi) {
+    void *arraylen_ptr = NULL;
+    size_t arraylen = 0;
+    streamIterator si;
+    int64_t numfields;
+    streamID id;
+    int propagate_last_id = 0;
+    int noack = flags & STREAM_RWR_NOACK;
+
+    /* If the client is asking for some history, we serve it using a
+     * different function, so that we return entries *solely* from its
+     * own PEL. This ensures each consumer will always and only see
+     * the history of messages delivered to it and not yet confirmed
+     * as delivered. */
+    if (group && (flags & STREAM_RWR_HISTORY)) {
+        return streamReplyWithRangeFromConsumerPEL(c,s,start,end,count,
+                                                   consumer);
+    }
+
+    if (!(flags & STREAM_RWR_RAWENTRIES))
+        arraylen_ptr = addReplyDeferredLen(c);
+    streamIteratorStart(&si,s,start,end,rev);
+    while(streamIteratorGetID(&si,&id,&numfields)) {
+        /* Update the group last_id if needed. */
+        if (group && streamCompareID(&id,&group->last_id) > 0) {
+            if (group->entries_read != SCG_INVALID_ENTRIES_READ && !streamRangeHasTombstones(s,&id,NULL)) {
+                /* A valid counter and no future tombstones mean we can 
+                 * increment the read counter to keep tracking the group's
+                 * progress. */
+                group->entries_read++;
+            } else if (s->entries_added) {
+                /* The group's counter may be invalid, so we try to obtain it. */
+                group->entries_read = streamEstimateDistanceFromFirstEverEntry(s,&id);
+            }
+            group->last_id = id;
+            /* Group last ID should be propagated only if NOACK was
+             * specified, otherwise the last id will be included
+             * in the propagation of XCLAIM itself. */
+            if (noack) propagate_last_id = 1;
+        }
+
+        /* Emit a two elements array for each item. The first is
+         * the ID, the second is an array of field-value pairs. */
+        addReplyArrayLen(c,2);
+        addReplyStreamID(c,&id);
+
+        addReplyArrayLen(c,numfields*2);
+
+        /* Emit the field-value pairs. */
+        while(numfields--) {
+            unsigned char *key, *value;
+            int64_t key_len, value_len;
+            streamIteratorGetField(&si,&key,&value,&key_len,&value_len);
+            addReplyBulkCBuffer(c,key,key_len);
+            addReplyBulkCBuffer(c,value,value_len);
+        }
+
+        /* If a group is passed, we need to create an entry in the
+         * PEL (pending entries list) of this group *and* this consumer.
+         *
+         * Note that we cannot be sure about the fact the message is not
+         * already owned by another consumer, because the admin is able
+         * to change the consumer group last delivered ID using the
+         * XGROUP SETID command. So if we find that there is already
+         * a NACK for the entry, we need to associate it to the new
+         * consumer. */
+        if (group && !noack) {
+            unsigned char buf[sizeof(streamID)];
+            streamEncodeID(buf,&id);
+
+            /* Try to add a new NACK. Most of the time this will work and
+             * will not require extra lookups. We'll fix the problem later
+             * if we find that there is already a entry for this ID. */
+            streamNACK *nack = streamCreateNACK(consumer);
+            int group_inserted =
+                raxTryInsert(group->pel,buf,sizeof(buf),nack,NULL);
+            int consumer_inserted =
+                raxTryInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
+
+            /* Now we can check if the entry was already busy, and
+             * in that case reassign the entry to the new consumer,
+             * or update it if the consumer is the same as before. */
+            if (group_inserted == 0) {
+                streamFreeNACK(nack);
+                nack = raxFind(group->pel,buf,sizeof(buf));
+                serverAssert(nack != raxNotFound);
+                raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
+                /* Update the consumer and NACK metadata. */
+                nack->consumer = consumer;
+                nack->delivery_time = commandTimeSnapshot();
+                nack->delivery_count = 1;
+                /* Add the entry in the new consumer local PEL. */
+                raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
+            } else if (group_inserted == 1 && consumer_inserted == 0) {
+                serverPanic("NACK half-created. Should not be possible.");
+            }
+
+            consumer->active_time = commandTimeSnapshot();
+
+            /* Propagate as XCLAIM. */
+            if (spi) {
+                robj *idarg = createObjectFromStreamID(&id);
+                streamPropagateXCLAIM(c,spi->keyname,group,spi->groupname,idarg,nack);
+                decrRefCount(idarg);
+            }
+        }
+
+        arraylen++;
+        if (count && count == arraylen) break;
+    }
+
+    if (spi && propagate_last_id)
+        streamPropagateGroupID(c,spi->keyname,group,spi->groupname);
+
+    streamIteratorStop(&si);
+    if (arraylen_ptr) setDeferredArrayLen(c,arraylen_ptr,arraylen);
+    return arraylen;
+}
+
+/* This is a helper function for streamReplyWithRange() when called with
+ * group and consumer arguments, but with a range that is referring to already
+ * delivered messages. In this case we just emit messages that are already
+ * in the history of the consumer, fetching the IDs from its PEL.
+ *
+ * Note that this function does not have a 'rev' argument because it's not
+ * possible to iterate in reverse using a group. Basically this function
+ * is only called as a result of the XREADGROUP command.
+ *
+ * This function is more expensive because it needs to inspect the PEL and then
+ * seek into the radix tree of the messages in order to emit the full message
+ * to the client. However clients only reach this code path when they are
+ * fetching the history of already retrieved messages, which is rare. */
+size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamConsumer *consumer) {
+    raxIterator ri;
+    unsigned char startkey[sizeof(streamID)];
+    unsigned char endkey[sizeof(streamID)];
+    streamEncodeID(startkey,start);
+    if (end) streamEncodeID(endkey,end);
+
+    size_t arraylen = 0;
+    void *arraylen_ptr = addReplyDeferredLen(c);
+    raxStart(&ri,consumer->pel);
+    raxSeek(&ri,">=",startkey,sizeof(startkey));
+    while(raxNext(&ri) && (!count || arraylen < count)) {
+        if (end && memcmp(ri.key,end,ri.key_len) > 0) break;
+        streamID thisid;
+        streamDecodeID(ri.key,&thisid);
+        if (streamReplyWithRange(c,s,&thisid,&thisid,1,0,NULL,NULL,
+                                 STREAM_RWR_RAWENTRIES,NULL) == 0)
+        {
+            /* Note that we may have a not acknowledged entry in the PEL
+             * about a message that's no longer here because was removed
+             * by the user by other means. In that case we signal it emitting
+             * the ID but then a NULL entry for the fields. */
+            addReplyArrayLen(c,2);
+            addReplyStreamID(c,&thisid);
+            addReplyNullArray(c);
+        } else {
+            streamNACK *nack = ri.data;
+            nack->delivery_time = commandTimeSnapshot();
+            nack->delivery_count++;
+        }
+        arraylen++;
+    }
+    raxStop(&ri);
+    setDeferredArrayLen(c,arraylen_ptr,arraylen);
+    return arraylen;
+}
+
+/* -----------------------------------------------------------------------
+ * Stream commands implementation
+ * ----------------------------------------------------------------------- */
+
+/* Look the stream at 'key' and return the corresponding stream object.
+ * The function creates a key setting it to an empty stream if needed. */
+robj *streamTypeLookupWriteOrCreate(client *c, robj *key, int no_create) {
+    robj *o = lookupKeyWrite(c->db,key);
+    if (checkType(c,o,OBJ_STREAM)) return NULL;
+    if (o == NULL) {
+        if (no_create) {
+            addReplyNull(c);
+            return NULL;
+        }
+        o = createStreamObject();
+        dbAdd(c->db,key,o);
+    }
+    return o;
+}
+
+/* Parse a stream ID in the format given by clients to Redis, that is
+ * <ms>-<seq>, and converts it into a streamID structure. If
+ * the specified ID is invalid C_ERR is returned and an error is reported
+ * to the client, otherwise C_OK is returned. The ID may be in incomplete
+ * form, just stating the milliseconds time part of the stream. In such a case
+ * the missing part is set according to the value of 'missing_seq' parameter.
+ *
+ * The IDs "-" and "+" specify respectively the minimum and maximum IDs
+ * that can be represented. If 'strict' is set to 1, "-" and "+" will be
+ * treated as an invalid ID.
+ *
+ * The ID form <ms>-* specifies a millisconds-only ID, leaving the sequence part
+ * to be autogenerated. When a non-NULL 'seq_given' argument is provided, this
+ * form is accepted and the argument is set to 0 unless the sequence part is
+ * specified.
+ * 
+ * If 'c' is set to NULL, no reply is sent to the client. */
+int streamGenericParseIDOrReply(client *c, const robj *o, streamID *id, uint64_t missing_seq, int strict, int *seq_given) {
+    char buf[128];
+    if (sdslen(o->ptr) > sizeof(buf)-1) goto invalid;
+    memcpy(buf,o->ptr,sdslen(o->ptr)+1);
+
+    if (strict && (buf[0] == '-' || buf[0] == '+') && buf[1] == '\0')
+        goto invalid;
+
+    if (seq_given != NULL) {
+        *seq_given = 1;
+    }
+
+    /* Handle the "-" and "+" special cases. */
+    if (buf[0] == '-' && buf[1] == '\0') {
+        id->ms = 0;
+        id->seq = 0;
+        return C_OK;
+    } else if (buf[0] == '+' && buf[1] == '\0') {
+        id->ms = UINT64_MAX;
+        id->seq = UINT64_MAX;
+        return C_OK;
+    }
+
+    /* Parse <ms>-<seq> form. */
+    unsigned long long ms, seq;
+    char *dot = strchr(buf,'-');
+    if (dot) *dot = '\0';
+    if (string2ull(buf,&ms) == 0) goto invalid;
+    if (dot) {
+        size_t seqlen = strlen(dot+1);
+        if (seq_given != NULL && seqlen == 1 && *(dot + 1) == '*') {
+            /* Handle the <ms>-* form. */
+            seq = 0;
+            *seq_given = 0;
+        } else if (string2ull(dot+1,&seq) == 0) {
+            goto invalid;
+        }
+    } else {
+        seq = missing_seq;
+    }
+    id->ms = ms;
+    id->seq = seq;
+    return C_OK;
+
+invalid:
+    if (c) addReplyError(c,"Invalid stream ID specified as stream "
+                           "command argument");
+    return C_ERR;
+}
+
+/* Wrapper for streamGenericParseIDOrReply() used by module API. */
+int streamParseID(const robj *o, streamID *id) {
+    return streamGenericParseIDOrReply(NULL,o,id,0,0,NULL);
+}
+
+/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
+ * 0, to be used when - and + are acceptable IDs. */
+int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
+    return streamGenericParseIDOrReply(c,o,id,missing_seq,0,NULL);
+}
+
+/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
+ * 1, to be used when we want to return an error if the special IDs + or -
+ * are provided. */
+int streamParseStrictIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int *seq_given) {
+    return streamGenericParseIDOrReply(c,o,id,missing_seq,1,seq_given);
+}
+
+/* Helper for parsing a stream ID that is a range query interval. When the
+ * exclude argument is NULL, streamParseIDOrReply() is called and the interval
+ * is treated as close (inclusive). Otherwise, the exclude argument is set if 
+ * the interval is open (the "(" prefix) and streamParseStrictIDOrReply() is
+ * called in that case.
+ */
+int streamParseIntervalIDOrReply(client *c, robj *o, streamID *id, int *exclude, uint64_t missing_seq) {
+    char *p = o->ptr;
+    size_t len = sdslen(p);
+    int invalid = 0;
+    
+    if (exclude != NULL) *exclude = (len > 1 && p[0] == '(');
+    if (exclude != NULL && *exclude) {
+        robj *t = createStringObject(p+1,len-1);
+        invalid = (streamParseStrictIDOrReply(c,t,id,missing_seq,NULL) == C_ERR);
+        decrRefCount(t);
+    } else 
+        invalid = (streamParseIDOrReply(c,o,id,missing_seq) == C_ERR);
+    if (invalid)
+        return C_ERR;
+    return C_OK;
+}
+
+void streamRewriteApproxSpecifier(client *c, int idx) {
+    rewriteClientCommandArgument(c,idx,shared.special_equals);
+}
+
+/* We propagate MAXLEN/MINID ~ <count> as MAXLEN/MINID = <resulting-len-of-stream>
+ * otherwise trimming is no longer deterministic on replicas / AOF. */
+void streamRewriteTrimArgument(client *c, stream *s, int trim_strategy, int idx) {
+    robj *arg;
+    if (trim_strategy == TRIM_STRATEGY_MAXLEN) {
+        arg = createStringObjectFromLongLong(s->length);
+    } else {
+        streamID first_id;
+        streamGetEdgeID(s,1,0,&first_id);
+        arg = createObjectFromStreamID(&first_id);
+    }
+
+    rewriteClientCommandArgument(c,idx,arg);
+    decrRefCount(arg);
+}
+
+/* XADD key [(MAXLEN [~|=] <count> | MINID [~|=] <id>) [LIMIT <entries>]] [NOMKSTREAM] <ID or *> [field value] [field value] ... */
+void xaddCommand(client *c) {
+    /* Parse options. */
+    streamAddTrimArgs parsed_args;
+    int idpos = streamParseAddOrTrimArgsOrReply(c, &parsed_args, 1);
+    if (idpos < 0)
+        return; /* streamParseAddOrTrimArgsOrReply already replied. */
+    int field_pos = idpos+1; /* The ID is always one argument before the first field */
+
+    /* Check arity. */
+    if ((c->argc - field_pos) < 2 || ((c->argc-field_pos) % 2) == 1) {
+        addReplyErrorArity(c);
+        return;
+    }
+
+    /* Return ASAP if minimal ID (0-0) was given so we avoid possibly creating
+     * a new stream and have streamAppendItem fail, leaving an empty key in the
+     * database. */
+    if (parsed_args.id_given && parsed_args.seq_given &&
+        parsed_args.id.ms == 0 && parsed_args.id.seq == 0)
+    {
+        addReplyError(c,"The ID specified in XADD must be greater than 0-0");
+        return;
+    }
+
+    /* Lookup the stream at key. */
+    robj *o;
+    stream *s;
+    if ((o = streamTypeLookupWriteOrCreate(c,c->argv[1],parsed_args.no_mkstream)) == NULL) return;
+    s = o->ptr;
+
+    /* Return ASAP if the stream has reached the last possible ID */
+    if (s->last_id.ms == UINT64_MAX && s->last_id.seq == UINT64_MAX) {
+        addReplyError(c,"The stream has exhausted the last possible ID, "
+                        "unable to add more items");
+        return;
+    }
+
+    /* Append using the low level function and return the ID. */
+    errno = 0;
+    streamID id;
+    if (streamAppendItem(s,c->argv+field_pos,(c->argc-field_pos)/2,
+        &id,parsed_args.id_given ? &parsed_args.id : NULL,parsed_args.seq_given) == C_ERR)
+    {
+        serverAssert(errno != 0);
+        if (errno == EDOM)
+            addReplyError(c,"The ID specified in XADD is equal or smaller than "
+                            "the target stream top item");
+        else
+            addReplyError(c,"Elements are too large to be stored");
+        return;
+    }
+    sds replyid = createStreamIDString(&id);
+    addReplyBulkCBuffer(c, replyid, sdslen(replyid));
+
+    signalModifiedKey(c,c->db,c->argv[1]);
+    notifyKeyspaceEvent(NOTIFY_STREAM,"xadd",c->argv[1],c->db->id);
+    server.dirty++;
+
+    /* Trim if needed. */
+    if (parsed_args.trim_strategy != TRIM_STRATEGY_NONE) {
+        if (streamTrim(s, &parsed_args)) {
+            notifyKeyspaceEvent(NOTIFY_STREAM,"xtrim",c->argv[1],c->db->id);
+        }
+        if (parsed_args.approx_trim) {
+            /* In case our trimming was limited (by LIMIT or by ~) we must
+             * re-write the relevant trim argument to make sure there will be
+             * no inconsistencies in AOF loading or in the replica.
+             * It's enough to check only args->approx because there is no
+             * way LIMIT is given without the ~ option. */
+            streamRewriteApproxSpecifier(c,parsed_args.trim_strategy_arg_idx-1);
+            streamRewriteTrimArgument(c,s,parsed_args.trim_strategy,parsed_args.trim_strategy_arg_idx);
+        }
+    }
+
+    /* Let's rewrite the ID argument with the one actually generated for
+     * AOF/replication propagation. */
+    if (!parsed_args.id_given || !parsed_args.seq_given) {
+        robj *idarg = createObject(OBJ_STRING, replyid);
+        rewriteClientCommandArgument(c, idpos, idarg);
+        decrRefCount(idarg);
+    } else {
+        sdsfree(replyid);
+    }
+
+    /* We need to signal to blocked clients that there is new data on this
+     * stream. */
+    signalKeyAsReady(c->db, c->argv[1], OBJ_STREAM);
+}
+
+/* XRANGE/XREVRANGE actual implementation.
+ * The 'start' and 'end' IDs are parsed as follows:
+ *   Incomplete 'start' has its sequence set to 0, and 'end' to UINT64_MAX.
+ *   "-" and "+"" mean the minimal and maximal ID values, respectively.
+ *   The "(" prefix means an open (exclusive) range, so XRANGE stream (1-0 (2-0
+ *   will match anything from 1-1 and 1-UINT64_MAX.
+ */
+void xrangeGenericCommand(client *c, int rev) {
+    robj *o;
+    stream *s;
+    streamID startid, endid;
+    long long count = -1;
+    robj *startarg = rev ? c->argv[3] : c->argv[2];
+    robj *endarg = rev ? c->argv[2] : c->argv[3];
+    int startex = 0, endex = 0;
+    
+    /* Parse start and end IDs. */
+    if (streamParseIntervalIDOrReply(c,startarg,&startid,&startex,0) != C_OK)
+        return;
+    if (startex && streamIncrID(&startid) != C_OK) {
+        addReplyError(c,"invalid start ID for the interval");
+        return;
+    }
+    if (streamParseIntervalIDOrReply(c,endarg,&endid,&endex,UINT64_MAX) != C_OK)
+        return;
+    if (endex && streamDecrID(&endid) != C_OK) {
+        addReplyError(c,"invalid end ID for the interval");
+        return;
+    }
+
+    /* Parse the COUNT option if any. */
+    if (c->argc > 4) {
+        for (int j = 4; j < c->argc; j++) {
+            int additional = c->argc-j-1;
+            if (strcasecmp(c->argv[j]->ptr,"COUNT") == 0 && additional >= 1) {
+                if (getLongLongFromObjectOrReply(c,c->argv[j+1],&count,NULL)
+                    != C_OK) return;
+                if (count < 0) count = 0;
+                j++; /* Consume additional arg. */
+            } else {
+                addReplyErrorObject(c,shared.syntaxerr);
+                return;
+            }
+        }
+    }
+
+    /* Return the specified range to the user. */
+    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptyarray)) == NULL ||
+         checkType(c,o,OBJ_STREAM)) return;
+
+    s = o->ptr;
+
+    if (count == 0) {
+        addReplyNullArray(c);
+    } else {
+        if (count == -1) count = 0;
+        streamReplyWithRange(c,s,&startid,&endid,count,rev,NULL,NULL,0,NULL);
+    }
+}
+
+/* XRANGE key start end [COUNT <n>] */
+void xrangeCommand(client *c) {
+    xrangeGenericCommand(c,0);
+}
+
+/* XREVRANGE key end start [COUNT <n>] */
+void xrevrangeCommand(client *c) {
+    xrangeGenericCommand(c,1);
+}
+
+/* XLEN key*/
+void xlenCommand(client *c) {
+    robj *o;
+    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL
+        || checkType(c,o,OBJ_STREAM)) return;
+    stream *s = o->ptr;
+    addReplyLongLong(c,s->length);
+}
+
+/* XREAD [BLOCK <milliseconds>] [COUNT <count>] STREAMS key_1 key_2 ... key_N
+ *       ID_1 ID_2 ... ID_N
+ *
+ * This function also implements the XREADGROUP command, which is like XREAD
+ * but accepting the [GROUP group-name consumer-name] additional option.
+ * This is useful because while XREAD is a read command and can be called
+ * on slaves, XREADGROUP is not. */
+#define XREAD_BLOCKED_DEFAULT_COUNT 1000
+void xreadCommand(client *c) {
+    long long timeout = -1; /* -1 means, no BLOCK argument given. */
+    long long count = 0;
+    int streams_count = 0;
+    int streams_arg = 0;
+    int noack = 0;          /* True if NOACK option was specified. */
+    streamID static_ids[STREAMID_STATIC_VECTOR_LEN];
+    streamID *ids = static_ids;
+    streamCG **groups = NULL;
+    int xreadgroup = sdslen(c->argv[0]->ptr) == 10; /* XREAD or XREADGROUP? */
+    robj *groupname = NULL;
+    robj *consumername = NULL;
+
+    /* Parse arguments. */
+    for (int i = 1; i < c->argc; i++) {
+        int moreargs = c->argc-i-1;
+        char *o = c->argv[i]->ptr;
+        if (!strcasecmp(o,"BLOCK") && moreargs) {
+            if (c->flags & CLIENT_SCRIPT) {
+                /*
+                 * Although the CLIENT_DENY_BLOCKING flag should protect from blocking the client
+                 * on Lua/MULTI/RM_Call we want special treatment for Lua to keep backward compatibility.
+                 * There is no sense to use BLOCK option within Lua. */
+                addReplyErrorFormat(c, "%s command is not allowed with BLOCK option from scripts", (char *)c->argv[0]->ptr);
+                return;
+            }
+            i++;
+            if (getTimeoutFromObjectOrReply(c,c->argv[i],&timeout,
+                UNIT_MILLISECONDS) != C_OK) return;
+        } else if (!strcasecmp(o,"COUNT") && moreargs) {
+            i++;
+            if (getLongLongFromObjectOrReply(c,c->argv[i],&count,NULL) != C_OK)
+                return;
+            if (count < 0) count = 0;
+        } else if (!strcasecmp(o,"STREAMS") && moreargs) {
+            streams_arg = i+1;
+            streams_count = (c->argc-streams_arg);
+            if ((streams_count % 2) != 0) {
+                char symbol = xreadgroup ? '>' : '$';
+                addReplyErrorFormat(c,"Unbalanced '%s' list of streams: "
+                                      "for each stream key an ID or '%c' must be "
+                                      "specified.", c->cmd->fullname,symbol);
+                return;
+            }
+            streams_count /= 2; /* We have two arguments for each stream. */
+            break;
+        } else if (!strcasecmp(o,"GROUP") && moreargs >= 2) {
+            if (!xreadgroup) {
+                addReplyError(c,"The GROUP option is only supported by "
+                                "XREADGROUP. You called XREAD instead.");
+                return;
+            }
+            groupname = c->argv[i+1];
+            consumername = c->argv[i+2];
+            i += 2;
+        } else if (!strcasecmp(o,"NOACK")) {
+            if (!xreadgroup) {
+                addReplyError(c,"The NOACK option is only supported by "
+                                "XREADGROUP. You called XREAD instead.");
+                return;
+            }
+            noack = 1;
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return;
+        }
+    }
+
+    /* STREAMS option is mandatory. */
+    if (streams_arg == 0) {
+        addReplyErrorObject(c,shared.syntaxerr);
+        return;
+    }
+
+    /* If the user specified XREADGROUP then it must also
+     * provide the GROUP option. */
+    if (xreadgroup && groupname == NULL) {
+        addReplyError(c,"Missing GROUP option for XREADGROUP");
+        return;
+    }
+
+    /* Parse the IDs and resolve the group name. */
+    if (streams_count > STREAMID_STATIC_VECTOR_LEN)
+        ids = zmalloc(sizeof(streamID)*streams_count);
+    if (groupname) groups = zmalloc(sizeof(streamCG*)*streams_count);
+
+    for (int i = streams_arg + streams_count; i < c->argc; i++) {
+        /* Specifying "$" as last-known-id means that the client wants to be
+         * served with just the messages that will arrive into the stream
+         * starting from now. */
+        int id_idx = i - streams_arg - streams_count;
+        robj *key = c->argv[i-streams_count];
+        robj *o = lookupKeyRead(c->db,key);
+        if (checkType(c,o,OBJ_STREAM)) goto cleanup;
+        streamCG *group = NULL;
+
+        /* If a group was specified, than we need to be sure that the
+         * key and group actually exist. */
+        if (groupname) {
+            if (o == NULL ||
+                (group = streamLookupCG(o->ptr,groupname->ptr)) == NULL)
+            {
+                addReplyErrorFormat(c, "-NOGROUP No such key '%s' or consumer "
+                                       "group '%s' in XREADGROUP with GROUP "
+                                       "option",
+                                    (char*)key->ptr,(char*)groupname->ptr);
+                goto cleanup;
+            }
+            groups[id_idx] = group;
+        }
+
+        if (strcmp(c->argv[i]->ptr,"$") == 0) {
+            if (xreadgroup) {
+                addReplyError(c,"The $ ID is meaningless in the context of "
+                                "XREADGROUP: you want to read the history of "
+                                "this consumer by specifying a proper ID, or "
+                                "use the > ID to get new messages. The $ ID would "
+                                "just return an empty result set.");
+                goto cleanup;
+            }
+            if (o) {
+                stream *s = o->ptr;
+                ids[id_idx] = s->last_id;
+            } else {
+                ids[id_idx].ms = 0;
+                ids[id_idx].seq = 0;
+            }
+            continue;
+        } else if (strcmp(c->argv[i]->ptr,">") == 0) {
+            if (!xreadgroup) {
+                addReplyError(c,"The > ID can be specified only when calling "
+                                "XREADGROUP using the GROUP <group> "
+                                "<consumer> option.");
+                goto cleanup;
+            }
+            /* We use just the maximum ID to signal this is a ">" ID, anyway
+             * the code handling the blocking clients will have to update the
+             * ID later in order to match the changing consumer group last ID. */
+            ids[id_idx].ms = UINT64_MAX;
+            ids[id_idx].seq = UINT64_MAX;
+            continue;
+        }
+        if (streamParseStrictIDOrReply(c,c->argv[i],ids+id_idx,0,NULL) != C_OK)
+            goto cleanup;
+    }
+
+    /* Try to serve the client synchronously. */
+    size_t arraylen = 0;
+    void *arraylen_ptr = NULL;
+    for (int i = 0; i < streams_count; i++) {
+        robj *o = lookupKeyRead(c->db,c->argv[streams_arg+i]);
+        if (o == NULL) continue;
+        stream *s = o->ptr;
+        streamID *gt = ids+i; /* ID must be greater than this. */
+        int serve_synchronously = 0;
+        int serve_history = 0; /* True for XREADGROUP with ID != ">". */
+        streamConsumer *consumer = NULL; /* Unused if XREAD */
+        streamPropInfo spi = {c->argv[streams_arg+i],groupname}; /* Unused if XREAD */
+
+        /* Check if there are the conditions to serve the client
+         * synchronously. */
+        if (groups) {
+            /* If the consumer is blocked on a group, we always serve it
+             * synchronously (serving its local history) if the ID specified
+             * was not the special ">" ID. */
+            if (gt->ms != UINT64_MAX ||
+                gt->seq != UINT64_MAX)
+            {
+                serve_synchronously = 1;
+                serve_history = 1;
+            } else if (s->length) {
+                /* We also want to serve a consumer in a consumer group
+                 * synchronously in case the group top item delivered is smaller
+                 * than what the stream has inside. */
+                streamID maxid, *last = &groups[i]->last_id;
+                streamLastValidID(s, &maxid);
+                if (streamCompareID(&maxid, last) > 0) {
+                    serve_synchronously = 1;
+                    *gt = *last;
+                }
+            }
+            consumer = streamLookupConsumer(groups[i],consumername->ptr);
+            if (consumer == NULL) {
+                consumer = streamCreateConsumer(groups[i],consumername->ptr,
+                                                c->argv[streams_arg+i],
+                                                c->db->id,SCC_DEFAULT);
+                if (noack)
+                    streamPropagateConsumerCreation(c,spi.keyname,
+                                                    spi.groupname,
+                                                    consumer->name);
+            }
+            consumer->seen_time = commandTimeSnapshot();
+        } else if (s->length) {
+            /* For consumers without a group, we serve synchronously if we can
+             * actually provide at least one item from the stream. */
+            streamID maxid;
+            streamLastValidID(s, &maxid);
+            if (streamCompareID(&maxid, gt) > 0) {
+                serve_synchronously = 1;
+            }
+        }
+
+        if (serve_synchronously) {
+            arraylen++;
+            if (arraylen == 1) arraylen_ptr = addReplyDeferredLen(c);
+            /* streamReplyWithRange() handles the 'start' ID as inclusive,
+             * so start from the next ID, since we want only messages with
+             * IDs greater than start. */
+            streamID start = *gt;
+            streamIncrID(&start);
+
+            /* Emit the two elements sub-array consisting of the name
+             * of the stream and the data we extracted from it. */
+            if (c->resp == 2) addReplyArrayLen(c,2);
+            addReplyBulk(c,c->argv[streams_arg+i]);
+            
+            int flags = 0;
+            if (noack) flags |= STREAM_RWR_NOACK;
+            if (serve_history) flags |= STREAM_RWR_HISTORY;
+            streamReplyWithRange(c,s,&start,NULL,count,0,
+                                 groups ? groups[i] : NULL,
+                                 consumer, flags, &spi);
+            if (groups) server.dirty++;
+        }
+    }
+
+     /* We replied synchronously! Set the top array len and return to caller. */
+    if (arraylen) {
+        if (c->resp == 2)
+            setDeferredArrayLen(c,arraylen_ptr,arraylen);
+        else
+            setDeferredMapLen(c,arraylen_ptr,arraylen);
+        goto cleanup;
+    }
+
+    /* Block if needed. */
+    if (timeout != -1) {
+        /* If we are not allowed to block the client, the only thing
+         * we can do is treating it as a timeout (even with timeout 0). */
+        if (c->flags & CLIENT_DENY_BLOCKING) {
+            addReplyNullArray(c);
+            goto cleanup;
+        }
+        /* We change the '$' to the current last ID for this stream. this is
+         * Since later on when we unblock on arriving data - we would like to
+         * re-process the command and in case '$' stays we will spin-block forever.
+         */
+        for (int id_idx = 0; id_idx < streams_count; id_idx++) {
+            int arg_idx = id_idx + streams_arg + streams_count;
+            if (strcmp(c->argv[arg_idx]->ptr,"$") == 0) {
+                robj *argv_streamid = createObjectFromStreamID(&ids[id_idx]);
+                rewriteClientCommandArgument(c, arg_idx, argv_streamid);
+                decrRefCount(argv_streamid);
+            }
+        }
+        blockForKeys(c, BLOCKED_STREAM, c->argv+streams_arg, streams_count, timeout, xreadgroup);
+        goto cleanup;
+    }
+
+    /* No BLOCK option, nor any stream we can serve. Reply as with a
+     * timeout happened. */
+    addReplyNullArray(c);
+    /* Continue to cleanup... */
+
+cleanup: /* Cleanup. */
+
+    /* The command is propagated (in the READGROUP form) as a side effect
+     * of calling lower level APIs. So stop any implicit propagation. */
+    preventCommandPropagation(c);
+    if (ids != static_ids) zfree(ids);
+    zfree(groups);
+}
+
+/* -----------------------------------------------------------------------
+ * Low level implementation of consumer groups
+ * ----------------------------------------------------------------------- */
+
+/* Create a NACK entry setting the delivery count to 1 and the delivery
+ * time to the current time. The NACK consumer will be set to the one
+ * specified as argument of the function. */
+streamNACK *streamCreateNACK(streamConsumer *consumer) {
+    streamNACK *nack = zmalloc(sizeof(*nack));
+    nack->delivery_time = commandTimeSnapshot();
+    nack->delivery_count = 1;
+    nack->consumer = consumer;
+    return nack;
+}
+
+/* Free a NACK entry. */
+void streamFreeNACK(streamNACK *na) {
+    zfree(na);
+}
+
+/* Free a consumer and associated data structures. Note that this function
+ * will not reassign the pending messages associated with this consumer
+ * nor will delete them from the stream, so when this function is called
+ * to delete a consumer, and not when the whole stream is destroyed, the caller
+ * should do some work before. */
+void streamFreeConsumer(streamConsumer *sc) {
+    raxFree(sc->pel); /* No value free callback: the PEL entries are shared
+                         between the consumer and the main stream PEL. */
+    sdsfree(sc->name);
+    zfree(sc);
+}
+
+/* Create a new consumer group in the context of the stream 's', having the
+ * specified name, last server ID and reads counter. If a consumer group with
+ * the same name already exists NULL is returned, otherwise the pointer to the
+ * consumer group is returned. */
+streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id, long long entries_read) {
+    if (s->cgroups == NULL) s->cgroups = raxNew();
+    if (raxFind(s->cgroups,(unsigned char*)name,namelen) != raxNotFound)
+        return NULL;
+
+    streamCG *cg = zmalloc(sizeof(*cg));
+    cg->pel = raxNew();
+    cg->consumers = raxNew();
+    cg->last_id = *id;
+    cg->entries_read = entries_read;
+    raxInsert(s->cgroups,(unsigned char*)name,namelen,cg,NULL);
+    return cg;
+}
+
+/* Free a consumer group and all its associated data. */
+void streamFreeCG(streamCG *cg) {
+    raxFreeWithCallback(cg->pel,(void(*)(void*))streamFreeNACK);
+    raxFreeWithCallback(cg->consumers,(void(*)(void*))streamFreeConsumer);
+    zfree(cg);
+}
+
+/* Lookup the consumer group in the specified stream and returns its
+ * pointer, otherwise if there is no such group, NULL is returned. */
+streamCG *streamLookupCG(stream *s, sds groupname) {
+    if (s->cgroups == NULL) return NULL;
+    streamCG *cg = raxFind(s->cgroups,(unsigned char*)groupname,
+                           sdslen(groupname));
+    return (cg == raxNotFound) ? NULL : cg;
+}
+
+/* Create a consumer with the specified name in the group 'cg' and return.
+ * If the consumer exists, return NULL. As a side effect, when the consumer
+ * is successfully created, the key space will be notified and dirty++ unless
+ * the SCC_NO_NOTIFY or SCC_NO_DIRTIFY flags is specified. */
+streamConsumer *streamCreateConsumer(streamCG *cg, sds name, robj *key, int dbid, int flags) {
+    if (cg == NULL) return NULL;
+    int notify = !(flags & SCC_NO_NOTIFY);
+    int dirty = !(flags & SCC_NO_DIRTIFY);
+    streamConsumer *consumer = zmalloc(sizeof(*consumer));
+    int success = raxTryInsert(cg->consumers,(unsigned char*)name,
+                               sdslen(name),consumer,NULL);
+    if (!success) {
+        zfree(consumer);
+        return NULL;
+    }
+    consumer->name = sdsdup(name);
+    consumer->pel = raxNew();
+    consumer->active_time = -1;
+    consumer->seen_time = commandTimeSnapshot();
+    if (dirty) server.dirty++;
+    if (notify) notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-createconsumer",key,dbid);
+    return consumer;
+}
+
+/* Lookup the consumer with the specified name in the group 'cg'. */
+streamConsumer *streamLookupConsumer(streamCG *cg, sds name) {
+    if (cg == NULL) return NULL;
+    streamConsumer *consumer = raxFind(cg->consumers,(unsigned char*)name,
+                                       sdslen(name));
+    if (consumer == raxNotFound) return NULL;
+    return consumer;
+}
+
+/* Delete the consumer specified in the consumer group 'cg'. */
+void streamDelConsumer(streamCG *cg, streamConsumer *consumer) {
+    /* Iterate all the consumer pending messages, deleting every corresponding
+     * entry from the global entry. */
+    raxIterator ri;
+    raxStart(&ri,consumer->pel);
+    raxSeek(&ri,"^",NULL,0);
+    while(raxNext(&ri)) {
+        streamNACK *nack = ri.data;
+        raxRemove(cg->pel,ri.key,ri.key_len,NULL);
+        streamFreeNACK(nack);
+    }
+    raxStop(&ri);
+
+    /* Deallocate the consumer. */
+    raxRemove(cg->consumers,(unsigned char*)consumer->name,
+              sdslen(consumer->name),NULL);
+    streamFreeConsumer(consumer);
+}
+
+/* -----------------------------------------------------------------------
+ * Consumer groups commands
+ * ----------------------------------------------------------------------- */
+
+/* XGROUP CREATE <key> <groupname> <id or $> [MKSTREAM] [ENTRIESREAD entries_read]
+ * XGROUP SETID <key> <groupname> <id or $> [ENTRIESREAD entries_read]
+ * XGROUP DESTROY <key> <groupname>
+ * XGROUP CREATECONSUMER <key> <groupname> <consumer>
+ * XGROUP DELCONSUMER <key> <groupname> <consumername> */
+void xgroupCommand(client *c) {
+    stream *s = NULL;
+    sds grpname = NULL;
+    streamCG *cg = NULL;
+    char *opt = c->argv[1]->ptr; /* Subcommand name. */
+    int mkstream = 0;
+    long long entries_read = SCG_INVALID_ENTRIES_READ;
+    robj *o;
+
+    /* Everything but the "HELP" option requires a key and group name. */
+    if (c->argc >= 4) {
+        /* Parse optional arguments for CREATE and SETID */
+        int i = 5;
+        int create_subcmd = !strcasecmp(opt,"CREATE");
+        int setid_subcmd = !strcasecmp(opt,"SETID");
+        while (i < c->argc) {
+            if (create_subcmd && !strcasecmp(c->argv[i]->ptr,"MKSTREAM")) {
+                mkstream = 1;
+                i++;
+            } else if ((create_subcmd || setid_subcmd) && !strcasecmp(c->argv[i]->ptr,"ENTRIESREAD") && i + 1 < c->argc) {
+                if (getLongLongFromObjectOrReply(c,c->argv[i+1],&entries_read,NULL) != C_OK)
+                    return;
+                if (entries_read < 0 && entries_read != SCG_INVALID_ENTRIES_READ) {
+                    addReplyError(c,"value for ENTRIESREAD must be positive or -1");
+                    return;
+                }
+                i += 2;
+            } else {
+                addReplySubcommandSyntaxError(c);
+                return;
+            }
+        }
+
+        o = lookupKeyWrite(c->db,c->argv[2]);
+        if (o) {
+            if (checkType(c,o,OBJ_STREAM)) return;
+            s = o->ptr;
+        }
+        grpname = c->argv[3]->ptr;
+    }
+
+    /* Check for missing key/group. */
+    if (c->argc >= 4 && !mkstream) {
+        /* At this point key must exist, or there is an error. */
+        if (s == NULL) {
+            addReplyError(c,
+                "The XGROUP subcommand requires the key to exist. "
+                "Note that for CREATE you may want to use the MKSTREAM "
+                "option to create an empty stream automatically.");
+            return;
+        }
+
+        /* Certain subcommands require the group to exist. */
+        if ((cg = streamLookupCG(s,grpname)) == NULL &&
+            (!strcasecmp(opt,"SETID") ||
+             !strcasecmp(opt,"CREATECONSUMER") ||
+             !strcasecmp(opt,"DELCONSUMER")))
+        {
+            addReplyErrorFormat(c, "-NOGROUP No such consumer group '%s' "
+                                   "for key name '%s'",
+                                   (char*)grpname, (char*)c->argv[2]->ptr);
+            return;
+        }
+    }
+
+    /* Dispatch the different subcommands. */
+    if (c->argc == 2 && !strcasecmp(opt,"HELP")) {
+        const char *help[] = {
+"CREATE <key> <groupname> <id|$> [option]",
+"    Create a new consumer group. Options are:",
+"    * MKSTREAM",
+"      Create the empty stream if it does not exist.",
+"    * ENTRIESREAD entries_read",
+"      Set the group's entries_read counter (internal use).",
+"CREATECONSUMER <key> <groupname> <consumer>",
+"    Create a new consumer in the specified group.",
+"DELCONSUMER <key> <groupname> <consumer>",
+"    Remove the specified consumer.",
+"DESTROY <key> <groupname>",
+"    Remove the specified group.",
+"SETID <key> <groupname> <id|$> [ENTRIESREAD entries_read]",
+"    Set the current group ID and entries_read counter.",
+NULL
+        };
+        addReplyHelp(c, help);
+    } else if (!strcasecmp(opt,"CREATE") && (c->argc >= 5 && c->argc <= 8)) {
+        streamID id;
+        if (!strcmp(c->argv[4]->ptr,"$")) {
+            if (s) {
+                id = s->last_id;
+            } else {
+                id.ms = 0;
+                id.seq = 0;
+            }
+        } else if (streamParseStrictIDOrReply(c,c->argv[4],&id,0,NULL) != C_OK) {
+            return;
+        }
+
+        /* Handle the MKSTREAM option now that the command can no longer fail. */
+        if (s == NULL) {
+            serverAssert(mkstream);
+            o = createStreamObject();
+            dbAdd(c->db,c->argv[2],o);
+            s = o->ptr;
+            signalModifiedKey(c,c->db,c->argv[2]);
+        }
+
+        streamCG *cg = streamCreateCG(s,grpname,sdslen(grpname),&id,entries_read);
+        if (cg) {
+            addReply(c,shared.ok);
+            server.dirty++;
+            notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-create",
+                                c->argv[2],c->db->id);
+        } else {
+            addReplyError(c,"-BUSYGROUP Consumer Group name already exists");
+        }
+    } else if (!strcasecmp(opt,"SETID") && (c->argc == 5 || c->argc == 7)) {
+        streamID id;
+        if (!strcmp(c->argv[4]->ptr,"$")) {
+            id = s->last_id;
+        } else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) {
+            return;
+        }
+        cg->last_id = id;
+        cg->entries_read = entries_read;
+        addReply(c,shared.ok);
+        server.dirty++;
+        notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-setid",c->argv[2],c->db->id);
+    } else if (!strcasecmp(opt,"DESTROY") && c->argc == 4) {
+        if (cg) {
+            raxRemove(s->cgroups,(unsigned char*)grpname,sdslen(grpname),NULL);
+            streamFreeCG(cg);
+            addReply(c,shared.cone);
+            server.dirty++;
+            notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-destroy",
+                                c->argv[2],c->db->id);
+            /* We want to unblock any XREADGROUP consumers with -NOGROUP. */
+            signalKeyAsReady(c->db,c->argv[2],OBJ_STREAM);
+        } else {
+            addReply(c,shared.czero);
+        }
+    } else if (!strcasecmp(opt,"CREATECONSUMER") && c->argc == 5) {
+        streamConsumer *created = streamCreateConsumer(cg,c->argv[4]->ptr,c->argv[2],
+                                                       c->db->id,SCC_DEFAULT);
+        addReplyLongLong(c,created ? 1 : 0);
+    } else if (!strcasecmp(opt,"DELCONSUMER") && c->argc == 5) {
+        long long pending = 0;
+        streamConsumer *consumer = streamLookupConsumer(cg,c->argv[4]->ptr);
+        if (consumer) {
+            /* Delete the consumer and returns the number of pending messages
+             * that were yet associated with such a consumer. */
+            pending = raxSize(consumer->pel);
+            streamDelConsumer(cg,consumer);
+            server.dirty++;
+            notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-delconsumer",
+                                c->argv[2],c->db->id);
+        }
+        addReplyLongLong(c,pending);
+    } else {
+        addReplySubcommandSyntaxError(c);
+    }
+}
+
+/* XSETID <stream> <id> [ENTRIESADDED entries_added] [MAXDELETEDID max_deleted_entry_id]
+ *
+ * Set the internal "last ID", "added entries" and "maximal deleted entry ID"
+ * of a stream. */
+void xsetidCommand(client *c) {
+    streamID id, max_xdel_id = {0, 0};
+    long long entries_added = -1;
+
+    if (streamParseStrictIDOrReply(c,c->argv[2],&id,0,NULL) != C_OK)
+        return;
+
+    int i = 3;
+    while (i < c->argc) {
+        int moreargs = (c->argc-1) - i; /* Number of additional arguments. */
+        char *opt = c->argv[i]->ptr;
+        if (!strcasecmp(opt,"ENTRIESADDED") && moreargs) {
+            if (getLongLongFromObjectOrReply(c,c->argv[i+1],&entries_added,NULL) != C_OK) {
+                return;
+            } else if (entries_added < 0) {
+                addReplyError(c,"entries_added must be positive");
+                return;
+            }
+            i += 2;
+        } else if (!strcasecmp(opt,"MAXDELETEDID") && moreargs) {
+            if (streamParseStrictIDOrReply(c,c->argv[i+1],&max_xdel_id,0,NULL) != C_OK) {
+                return;
+            } else if (streamCompareID(&id,&max_xdel_id) < 0) {
+                addReplyError(c,"The ID specified in XSETID is smaller than the provided max_deleted_entry_id");
+                return;
+            }
+            i += 2;
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return;
+        }
+    }
+
+    robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr);
+    if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
+    stream *s = o->ptr;
+
+    if (streamCompareID(&id,&s->max_deleted_entry_id) < 0) {
+        addReplyError(c,"The ID specified in XSETID is smaller than current max_deleted_entry_id");
+        return;
+    }
+
+    /* If the stream has at least one item, we want to check that the user
+     * is setting a last ID that is equal or greater than the current top
+     * item, otherwise the fundamental ID monotonicity assumption is violated. */
+    if (s->length > 0) {
+        streamID maxid;
+        streamLastValidID(s,&maxid);
+
+        if (streamCompareID(&id,&maxid) < 0) {
+            addReplyError(c,"The ID specified in XSETID is smaller than the target stream top item");
+            return;
+        }
+
+        /* If an entries_added was provided, it can't be lower than the length. */
+        if (entries_added != -1 && s->length > (uint64_t)entries_added) {
+            addReplyError(c,"The entries_added specified in XSETID is smaller than the target stream length");
+            return;
+        }
+    }
+
+    s->last_id = id;
+    if (entries_added != -1)
+        s->entries_added = entries_added;
+    if (!streamIDEqZero(&max_xdel_id))
+        s->max_deleted_entry_id = max_xdel_id;
+    addReply(c,shared.ok);
+    server.dirty++;
+    notifyKeyspaceEvent(NOTIFY_STREAM,"xsetid",c->argv[1],c->db->id);
+}
+
+/* XACK <key> <group> <id> <id> ... <id>
+ * Acknowledge a message as processed. In practical terms we just check the
+ * pending entries list (PEL) of the group, and delete the PEL entry both from
+ * the group and the consumer (pending messages are referenced in both places).
+ *
+ * Return value of the command is the number of messages successfully
+ * acknowledged, that is, the IDs we were actually able to resolve in the PEL.
+ */
+void xackCommand(client *c) {
+    streamCG *group = NULL;
+    robj *o = lookupKeyRead(c->db,c->argv[1]);
+    if (o) {
+        if (checkType(c,o,OBJ_STREAM)) return; /* Type error. */
+        group = streamLookupCG(o->ptr,c->argv[2]->ptr);
+    }
+
+    /* No key or group? Nothing to ack. */
+    if (o == NULL || group == NULL) {
+        addReply(c,shared.czero);
+        return;
+    }
+
+    /* Start parsing the IDs, so that we abort ASAP if there is a syntax
+     * error: the return value of this command cannot be an error in case
+     * the client successfully acknowledged some messages, so it should be
+     * executed in a "all or nothing" fashion. */
+    streamID static_ids[STREAMID_STATIC_VECTOR_LEN];
+    streamID *ids = static_ids;
+    int id_count = c->argc-3;
+    if (id_count > STREAMID_STATIC_VECTOR_LEN)
+        ids = zmalloc(sizeof(streamID)*id_count);
+    for (int j = 3; j < c->argc; j++) {
+        if (streamParseStrictIDOrReply(c,c->argv[j],&ids[j-3],0,NULL) != C_OK) goto cleanup;
+    }
+
+    int acknowledged = 0;
+    for (int j = 3; j < c->argc; j++) {
+        unsigned char buf[sizeof(streamID)];
+        streamEncodeID(buf,&ids[j-3]);
+
+        /* Lookup the ID in the group PEL: it will have a reference to the
+         * NACK structure that will have a reference to the consumer, so that
+         * we are able to remove the entry from both PELs. */
+        streamNACK *nack = raxFind(group->pel,buf,sizeof(buf));
+        if (nack != raxNotFound) {
+            raxRemove(group->pel,buf,sizeof(buf),NULL);
+            raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
+            streamFreeNACK(nack);
+            acknowledged++;
+            server.dirty++;
+        }
+    }
+    addReplyLongLong(c,acknowledged);
+cleanup:
+    if (ids != static_ids) zfree(ids);
+}
+
+/* XPENDING <key> <group> [[IDLE <idle>] <start> <stop> <count> [<consumer>]]
+ *
+ * If start and stop are omitted, the command just outputs information about
+ * the amount of pending messages for the key/group pair, together with
+ * the minimum and maximum ID of pending messages.
+ *
+ * If start and stop are provided instead, the pending messages are returned
+ * with information about the current owner, number of deliveries and last
+ * delivery time and so forth. */
+void xpendingCommand(client *c) {
+    int justinfo = c->argc == 3; /* Without the range just outputs general
+                                    information about the PEL. */
+    robj *key = c->argv[1];
+    robj *groupname = c->argv[2];
+    robj *consumername = NULL;
+    streamID startid, endid;
+    long long count = 0;
+    long long minidle = 0;
+    int startex = 0, endex = 0;
+
+    /* Start and stop, and the consumer, can be omitted. Also the IDLE modifier. */
+    if (c->argc != 3 && (c->argc < 6 || c->argc > 9)) {
+        addReplyErrorObject(c,shared.syntaxerr);
+        return;
+    }
+
+    /* Parse start/end/count arguments ASAP if needed, in order to report
+     * syntax errors before any other error. */
+    if (c->argc >= 6) {
+        int startidx = 3; /* Without IDLE */
+
+        if (!strcasecmp(c->argv[3]->ptr, "IDLE")) {
+            if (getLongLongFromObjectOrReply(c, c->argv[4], &minidle, NULL) == C_ERR)
+                return;
+            if (c->argc < 8) {
+                /* If IDLE was provided we must have at least 'start end count' */
+                addReplyErrorObject(c,shared.syntaxerr);
+                return;
+            }
+            /* Search for rest of arguments after 'IDLE <idle>' */
+            startidx += 2;
+        }
+
+        /* count argument. */
+        if (getLongLongFromObjectOrReply(c,c->argv[startidx+2],&count,NULL) == C_ERR)
+            return;
+        if (count < 0) count = 0;
+
+        /* start and end arguments. */
+        if (streamParseIntervalIDOrReply(c,c->argv[startidx],&startid,&startex,0) != C_OK)
+            return;
+        if (startex && streamIncrID(&startid) != C_OK) {
+            addReplyError(c,"invalid start ID for the interval");
+            return;
+        }
+        if (streamParseIntervalIDOrReply(c,c->argv[startidx+1],&endid,&endex,UINT64_MAX) != C_OK)
+            return;
+        if (endex && streamDecrID(&endid) != C_OK) {
+            addReplyError(c,"invalid end ID for the interval");
+            return;
+        }
+
+        if (startidx+3 < c->argc) {
+            /* 'consumer' was provided */
+            consumername = c->argv[startidx+3];
+        }
+    }
+
+    /* Lookup the key and the group inside the stream. */
+    robj *o = lookupKeyRead(c->db,c->argv[1]);
+    streamCG *group;
+
+    if (checkType(c,o,OBJ_STREAM)) return;
+    if (o == NULL ||
+        (group = streamLookupCG(o->ptr,groupname->ptr)) == NULL)
+    {
+        addReplyErrorFormat(c, "-NOGROUP No such key '%s' or consumer "
+                               "group '%s'",
+                               (char*)key->ptr,(char*)groupname->ptr);
+        return;
+    }
+
+    /* XPENDING <key> <group> variant. */
+    if (justinfo) {
+        addReplyArrayLen(c,4);
+        /* Total number of messages in the PEL. */
+        addReplyLongLong(c,raxSize(group->pel));
+        /* First and last IDs. */
+        if (raxSize(group->pel) == 0) {
+            addReplyNull(c); /* Start. */
+            addReplyNull(c); /* End. */
+            addReplyNullArray(c); /* Clients. */
+        } else {
+            /* Start. */
+            raxIterator ri;
+            raxStart(&ri,group->pel);
+            raxSeek(&ri,"^",NULL,0);
+            raxNext(&ri);
+            streamDecodeID(ri.key,&startid);
+            addReplyStreamID(c,&startid);
+
+            /* End. */
+            raxSeek(&ri,"$",NULL,0);
+            raxNext(&ri);
+            streamDecodeID(ri.key,&endid);
+            addReplyStreamID(c,&endid);
+            raxStop(&ri);
+
+            /* Consumers with pending messages. */
+            raxStart(&ri,group->consumers);
+            raxSeek(&ri,"^",NULL,0);
+            void *arraylen_ptr = addReplyDeferredLen(c);
+            size_t arraylen = 0;
+            while(raxNext(&ri)) {
+                streamConsumer *consumer = ri.data;
+                if (raxSize(consumer->pel) == 0) continue;
+                addReplyArrayLen(c,2);
+                addReplyBulkCBuffer(c,ri.key,ri.key_len);
+                addReplyBulkLongLong(c,raxSize(consumer->pel));
+                arraylen++;
+            }
+            setDeferredArrayLen(c,arraylen_ptr,arraylen);
+            raxStop(&ri);
+        }
+    } else { /* <start>, <stop> and <count> provided, return actual pending entries (not just info) */
+        streamConsumer *consumer = NULL;
+        if (consumername) {
+            consumer = streamLookupConsumer(group,consumername->ptr);
+
+            /* If a consumer name was mentioned but it does not exist, we can
+             * just return an empty array. */
+            if (consumer == NULL) {
+                addReplyArrayLen(c,0);
+                return;
+            }
+        }
+
+        rax *pel = consumer ? consumer->pel : group->pel;
+        unsigned char startkey[sizeof(streamID)];
+        unsigned char endkey[sizeof(streamID)];
+        raxIterator ri;
+        mstime_t now = commandTimeSnapshot();
+
+        streamEncodeID(startkey,&startid);
+        streamEncodeID(endkey,&endid);
+        raxStart(&ri,pel);
+        raxSeek(&ri,">=",startkey,sizeof(startkey));
+        void *arraylen_ptr = addReplyDeferredLen(c);
+        size_t arraylen = 0;
+
+        while(count && raxNext(&ri) && memcmp(ri.key,endkey,ri.key_len) <= 0) {
+            streamNACK *nack = ri.data;
+
+            if (minidle) {
+                mstime_t this_idle = now - nack->delivery_time;
+                if (this_idle < minidle) continue;
+            }
+
+            arraylen++;
+            count--;
+            addReplyArrayLen(c,4);
+
+            /* Entry ID. */
+            streamID id;
+            streamDecodeID(ri.key,&id);
+            addReplyStreamID(c,&id);
+
+            /* Consumer name. */
+            addReplyBulkCBuffer(c,nack->consumer->name,
+                                sdslen(nack->consumer->name));
+
+            /* Milliseconds elapsed since last delivery. */
+            mstime_t elapsed = now - nack->delivery_time;
+            if (elapsed < 0) elapsed = 0;
+            addReplyLongLong(c,elapsed);
+
+            /* Number of deliveries. */
+            addReplyLongLong(c,nack->delivery_count);
+        }
+        raxStop(&ri);
+        setDeferredArrayLen(c,arraylen_ptr,arraylen);
+    }
+}
+
+/* XCLAIM <key> <group> <consumer> <min-idle-time> <ID-1> <ID-2>
+ *        [IDLE <milliseconds>] [TIME <mstime>] [RETRYCOUNT <count>]
+ *        [FORCE] [JUSTID]
+ *
+ * Changes ownership of one or multiple messages in the Pending Entries List
+ * of a given stream consumer group.
+ *
+ * If the message ID (among the specified ones) exists, and its idle
+ * time greater or equal to <min-idle-time>, then the message new owner
+ * becomes the specified <consumer>. If the minimum idle time specified
+ * is zero, messages are claimed regardless of their idle time.
+ *
+ * All the messages that cannot be found inside the pending entries list
+ * are ignored, but in case the FORCE option is used. In that case we
+ * create the NACK (representing a not yet acknowledged message) entry in
+ * the consumer group PEL.
+ *
+ * This command creates the consumer as side effect if it does not yet
+ * exists. Moreover the command reset the idle time of the message to 0,
+ * even if by using the IDLE or TIME options, the user can control the
+ * new idle time.
+ *
+ * The options at the end can be used in order to specify more attributes
+ * to set in the representation of the pending message:
+ *
+ * 1. IDLE <ms>:
+ *      Set the idle time (last time it was delivered) of the message.
+ *      If IDLE is not specified, an IDLE of 0 is assumed, that is,
+ *      the time count is reset because the message has now a new
+ *      owner trying to process it.
+ *
+ * 2. TIME <ms-unix-time>:
+ *      This is the same as IDLE but instead of a relative amount of
+ *      milliseconds, it sets the idle time to a specific unix time
+ *      (in milliseconds). This is useful in order to rewrite the AOF
+ *      file generating XCLAIM commands.
+ *
+ * 3. RETRYCOUNT <count>:
+ *      Set the retry counter to the specified value. This counter is
+ *      incremented every time a message is delivered again. Normally
+ *      XCLAIM does not alter this counter, which is just served to clients
+ *      when the XPENDING command is called: this way clients can detect
+ *      anomalies, like messages that are never processed for some reason
+ *      after a big number of delivery attempts.
+ *
+ * 4. FORCE:
+ *      Creates the pending message entry in the PEL even if certain
+ *      specified IDs are not already in the PEL assigned to a different
+ *      client. However the message must be exist in the stream, otherwise
+ *      the IDs of non existing messages are ignored.
+ *
+ * 5. JUSTID:
+ *      Return just an array of IDs of messages successfully claimed,
+ *      without returning the actual message.
+ *
+ * 6. LASTID <id>:
+ *      Update the consumer group last ID with the specified ID if the
+ *      current last ID is smaller than the provided one.
+ *      This is used for replication / AOF, so that when we read from a
+ *      consumer group, the XCLAIM that gets propagated to give ownership
+ *      to the consumer, is also used in order to update the group current
+ *      ID.
+ *
+ * The command returns an array of messages that the user
+ * successfully claimed, so that the caller is able to understand
+ * what messages it is now in charge of. */
+void xclaimCommand(client *c) {
+    streamCG *group = NULL;
+    robj *o = lookupKeyRead(c->db,c->argv[1]);
+    long long minidle; /* Minimum idle time argument. */
+    long long retrycount = -1;   /* -1 means RETRYCOUNT option not given. */
+    mstime_t deliverytime = -1;  /* -1 means IDLE/TIME options not given. */
+    int force = 0;
+    int justid = 0;
+
+    if (o) {
+        if (checkType(c,o,OBJ_STREAM)) return; /* Type error. */
+        group = streamLookupCG(o->ptr,c->argv[2]->ptr);
+    }
+
+    /* No key or group? Send an error given that the group creation
+     * is mandatory. */
+    if (o == NULL || group == NULL) {
+        addReplyErrorFormat(c,"-NOGROUP No such key '%s' or "
+                              "consumer group '%s'", (char*)c->argv[1]->ptr,
+                              (char*)c->argv[2]->ptr);
+        return;
+    }
+
+    if (getLongLongFromObjectOrReply(c,c->argv[4],&minidle,
+        "Invalid min-idle-time argument for XCLAIM")
+        != C_OK) return;
+    if (minidle < 0) minidle = 0;
+
+    /* Start parsing the IDs, so that we abort ASAP if there is a syntax
+     * error: the return value of this command cannot be an error in case
+     * the client successfully claimed some message, so it should be
+     * executed in a "all or nothing" fashion. */
+    int j;
+    streamID static_ids[STREAMID_STATIC_VECTOR_LEN];
+    streamID *ids = static_ids;
+    int id_count = c->argc-5;
+    if (id_count > STREAMID_STATIC_VECTOR_LEN)
+        ids = zmalloc(sizeof(streamID)*id_count);
+    for (j = 5; j < c->argc; j++) {
+        if (streamParseStrictIDOrReply(NULL,c->argv[j],&ids[j-5],0,NULL) != C_OK) break;
+    }
+    int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */
+
+    /* If we stopped because some IDs cannot be parsed, perhaps they
+     * are trailing options. */
+    mstime_t now = commandTimeSnapshot();
+    streamID last_id = {0,0};
+    int propagate_last_id = 0;
+    for (; j < c->argc; j++) {
+        int moreargs = (c->argc-1) - j; /* Number of additional arguments. */
+        char *opt = c->argv[j]->ptr;
+        if (!strcasecmp(opt,"FORCE")) {
+            force = 1;
+        } else if (!strcasecmp(opt,"JUSTID")) {
+            justid = 1;
+        } else if (!strcasecmp(opt,"IDLE") && moreargs) {
+            j++;
+            if (getLongLongFromObjectOrReply(c,c->argv[j],&deliverytime,
+                "Invalid IDLE option argument for XCLAIM")
+                != C_OK) goto cleanup;
+            deliverytime = now - deliverytime;
+        } else if (!strcasecmp(opt,"TIME") && moreargs) {
+            j++;
+            if (getLongLongFromObjectOrReply(c,c->argv[j],&deliverytime,
+                "Invalid TIME option argument for XCLAIM")
+                != C_OK) goto cleanup;
+        } else if (!strcasecmp(opt,"RETRYCOUNT") && moreargs) {
+            j++;
+            if (getLongLongFromObjectOrReply(c,c->argv[j],&retrycount,
+                "Invalid RETRYCOUNT option argument for XCLAIM")
+                != C_OK) goto cleanup;
+        } else if (!strcasecmp(opt,"LASTID") && moreargs) {
+            j++;
+            if (streamParseStrictIDOrReply(c,c->argv[j],&last_id,0,NULL) != C_OK) goto cleanup;
+        } else {
+            addReplyErrorFormat(c,"Unrecognized XCLAIM option '%s'",opt);
+            goto cleanup;
+        }
+    }
+
+    if (streamCompareID(&last_id,&group->last_id) > 0) {
+        group->last_id = last_id;
+        propagate_last_id = 1;
+    }
+
+    if (deliverytime != -1) {
+        /* If a delivery time was passed, either with IDLE or TIME, we
+         * do some sanity check on it, and set the deliverytime to now
+         * (which is a sane choice usually) if the value is bogus.
+         * To raise an error here is not wise because clients may compute
+         * the idle time doing some math starting from their local time,
+         * and this is not a good excuse to fail in case, for instance,
+         * the computer time is a bit in the future from our POV. */
+        if (deliverytime < 0 || deliverytime > now) deliverytime = now;
+    } else {
+        /* If no IDLE/TIME option was passed, we want the last delivery
+         * time to be now, so that the idle time of the message will be
+         * zero. */
+        deliverytime = now;
+    }
+
+    /* Do the actual claiming. */
+    streamConsumer *consumer = streamLookupConsumer(group,c->argv[3]->ptr);
+    if (consumer == NULL) {
+        consumer = streamCreateConsumer(group,c->argv[3]->ptr,c->argv[1],c->db->id,SCC_DEFAULT);
+    }
+    consumer->seen_time = commandTimeSnapshot();
+
+    void *arraylenptr = addReplyDeferredLen(c);
+    size_t arraylen = 0;
+    for (int j = 5; j <= last_id_arg; j++) {
+        streamID id = ids[j-5];
+        unsigned char buf[sizeof(streamID)];
+        streamEncodeID(buf,&id);
+
+        /* Lookup the ID in the group PEL. */
+        streamNACK *nack = raxFind(group->pel,buf,sizeof(buf));
+
+        /* Item must exist for us to transfer it to another consumer. */
+        if (!streamEntryExists(o->ptr,&id)) {
+            /* Clear this entry from the PEL, it no longer exists */
+            if (nack != raxNotFound) {
+                /* Propagate this change (we are going to delete the NACK). */
+                streamPropagateXCLAIM(c,c->argv[1],group,c->argv[2],c->argv[j],nack);
+                propagate_last_id = 0; /* Will be propagated by XCLAIM itself. */
+                server.dirty++;
+                /* Release the NACK */
+                raxRemove(group->pel,buf,sizeof(buf),NULL);
+                raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
+                streamFreeNACK(nack);
+            }
+            continue;
+        }
+
+        /* If FORCE is passed, let's check if at least the entry
+         * exists in the Stream. In such case, we'll create a new
+         * entry in the PEL from scratch, so that XCLAIM can also
+         * be used to create entries in the PEL. Useful for AOF
+         * and replication of consumer groups. */
+        if (force && nack == raxNotFound) {
+            /* Create the NACK. */
+            nack = streamCreateNACK(NULL);
+            raxInsert(group->pel,buf,sizeof(buf),nack,NULL);
+        }
+
+        if (nack != raxNotFound) {
+            /* We need to check if the minimum idle time requested
+             * by the caller is satisfied by this entry.
+             *
+             * Note that the nack could be created by FORCE, in this
+             * case there was no pre-existing entry and minidle should
+             * be ignored, but in that case nack->consumer is NULL. */
+            if (nack->consumer && minidle) {
+                mstime_t this_idle = now - nack->delivery_time;
+                if (this_idle < minidle) continue;
+            }
+
+            if (nack->consumer != consumer) {
+                /* Remove the entry from the old consumer.
+                 * Note that nack->consumer is NULL if we created the
+                 * NACK above because of the FORCE option. */
+                if (nack->consumer)
+                    raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
+            }
+            nack->delivery_time = deliverytime;
+            /* Set the delivery attempts counter if given, otherwise
+             * autoincrement unless JUSTID option provided */
+            if (retrycount >= 0) {
+                nack->delivery_count = retrycount;
+            } else if (!justid) {
+                nack->delivery_count++;
+            }
+            if (nack->consumer != consumer) {
+                /* Add the entry in the new consumer local PEL. */
+                raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
+                nack->consumer = consumer;
+            }
+            /* Send the reply for this entry. */
+            if (justid) {
+                addReplyStreamID(c,&id);
+            } else {
+                serverAssert(streamReplyWithRange(c,o->ptr,&id,&id,1,0,NULL,NULL,STREAM_RWR_RAWENTRIES,NULL) == 1);
+            }
+            arraylen++;
+
+            consumer->active_time = commandTimeSnapshot();
+
+            /* Propagate this change. */
+            streamPropagateXCLAIM(c,c->argv[1],group,c->argv[2],c->argv[j],nack);
+            propagate_last_id = 0; /* Will be propagated by XCLAIM itself. */
+            server.dirty++;
+        }
+    }
+    if (propagate_last_id) {
+        streamPropagateGroupID(c,c->argv[1],group,c->argv[2]);
+        server.dirty++;
+    }
+    setDeferredArrayLen(c,arraylenptr,arraylen);
+    preventCommandPropagation(c);
+cleanup:
+    if (ids != static_ids) zfree(ids);
+}
+
+/* XAUTOCLAIM <key> <group> <consumer> <min-idle-time> <start> [COUNT <count>] [JUSTID]
+ *
+ * Changes ownership of one or multiple messages in the Pending Entries List
+ * of a given stream consumer group.
+ *
+ * For each PEL entry, if its idle time greater or equal to <min-idle-time>,
+ * then the message new owner becomes the specified <consumer>.
+ * If the minimum idle time specified is zero, messages are claimed
+ * regardless of their idle time.
+ *
+ * This command creates the consumer as side effect if it does not yet
+ * exists. Moreover the command reset the idle time of the message to 0.
+ *
+ * The command returns an array of messages that the user
+ * successfully claimed, so that the caller is able to understand
+ * what messages it is now in charge of. */
+void xautoclaimCommand(client *c) {
+    streamCG *group = NULL;
+    robj *o = lookupKeyRead(c->db,c->argv[1]);
+    long long minidle; /* Minimum idle time argument, in milliseconds. */
+    long count = 100; /* Maximum entries to claim. */
+    const unsigned attempts_factor = 10;
+    streamID startid;
+    int startex;
+    int justid = 0;
+
+    /* Parse idle/start/end/count arguments ASAP if needed, in order to report
+     * syntax errors before any other error. */
+    if (getLongLongFromObjectOrReply(c,c->argv[4],&minidle,"Invalid min-idle-time argument for XAUTOCLAIM") != C_OK)
+        return;
+    if (minidle < 0) minidle = 0;
+
+    if (streamParseIntervalIDOrReply(c,c->argv[5],&startid,&startex,0) != C_OK)
+        return;
+    if (startex && streamIncrID(&startid) != C_OK) {
+        addReplyError(c,"invalid start ID for the interval");
+        return;
+    }
+
+    int j = 6; /* options start at argv[6] */
+    while(j < c->argc) {
+        int moreargs = (c->argc-1) - j; /* Number of additional arguments. */
+        char *opt = c->argv[j]->ptr;
+        if (!strcasecmp(opt,"COUNT") && moreargs) {
+            long max_count = LONG_MAX / (max(sizeof(streamID), attempts_factor));
+            if (getRangeLongFromObjectOrReply(c,c->argv[j+1],1,max_count,&count,"COUNT must be > 0") != C_OK)
+                return;
+            j++;
+        } else if (!strcasecmp(opt,"JUSTID")) {
+            justid = 1;
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return;
+        }
+        j++;
+    }
+
+    if (o) {
+        if (checkType(c,o,OBJ_STREAM))
+            return; /* Type error. */
+        group = streamLookupCG(o->ptr,c->argv[2]->ptr);
+    }
+
+    /* No key or group? Send an error given that the group creation
+     * is mandatory. */
+    if (o == NULL || group == NULL) {
+        addReplyErrorFormat(c,"-NOGROUP No such key '%s' or consumer group '%s'",
+                            (char*)c->argv[1]->ptr,
+                            (char*)c->argv[2]->ptr);
+        return;
+    }
+
+    streamID *deleted_ids = ztrymalloc(count * sizeof(streamID));
+    if (!deleted_ids) {
+        addReplyError(c, "Insufficient memory, failed allocating transient memory, COUNT too high.");
+        return;
+    }
+
+    /* Do the actual claiming. */
+    streamConsumer *consumer = streamLookupConsumer(group,c->argv[3]->ptr);
+    if (consumer == NULL) {
+        consumer = streamCreateConsumer(group,c->argv[3]->ptr,c->argv[1],c->db->id,SCC_DEFAULT);
+    }
+    consumer->seen_time = commandTimeSnapshot();
+
+    long long attempts = count * attempts_factor;
+
+    addReplyArrayLen(c, 3); /* We add another reply later */
+    void *endidptr = addReplyDeferredLen(c); /* reply[0] */
+    void *arraylenptr = addReplyDeferredLen(c); /* reply[1] */
+
+    unsigned char startkey[sizeof(streamID)];
+    streamEncodeID(startkey,&startid);
+    raxIterator ri;
+    raxStart(&ri,group->pel);
+    raxSeek(&ri,">=",startkey,sizeof(startkey));
+    size_t arraylen = 0;
+    mstime_t now = commandTimeSnapshot();
+    int deleted_id_num = 0;
+    while (attempts-- && count && raxNext(&ri)) {
+        streamNACK *nack = ri.data;
+
+        streamID id;
+        streamDecodeID(ri.key, &id);
+
+        /* Item must exist for us to transfer it to another consumer. */
+        if (!streamEntryExists(o->ptr,&id)) {
+            /* Propagate this change (we are going to delete the NACK). */
+            robj *idstr = createObjectFromStreamID(&id);
+            streamPropagateXCLAIM(c,c->argv[1],group,c->argv[2],idstr,nack);
+            decrRefCount(idstr);
+            server.dirty++;
+            /* Clear this entry from the PEL, it no longer exists */
+            raxRemove(group->pel,ri.key,ri.key_len,NULL);
+            raxRemove(nack->consumer->pel,ri.key,ri.key_len,NULL);
+            streamFreeNACK(nack);
+            /* Remember the ID for later */
+            deleted_ids[deleted_id_num++] = id;
+            raxSeek(&ri,">=",ri.key,ri.key_len);
+            count--; /* Count is a limit of the command response size. */
+            continue;
+        }
+
+        if (minidle) {
+            mstime_t this_idle = now - nack->delivery_time;
+            if (this_idle < minidle)
+                continue;
+        }
+
+        if (nack->consumer != consumer) {
+            /* Remove the entry from the old consumer.
+             * Note that nack->consumer is NULL if we created the
+             * NACK above because of the FORCE option. */
+            if (nack->consumer)
+                raxRemove(nack->consumer->pel,ri.key,ri.key_len,NULL);
+        }
+
+        /* Update the consumer and idle time. */
+        nack->delivery_time = now;
+        /* Increment the delivery attempts counter unless JUSTID option provided */
+        if (!justid)
+            nack->delivery_count++;
+
+        if (nack->consumer != consumer) {
+            /* Add the entry in the new consumer local PEL. */
+            raxInsert(consumer->pel,ri.key,ri.key_len,nack,NULL);
+            nack->consumer = consumer;
+        }
+
+        /* Send the reply for this entry. */
+        if (justid) {
+            addReplyStreamID(c,&id);
+        } else {
+            serverAssert(streamReplyWithRange(c,o->ptr,&id,&id,1,0,NULL,NULL,STREAM_RWR_RAWENTRIES,NULL) == 1);
+        }
+        arraylen++;
+        count--;
+
+        consumer->active_time = commandTimeSnapshot();
+
+        /* Propagate this change. */
+        robj *idstr = createObjectFromStreamID(&id);
+        streamPropagateXCLAIM(c,c->argv[1],group,c->argv[2],idstr,nack);
+        decrRefCount(idstr);
+        server.dirty++;
+    }
+
+    /* We need to return the next entry as a cursor for the next XAUTOCLAIM call */
+    raxNext(&ri);
+
+    streamID endid;
+    if (raxEOF(&ri)) {
+        endid.ms = endid.seq = 0;
+    } else {
+        streamDecodeID(ri.key, &endid);
+    }
+    raxStop(&ri);
+
+    setDeferredArrayLen(c,arraylenptr,arraylen);
+    setDeferredReplyStreamID(c,endidptr,&endid);
+
+    addReplyArrayLen(c, deleted_id_num); /* reply[2] */
+    for (int i = 0; i < deleted_id_num; i++) {
+        addReplyStreamID(c, &deleted_ids[i]);
+    }
+    zfree(deleted_ids);
+
+    preventCommandPropagation(c);
+}
+
+/* XDEL <key> [<ID1> <ID2> ... <IDN>]
+ *
+ * Removes the specified entries from the stream. Returns the number
+ * of items actually deleted, that may be different from the number
+ * of IDs passed in case certain IDs do not exist. */
+void xdelCommand(client *c) {
+    robj *o;
+
+    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
+        || checkType(c,o,OBJ_STREAM)) return;
+    stream *s = o->ptr;
+
+    /* We need to sanity check the IDs passed to start. Even if not
+     * a big issue, it is not great that the command is only partially
+     * executed because at some point an invalid ID is parsed. */
+    streamID static_ids[STREAMID_STATIC_VECTOR_LEN];
+    streamID *ids = static_ids;
+    int id_count = c->argc-2;
+    if (id_count > STREAMID_STATIC_VECTOR_LEN)
+        ids = zmalloc(sizeof(streamID)*id_count);
+    for (int j = 2; j < c->argc; j++) {
+        if (streamParseStrictIDOrReply(c,c->argv[j],&ids[j-2],0,NULL) != C_OK) goto cleanup;
+    }
+
+    /* Actually apply the command. */
+    int deleted = 0;
+    int first_entry = 0;
+    for (int j = 2; j < c->argc; j++) {
+        streamID *id = &ids[j-2];
+        if (streamDeleteItem(s,id)) {
+            /* We want to know if the first entry in the stream was deleted
+             * so we can later set the new one. */
+            if (streamCompareID(id,&s->first_id) == 0) {
+                first_entry = 1;
+            }
+            /* Update the stream's maximal tombstone if needed. */
+            if (streamCompareID(id,&s->max_deleted_entry_id) > 0) {
+                s->max_deleted_entry_id = *id;
+            }
+            deleted++;
+        };
+    }
+
+    /* Update the stream's first ID. */
+    if (deleted) {
+        if (s->length == 0) {
+            s->first_id.ms = 0;
+            s->first_id.seq = 0;
+        } else if (first_entry) {
+            streamGetEdgeID(s,1,1,&s->first_id);
+        }
+    }
+
+    /* Propagate the write if needed. */
+    if (deleted) {
+        signalModifiedKey(c,c->db,c->argv[1]);
+        notifyKeyspaceEvent(NOTIFY_STREAM,"xdel",c->argv[1],c->db->id);
+        server.dirty += deleted;
+    }
+    addReplyLongLong(c,deleted);
+cleanup:
+    if (ids != static_ids) zfree(ids);
+}
+
+/* General form: XTRIM <key> [... options ...]
+ *
+ * List of options:
+ *
+ * Trim strategies:
+ *
+ * MAXLEN [~|=] <count>     -- Trim so that the stream will be capped at
+ *                             the specified length. Use ~ before the
+ *                             count in order to demand approximated trimming
+ *                             (like XADD MAXLEN option).
+ * MINID [~|=] <id>         -- Trim so that the stream will not contain entries
+ *                             with IDs smaller than 'id'. Use ~ before the
+ *                             count in order to demand approximated trimming
+ *                             (like XADD MINID option).
+ *
+ * Other options:
+ *
+ * LIMIT <entries>          -- The maximum number of entries to trim.
+ *                             0 means unlimited. Unless specified, it is set
+ *                             to a default of 100*server.stream_node_max_entries,
+ *                             and that's in order to keep the trimming time sane.
+ *                             Has meaning only if `~` was provided.
+ */
+void xtrimCommand(client *c) {
+    robj *o;
+
+    /* Argument parsing. */
+    streamAddTrimArgs parsed_args;
+    if (streamParseAddOrTrimArgsOrReply(c, &parsed_args, 0) < 0)
+        return; /* streamParseAddOrTrimArgsOrReply already replied. */
+
+    /* If the key does not exist, we are ok returning zero, that is, the
+     * number of elements removed from the stream. */
+    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
+        || checkType(c,o,OBJ_STREAM)) return;
+    stream *s = o->ptr;
+
+    /* Perform the trimming. */
+    int64_t deleted = streamTrim(s, &parsed_args);
+    if (deleted) {
+        notifyKeyspaceEvent(NOTIFY_STREAM,"xtrim",c->argv[1],c->db->id);
+        if (parsed_args.approx_trim) {
+            /* In case our trimming was limited (by LIMIT or by ~) we must
+             * re-write the relevant trim argument to make sure there will be
+             * no inconsistencies in AOF loading or in the replica.
+             * It's enough to check only args->approx because there is no
+             * way LIMIT is given without the ~ option. */
+            streamRewriteApproxSpecifier(c,parsed_args.trim_strategy_arg_idx-1);
+            streamRewriteTrimArgument(c,s,parsed_args.trim_strategy,parsed_args.trim_strategy_arg_idx);
+        }
+
+        /* Propagate the write. */
+        signalModifiedKey(c, c->db,c->argv[1]);
+        server.dirty += deleted;
+    }
+    addReplyLongLong(c,deleted);
+}
+
+/* Helper function for xinfoCommand.
+ * Handles the variants of XINFO STREAM */
+void xinfoReplyWithStreamInfo(client *c, stream *s) {
+    int full = 1;
+    long long count = 10; /* Default COUNT is 10 so we don't block the server */
+    robj **optv = c->argv + 3; /* Options start after XINFO STREAM <key> */
+    int optc = c->argc - 3;
+
+    /* Parse options. */
+    if (optc == 0) {
+        full = 0;
+    } else {
+        /* Valid options are [FULL] or [FULL COUNT <count>] */
+        if (optc != 1 && optc != 3) {
+            addReplySubcommandSyntaxError(c);
+            return;
+        }
+
+        /* First option must be "FULL" */
+        if (strcasecmp(optv[0]->ptr,"full")) {
+            addReplySubcommandSyntaxError(c);
+            return;
+        }
+
+        if (optc == 3) {
+            /* First option must be "FULL" */
+            if (strcasecmp(optv[1]->ptr,"count")) {
+                addReplySubcommandSyntaxError(c);
+                return;
+            }
+            if (getLongLongFromObjectOrReply(c,optv[2],&count,NULL) == C_ERR)
+                return;
+            if (count < 0) count = 10;
+        }
+    }
+
+    addReplyMapLen(c,full ? 9 : 10);
+    addReplyBulkCString(c,"length");
+    addReplyLongLong(c,s->length);
+    addReplyBulkCString(c,"radix-tree-keys");
+    addReplyLongLong(c,raxSize(s->rax));
+    addReplyBulkCString(c,"radix-tree-nodes");
+    addReplyLongLong(c,s->rax->numnodes);
+    addReplyBulkCString(c,"last-generated-id");
+    addReplyStreamID(c,&s->last_id);
+    addReplyBulkCString(c,"max-deleted-entry-id");
+    addReplyStreamID(c,&s->max_deleted_entry_id);
+    addReplyBulkCString(c,"entries-added");
+    addReplyLongLong(c,s->entries_added);
+    addReplyBulkCString(c,"recorded-first-entry-id");
+    addReplyStreamID(c,&s->first_id);
+
+    if (!full) {
+        /* XINFO STREAM <key> */
+
+        addReplyBulkCString(c,"groups");
+        addReplyLongLong(c,s->cgroups ? raxSize(s->cgroups) : 0);
+
+        /* To emit the first/last entry we use streamReplyWithRange(). */
+        int emitted;
+        streamID start, end;
+        start.ms = start.seq = 0;
+        end.ms = end.seq = UINT64_MAX;
+        addReplyBulkCString(c,"first-entry");
+        emitted = streamReplyWithRange(c,s,&start,&end,1,0,NULL,NULL,
+                                       STREAM_RWR_RAWENTRIES,NULL);
+        if (!emitted) addReplyNull(c);
+        addReplyBulkCString(c,"last-entry");
+        emitted = streamReplyWithRange(c,s,&start,&end,1,1,NULL,NULL,
+                                       STREAM_RWR_RAWENTRIES,NULL);
+        if (!emitted) addReplyNull(c);
+    } else {
+        /* XINFO STREAM <key> FULL [COUNT <count>] */
+
+        /* Stream entries */
+        addReplyBulkCString(c,"entries");
+        streamReplyWithRange(c,s,NULL,NULL,count,0,NULL,NULL,0,NULL);
+
+        /* Consumer groups */
+        addReplyBulkCString(c,"groups");
+        if (s->cgroups == NULL) {
+            addReplyArrayLen(c,0);
+        } else {
+            addReplyArrayLen(c,raxSize(s->cgroups));
+            raxIterator ri_cgroups;
+            raxStart(&ri_cgroups,s->cgroups);
+            raxSeek(&ri_cgroups,"^",NULL,0);
+            while(raxNext(&ri_cgroups)) {
+                streamCG *cg = ri_cgroups.data;
+                addReplyMapLen(c,7);
+
+                /* Name */
+                addReplyBulkCString(c,"name");
+                addReplyBulkCBuffer(c,ri_cgroups.key,ri_cgroups.key_len);
+
+                /* Last delivered ID */
+                addReplyBulkCString(c,"last-delivered-id");
+                addReplyStreamID(c,&cg->last_id);
+
+                /* Read counter of the last delivered ID */
+                addReplyBulkCString(c,"entries-read");
+                if (cg->entries_read != SCG_INVALID_ENTRIES_READ) {
+                    addReplyLongLong(c,cg->entries_read);
+                } else {
+                    addReplyNull(c);
+                }
+
+                /* Group lag */
+                addReplyBulkCString(c,"lag");
+                streamReplyWithCGLag(c,s,cg);
+
+                /* Group PEL count */
+                addReplyBulkCString(c,"pel-count");
+                addReplyLongLong(c,raxSize(cg->pel));
+
+                /* Group PEL */
+                addReplyBulkCString(c,"pending");
+                long long arraylen_cg_pel = 0;
+                void *arrayptr_cg_pel = addReplyDeferredLen(c);
+                raxIterator ri_cg_pel;
+                raxStart(&ri_cg_pel,cg->pel);
+                raxSeek(&ri_cg_pel,"^",NULL,0);
+                while(raxNext(&ri_cg_pel) && (!count || arraylen_cg_pel < count)) {
+                    streamNACK *nack = ri_cg_pel.data;
+                    addReplyArrayLen(c,4);
+
+                    /* Entry ID. */
+                    streamID id;
+                    streamDecodeID(ri_cg_pel.key,&id);
+                    addReplyStreamID(c,&id);
+
+                    /* Consumer name. */
+                    serverAssert(nack->consumer); /* assertion for valgrind (avoid NPD) */
+                    addReplyBulkCBuffer(c,nack->consumer->name,
+                                        sdslen(nack->consumer->name));
+
+                    /* Last delivery. */
+                    addReplyLongLong(c,nack->delivery_time);
+
+                    /* Number of deliveries. */
+                    addReplyLongLong(c,nack->delivery_count);
+
+                    arraylen_cg_pel++;
+                }
+                setDeferredArrayLen(c,arrayptr_cg_pel,arraylen_cg_pel);
+                raxStop(&ri_cg_pel);
+
+                /* Consumers */
+                addReplyBulkCString(c,"consumers");
+                addReplyArrayLen(c,raxSize(cg->consumers));
+                raxIterator ri_consumers;
+                raxStart(&ri_consumers,cg->consumers);
+                raxSeek(&ri_consumers,"^",NULL,0);
+                while(raxNext(&ri_consumers)) {
+                    streamConsumer *consumer = ri_consumers.data;
+                    addReplyMapLen(c,5);
+
+                    /* Consumer name */
+                    addReplyBulkCString(c,"name");
+                    addReplyBulkCBuffer(c,consumer->name,sdslen(consumer->name));
+
+                    /* Seen-time */
+                    addReplyBulkCString(c,"seen-time");
+                    addReplyLongLong(c,consumer->seen_time);
+
+                    /* Active-time */
+                    addReplyBulkCString(c,"active-time");
+                    addReplyLongLong(c,consumer->active_time);
+
+                    /* Consumer PEL count */
+                    addReplyBulkCString(c,"pel-count");
+                    addReplyLongLong(c,raxSize(consumer->pel));
+
+                    /* Consumer PEL */
+                    addReplyBulkCString(c,"pending");
+                    long long arraylen_cpel = 0;
+                    void *arrayptr_cpel = addReplyDeferredLen(c);
+                    raxIterator ri_cpel;
+                    raxStart(&ri_cpel,consumer->pel);
+                    raxSeek(&ri_cpel,"^",NULL,0);
+                    while(raxNext(&ri_cpel) && (!count || arraylen_cpel < count)) {
+                        streamNACK *nack = ri_cpel.data;
+                        addReplyArrayLen(c,3);
+
+                        /* Entry ID. */
+                        streamID id;
+                        streamDecodeID(ri_cpel.key,&id);
+                        addReplyStreamID(c,&id);
+
+                        /* Last delivery. */
+                        addReplyLongLong(c,nack->delivery_time);
+
+                        /* Number of deliveries. */
+                        addReplyLongLong(c,nack->delivery_count);
+
+                        arraylen_cpel++;
+                    }
+                    setDeferredArrayLen(c,arrayptr_cpel,arraylen_cpel);
+                    raxStop(&ri_cpel);
+                }
+                raxStop(&ri_consumers);
+            }
+            raxStop(&ri_cgroups);
+        }
+    }
+}
+
+/* XINFO CONSUMERS <key> <group>
+ * XINFO GROUPS <key>
+ * XINFO STREAM <key> [FULL [COUNT <count>]]
+ * XINFO HELP. */
+void xinfoCommand(client *c) {
+    stream *s = NULL;
+    char *opt;
+    robj *key;
+
+    /* HELP is special. Handle it ASAP. */
+    if (!strcasecmp(c->argv[1]->ptr,"HELP")) {
+        const char *help[] = {
+"CONSUMERS <key> <groupname>",
+"    Show consumers of <groupname>.",
+"GROUPS <key>",
+"    Show the stream consumer groups.",
+"STREAM <key> [FULL [COUNT <count>]",
+"    Show information about the stream.",
+NULL
+        };
+        addReplyHelp(c, help);
+        return;
+    }
+
+    /* With the exception of HELP handled before any other sub commands, all
+     * the ones are in the form of "<subcommand> <key>". */
+    opt = c->argv[1]->ptr;
+    key = c->argv[2];
+
+    /* Lookup the key now, this is common for all the subcommands but HELP. */
+    robj *o = lookupKeyReadOrReply(c,key,shared.nokeyerr);
+    if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
+    s = o->ptr;
+
+    /* Dispatch the different subcommands. */
+    if (!strcasecmp(opt,"CONSUMERS") && c->argc == 4) {
+        /* XINFO CONSUMERS <key> <group>. */
+        streamCG *cg = streamLookupCG(s,c->argv[3]->ptr);
+        if (cg == NULL) {
+            addReplyErrorFormat(c, "-NOGROUP No such consumer group '%s' "
+                                   "for key name '%s'",
+                                   (char*)c->argv[3]->ptr, (char*)key->ptr);
+            return;
+        }
+
+        addReplyArrayLen(c,raxSize(cg->consumers));
+        raxIterator ri;
+        raxStart(&ri,cg->consumers);
+        raxSeek(&ri,"^",NULL,0);
+        mstime_t now = commandTimeSnapshot();
+        while(raxNext(&ri)) {
+            streamConsumer *consumer = ri.data;
+            mstime_t inactive = consumer->active_time != -1 ? now - consumer->active_time : consumer->active_time;
+            mstime_t idle = now - consumer->seen_time;
+            if (idle < 0) idle = 0;
+
+            addReplyMapLen(c,4);
+            addReplyBulkCString(c,"name");
+            addReplyBulkCBuffer(c,consumer->name,sdslen(consumer->name));
+            addReplyBulkCString(c,"pending");
+            addReplyLongLong(c,raxSize(consumer->pel));
+            addReplyBulkCString(c,"idle");
+            addReplyLongLong(c,idle);
+            addReplyBulkCString(c,"inactive");
+            addReplyLongLong(c,inactive);
+        }
+        raxStop(&ri);
+    } else if (!strcasecmp(opt,"GROUPS") && c->argc == 3) {
+        /* XINFO GROUPS <key>. */
+        if (s->cgroups == NULL) {
+            addReplyArrayLen(c,0);
+            return;
+        }
+
+        addReplyArrayLen(c,raxSize(s->cgroups));
+        raxIterator ri;
+        raxStart(&ri,s->cgroups);
+        raxSeek(&ri,"^",NULL,0);
+        while(raxNext(&ri)) {
+            streamCG *cg = ri.data;
+            addReplyMapLen(c,6);
+            addReplyBulkCString(c,"name");
+            addReplyBulkCBuffer(c,ri.key,ri.key_len);
+            addReplyBulkCString(c,"consumers");
+            addReplyLongLong(c,raxSize(cg->consumers));
+            addReplyBulkCString(c,"pending");
+            addReplyLongLong(c,raxSize(cg->pel));
+            addReplyBulkCString(c,"last-delivered-id");
+            addReplyStreamID(c,&cg->last_id);
+            addReplyBulkCString(c,"entries-read");
+            if (cg->entries_read != SCG_INVALID_ENTRIES_READ) {
+                addReplyLongLong(c,cg->entries_read);
+            } else {
+                addReplyNull(c);
+            }
+            addReplyBulkCString(c,"lag");
+            streamReplyWithCGLag(c,s,cg);
+        }
+        raxStop(&ri);
+    } else if (!strcasecmp(opt,"STREAM")) {
+        /* XINFO STREAM <key> [FULL [COUNT <count>]]. */
+        xinfoReplyWithStreamInfo(c,s);
+    } else {
+        addReplySubcommandSyntaxError(c);
+    }
+}
+
+/* Validate the integrity stream listpack entries structure. Both in term of a
+ * valid listpack, but also that the structure of the entries matches a valid
+ * stream. return 1 if valid 0 if not valid. */
+int streamValidateListpackIntegrity(unsigned char *lp, size_t size, int deep) {
+    int valid_record;
+    unsigned char *p, *next;
+
+    /* Since we don't want to run validation of all records twice, we'll
+     * run the listpack validation of just the header and do the rest here. */
+    if (!lpValidateIntegrity(lp, size, 0, NULL, NULL))
+        return 0;
+
+    /* In non-deep mode we just validated the listpack header (encoded size) */
+    if (!deep) return 1;
+
+    next = p = lpValidateFirst(lp);
+    if (!lpValidateNext(lp, &next, size)) return 0;
+    if (!p) return 0;
+
+    /* entry count */
+    int64_t entry_count = lpGetIntegerIfValid(p, &valid_record);
+    if (!valid_record) return 0;
+    p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+
+    /* deleted */
+    int64_t deleted_count = lpGetIntegerIfValid(p, &valid_record);
+    if (!valid_record) return 0;
+    p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+
+    /* num-of-fields */
+    int64_t master_fields = lpGetIntegerIfValid(p, &valid_record);
+    if (!valid_record) return 0;
+    p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+
+    /* the field names */
+    for (int64_t j = 0; j < master_fields; j++) {
+        p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+    }
+
+    /* the zero master entry terminator. */
+    int64_t zero = lpGetIntegerIfValid(p, &valid_record);
+    if (!valid_record || zero != 0) return 0;
+    p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+
+    entry_count += deleted_count;
+    while (entry_count--) {
+        if (!p) return 0;
+        int64_t fields = master_fields, extra_fields = 3;
+        int64_t flags = lpGetIntegerIfValid(p, &valid_record);
+        if (!valid_record) return 0;
+        p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+
+        /* entry id */
+        lpGetIntegerIfValid(p, &valid_record);
+        if (!valid_record) return 0;
+        p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+        lpGetIntegerIfValid(p, &valid_record);
+        if (!valid_record) return 0;
+        p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+
+        if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) {
+            /* num-of-fields */
+            fields = lpGetIntegerIfValid(p, &valid_record);
+            if (!valid_record) return 0;
+            p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+
+            /* the field names */
+            for (int64_t j = 0; j < fields; j++) {
+                p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+            }
+
+            extra_fields += fields + 1;
+        }
+
+        /* the values */
+        for (int64_t j = 0; j < fields; j++) {
+            p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+        }
+
+        /* lp-count */
+        int64_t lp_count = lpGetIntegerIfValid(p, &valid_record);
+        if (!valid_record) return 0;
+        if (lp_count != fields + extra_fields) return 0;
+        p = next; if (!lpValidateNext(lp, &next, size)) return 0;
+    }
+
+    if (next)
+        return 0;
+
+    return 1;
+}