"vscode:/vscode.git/clone" did not exist on "294bcfc4d207d025f7dc2b5716b787f03f390332"
Commit f0a26144 authored by Valentino Geron's avatar Valentino Geron Committed by antirez
Browse files

XREADGROUP with NOACK should propagate only one XGROUP SETID command

parent fbdef6a9
...@@ -935,6 +935,8 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end ...@@ -935,6 +935,8 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
streamIterator si; streamIterator si;
int64_t numfields; int64_t numfields;
streamID id; 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 /* If the client is asking for some history, we serve it using a
* different function, so that we return entries *solely* from its * different function, so that we return entries *solely* from its
...@@ -950,11 +952,13 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end ...@@ -950,11 +952,13 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
arraylen_ptr = addReplyDeferredLen(c); arraylen_ptr = addReplyDeferredLen(c);
streamIteratorStart(&si,s,start,end,rev); streamIteratorStart(&si,s,start,end,rev);
while(streamIteratorGetID(&si,&id,&numfields)) { while(streamIteratorGetID(&si,&id,&numfields)) {
int propagate_last_id = 0;
/* Update the group last_id if needed. */ /* Update the group last_id if needed. */
if (group && streamCompareID(&id,&group->last_id) > 0) { if (group && streamCompareID(&id,&group->last_id) > 0) {
group->last_id = id; group->last_id = id;
/* Group last id should be propagated only if NOACK was
* specified, otherwise the last id would be included
* in XCLAIM. */
if (noack)
propagate_last_id = 1; propagate_last_id = 1;
} }
...@@ -983,7 +987,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end ...@@ -983,7 +987,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
* XGROUP SETID command. So if we find that there is already * XGROUP SETID command. So if we find that there is already
* a NACK for the entry, we need to associate it to the new * a NACK for the entry, we need to associate it to the new
* consumer. */ * consumer. */
if (group && !(flags & STREAM_RWR_NOACK)) { if (group && !noack) {
unsigned char buf[sizeof(streamID)]; unsigned char buf[sizeof(streamID)];
streamEncodeID(buf,&id); streamEncodeID(buf,&id);
...@@ -1020,14 +1024,16 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end ...@@ -1020,14 +1024,16 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
streamPropagateXCLAIM(c,spi->keyname,group,spi->groupname,idarg,nack); streamPropagateXCLAIM(c,spi->keyname,group,spi->groupname,idarg,nack);
decrRefCount(idarg); decrRefCount(idarg);
} }
} else {
if (propagate_last_id)
streamPropagateGroupID(c,spi->keyname,group,spi->groupname);
} }
arraylen++; arraylen++;
if (count && count == arraylen) break; if (count && count == arraylen) break;
} }
if (spi && propagate_last_id) {
streamPropagateGroupID(c,spi->keyname,group,spi->groupname);
}
streamIteratorStop(&si); streamIteratorStop(&si);
if (arraylen_ptr) setDeferredArrayLen(c,arraylen_ptr,arraylen); if (arraylen_ptr) setDeferredArrayLen(c,arraylen_ptr,arraylen);
return arraylen; return arraylen;
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment