Commit 6491717c authored by dejun.xdj's avatar dejun.xdj
Browse files

Streams: rearrange the usage of '-' and '+' IDs in stream commands.

parent d6f5ec6f
...@@ -1137,7 +1137,7 @@ void xaddCommand(client *c) { ...@@ -1137,7 +1137,7 @@ void xaddCommand(client *c) {
maxlen_arg_idx = i; maxlen_arg_idx = i;
} else { } else {
/* If we are here is a syntax error or a valid ID. */ /* If we are here is a syntax error or a valid ID. */
if (streamParseIDOrReply(c,c->argv[i],&id,0) != C_OK) return; if (streamParseIDOrReply(c,c->argv[i],&id,0,0) != C_OK) return;
id_given = 1; id_given = 1;
break; break;
} }
...@@ -1206,8 +1206,8 @@ void xrangeGenericCommand(client *c, int rev) { ...@@ -1206,8 +1206,8 @@ void xrangeGenericCommand(client *c, int rev) {
robj *startarg = rev ? c->argv[3] : c->argv[2]; robj *startarg = rev ? c->argv[3] : c->argv[2];
robj *endarg = rev ? c->argv[2] : c->argv[3]; robj *endarg = rev ? c->argv[2] : c->argv[3];
if (streamParseIDOrReply(c,startarg,&startid,0) == C_ERR) return; if (streamParseIDOrReply(c,startarg,&startid,0,1) == C_ERR) return;
if (streamParseIDOrReply(c,endarg,&endid,UINT64_MAX) == C_ERR) return; if (streamParseIDOrReply(c,endarg,&endid,UINT64_MAX,1) == C_ERR) return;
/* Parse the COUNT option if any. */ /* Parse the COUNT option if any. */
if (c->argc > 4) { if (c->argc > 4) {
...@@ -1393,7 +1393,7 @@ void xreadCommand(client *c) { ...@@ -1393,7 +1393,7 @@ void xreadCommand(client *c) {
ids[id_idx].seq = UINT64_MAX; ids[id_idx].seq = UINT64_MAX;
continue; continue;
} }
if (streamParseIDOrReply(c,c->argv[i],ids+id_idx,0) != C_OK) if (streamParseIDOrReply(c,c->argv[i],ids+id_idx,0,0) != C_OK)
goto cleanup; goto cleanup;
} }
...@@ -1667,7 +1667,7 @@ NULL ...@@ -1667,7 +1667,7 @@ NULL
streamID id; streamID id;
if (!strcmp(c->argv[4]->ptr,"$")) { if (!strcmp(c->argv[4]->ptr,"$")) {
id = s->last_id; id = s->last_id;
} else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) { } else if (streamParseIDOrReply(c,c->argv[4],&id,0,0) != C_OK) {
return; return;
} }
streamCG *cg = streamCreateCG(s,grpname,sdslen(grpname),&id); streamCG *cg = streamCreateCG(s,grpname,sdslen(grpname),&id);
...@@ -1684,7 +1684,7 @@ NULL ...@@ -1684,7 +1684,7 @@ NULL
streamID id; streamID id;
if (!strcmp(c->argv[4]->ptr,"$")) { if (!strcmp(c->argv[4]->ptr,"$")) {
id = s->last_id; id = s->last_id;
} else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) { } else if (streamParseIDOrReply(c,c->argv[4],&id,0,1) != C_OK) {
return; return;
} }
cg->last_id = id; cg->last_id = id;
...@@ -1744,7 +1744,7 @@ void xackCommand(client *c) { ...@@ -1744,7 +1744,7 @@ void xackCommand(client *c) {
for (int j = 3; j < c->argc; j++) { for (int j = 3; j < c->argc; j++) {
streamID id; streamID id;
unsigned char buf[sizeof(streamID)]; unsigned char buf[sizeof(streamID)];
if (streamParseIDOrReply(c,c->argv[j],&id,0) != C_OK) return; if (streamParseIDOrReply(c,c->argv[j],&id,0,0) != C_OK) return;
streamEncodeID(buf,&id); streamEncodeID(buf,&id);
/* Lookup the ID in the group PEL: it will have a reference to the /* Lookup the ID in the group PEL: it will have a reference to the
...@@ -1791,9 +1791,9 @@ void xpendingCommand(client *c) { ...@@ -1791,9 +1791,9 @@ void xpendingCommand(client *c) {
if (c->argc >= 6) { if (c->argc >= 6) {
if (getLongLongFromObjectOrReply(c,c->argv[5],&count,NULL) == C_ERR) if (getLongLongFromObjectOrReply(c,c->argv[5],&count,NULL) == C_ERR)
return; return;
if (streamParseIDOrReply(c,c->argv[3],&startid,0) == C_ERR) if (streamParseIDOrReply(c,c->argv[3],&startid,0,1) == C_ERR)
return; return;
if (streamParseIDOrReply(c,c->argv[4],&endid,UINT64_MAX) == C_ERR) if (streamParseIDOrReply(c,c->argv[4],&endid,UINT64_MAX,1) == C_ERR)
return; return;
} }
...@@ -2002,7 +2002,7 @@ void xclaimCommand(client *c) { ...@@ -2002,7 +2002,7 @@ void xclaimCommand(client *c) {
int j; int j;
for (j = 4; j < c->argc; j++) { for (j = 4; j < c->argc; j++) {
streamID id; streamID id;
if (streamParseIDOrReply(NULL,c->argv[j],&id,0) != C_OK) break; if (streamParseIDOrReply(NULL,c->argv[j],&id,0,0) != C_OK) break;
} }
int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */ int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */
...@@ -2061,7 +2061,7 @@ void xclaimCommand(client *c) { ...@@ -2061,7 +2061,7 @@ void xclaimCommand(client *c) {
for (int j = 5; j <= last_id_arg; j++) { for (int j = 5; j <= last_id_arg; j++) {
streamID id; streamID id;
unsigned char buf[sizeof(streamID)]; unsigned char buf[sizeof(streamID)];
if (streamParseIDOrReply(c,c->argv[j],&id,0) != C_OK) return; if (streamParseIDOrReply(c,c->argv[j],&id,0,0) != C_OK) return;
streamEncodeID(buf,&id); streamEncodeID(buf,&id);
/* Lookup the ID in the group PEL. */ /* Lookup the ID in the group PEL. */
...@@ -2144,13 +2144,13 @@ void xdelCommand(client *c) { ...@@ -2144,13 +2144,13 @@ void xdelCommand(client *c) {
* executed because at some point an invalid ID is parsed. */ * executed because at some point an invalid ID is parsed. */
streamID id; streamID id;
for (int j = 2; j < c->argc; j++) { for (int j = 2; j < c->argc; j++) {
if (streamParseIDOrReply(c,c->argv[j],&id,0) != C_OK) return; if (streamParseIDOrReply(c,c->argv[j],&id,0,0) != C_OK) return;
} }
/* Actually apply the command. */ /* Actually apply the command. */
int deleted = 0; int deleted = 0;
for (int j = 2; j < c->argc; j++) { for (int j = 2; j < c->argc; j++) {
streamParseIDOrReply(c,c->argv[j],&id,0); /* Retval already checked. */ streamParseIDOrReply(c,c->argv[j],&id,0,0); /* Retval already checked. */
deleted += streamDeleteItem(s,&id); deleted += streamDeleteItem(s,&id);
} }
......
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