caleb-brown/redict
1
0
Fork 0
mirror of https://codeberg.org/redict/redict.git synced 2025-01-22 16:18:28 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Also use propagate() in streamPropagateGroupID().

Browse Source
This commit is contained in:
antirez 2020-04-24 10:13:20 +02:00
parent 72f0751905
commit 57a0c9c98d
1 changed files with 11 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/t_stream.c
View File

@ -848,6 +848,11 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
argv[11] = createStringObject("JUSTID",6); argv[11] = createStringObject("JUSTID",6);
argv[12] = createStringObject("LASTID",6); argv[12] = createStringObject("LASTID",6);
argv[13] = createObjectFromStreamID(&group->last_id); argv[13] = createObjectFromStreamID(&group->last_id);
/* We use progagate() because this code path is not always called from
* the command execution context. Moreover this will just alter the
* consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */
propagate(server.xclaimCommand,c->db->id,argv,14,PROPAGATE_AOF|PROPAGATE_REPL); propagate(server.xclaimCommand,c->db->id,argv,14,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[0]); decrRefCount(argv[0]);
decrRefCount(argv[3]); decrRefCount(argv[3]);
@ -875,7 +880,12 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
argv[2] = key; argv[2] = key;
argv[3] = groupname; argv[3] = groupname;
argv[4] = createObjectFromStreamID(&group->last_id); argv[4] = createObjectFromStreamID(&group->last_id);
alsoPropagate(server.xgroupCommand,c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL);
/* We use progagate() because this code path is not always called from
* the command execution context. Moreover this will just alter the
* consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */
propagate(server.xgroupCommand,c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[0]); decrRefCount(argv[0]);
decrRefCount(argv[1]); decrRefCount(argv[1]);
decrRefCount(argv[4]); decrRefCount(argv[4]);