X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/0537e7bf8042cf9954d3b0abab567edf3b5c0516..b2a7fd0cf7afef7e7ede9e46a317fcb9ae84768c:/src/t_list.c diff --git a/src/t_list.c b/src/t_list.c index db9ca18e..437f4004 100644 --- a/src/t_list.c +++ b/src/t_list.c @@ -260,6 +260,7 @@ void listTypeConvert(robj *subject, int enc) { void pushGenericCommand(redisClient *c, int where) { robj *lobj = lookupKeyWrite(c->db,c->argv[1]); + c->argv[2] = tryObjectEncoding(c->argv[2]); if (lobj == NULL) { if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) { addReply(c,shared.cone); @@ -342,18 +343,21 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) { server.dirty++; } - addReplyUlong(c,listTypeLength(subject)); + addReplyLongLong(c,listTypeLength(subject)); } void lpushxCommand(redisClient *c) { + c->argv[2] = tryObjectEncoding(c->argv[2]); pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD); } void rpushxCommand(redisClient *c) { + c->argv[2] = tryObjectEncoding(c->argv[2]); pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL); } void linsertCommand(redisClient *c) { + c->argv[4] = tryObjectEncoding(c->argv[4]); if (strcasecmp(c->argv[2]->ptr,"after") == 0) { pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL); } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) { @@ -366,7 +370,7 @@ void linsertCommand(redisClient *c) { void llenCommand(redisClient *c) { robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero); if (o == NULL || checkType(c,o,REDIS_LIST)) return; - addReplyUlong(c,listTypeLength(o)); + addReplyLongLong(c,listTypeLength(o)); } void lindexCommand(redisClient *c) { @@ -409,7 +413,7 @@ void lsetCommand(redisClient *c) { robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr); if (o == NULL || checkType(c,o,REDIS_LIST)) return; int index = atoi(c->argv[2]->ptr); - robj *value = c->argv[3]; + robj *value = (c->argv[3] = tryObjectEncoding(c->argv[3])); listTypeTryConversion(o,value); if (o->encoding == REDIS_ENCODING_ZIPLIST) { @@ -559,7 +563,8 @@ void ltrimCommand(redisClient *c) { } void lremCommand(redisClient *c) { - robj *subject, *obj = c->argv[3]; + robj *subject, *obj; + obj = c->argv[3] = tryObjectEncoding(c->argv[3]); int toremove = atoi(c->argv[2]->ptr); int removed = 0; listTypeEntry entry; @@ -594,7 +599,7 @@ void lremCommand(redisClient *c) { decrRefCount(obj); if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]); - addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed)); + addReplyLongLong(c,removed); if (removed) touchWatchedKey(c->db,c->argv[1]); } @@ -613,7 +618,7 @@ void lremCommand(redisClient *c) { * since the element is not just returned but pushed against another list * as well. This command was originally proposed by Ezra Zygmuntowicz. */ -void rpoplpushcommand(redisClient *c) { +void rpoplpushCommand(redisClient *c) { robj *sobj, *value; if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL || checkType(c,sobj,REDIS_LIST)) return; @@ -772,9 +777,28 @@ int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) { redisAssert(ln != NULL); receiver = ln->value; - addReplyMultiBulkLen(receiver,2); - addReplyBulk(receiver,key); - addReplyBulk(receiver,ele); + if (receiver->blocking_target == NULL) { + addReplyMultiBulkLen(receiver,2); + addReplyBulk(receiver,key); + addReplyBulk(receiver,ele); + } + else { + receiver->argc++; + + robj *dobj = lookupKeyWrite(receiver->db,receiver->blocking_target); + if (dobj && checkType(receiver,dobj,REDIS_LIST)) return 0; + + addReplyBulk(receiver,ele); + + /* Create the list if the key does not exist */ + if (!dobj) { + dobj = createZiplistObject(); + dbAdd(receiver->db,receiver->blocking_target,dobj); + } + + listTypePush(dobj,ele,REDIS_HEAD); + } + unblockClientWaitingData(receiver); return 1; } @@ -782,9 +806,20 @@ int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) { /* Blocking RPOP/LPOP */ void blockingPopGenericCommand(redisClient *c, int where) { robj *o; + long long lltimeout; time_t timeout; int j; + /* Make sure timeout is an integer value */ + if (getLongLongFromObjectOrReply(c,c->argv[c->argc-1],&lltimeout, + "timeout is not an integer") != REDIS_OK) return; + + /* Make sure the timeout is not negative */ + if (lltimeout < 0) { + addReplyError(c,"timeout is negative"); + return; + } + for (j = 1; j < c->argc-1; j++) { o = lookupKeyWrite(c->db,c->argv[j]); if (o != NULL) { @@ -798,33 +833,51 @@ void blockingPopGenericCommand(redisClient *c, int where) { robj *argv[2], **orig_argv; int orig_argc; - /* We need to alter the command arguments before to call - * popGenericCommand() as the command takes a single key. */ - orig_argv = c->argv; - orig_argc = c->argc; - argv[1] = c->argv[j]; - c->argv = argv; - c->argc = 2; - - /* Also the return value is different, we need to output - * the multi bulk reply header and the key name. The - * "real" command will add the last element (the value) - * for us. If this souds like an hack to you it's just - * because it is... */ - addReplyMultiBulkLen(c,2); - addReplyBulk(c,argv[1]); - popGenericCommand(c,where); - - /* Fix the client structure with the original stuff */ - c->argv = orig_argv; - c->argc = orig_argc; + if (c->blocking_target == NULL) { + /* We need to alter the command arguments before to call + * popGenericCommand() as the command takes a single key. */ + orig_argv = c->argv; + orig_argc = c->argc; + argv[1] = c->argv[j]; + c->argv = argv; + c->argc = 2; + + /* Also the return value is different, we need to output + * the multi bulk reply header and the key name. The + * "real" command will add the last element (the value) + * for us. If this souds like an hack to you it's just + * because it is... */ + addReplyMultiBulkLen(c,2); + addReplyBulk(c,argv[1]); + + popGenericCommand(c,where); + + /* Fix the client structure with the original stuff */ + c->argv = orig_argv; + c->argc = orig_argc; + } + else { + c->argv[2] = c->blocking_target; + c->blocking_target = NULL; + + rpoplpushCommand(c); + } + return; } } } } + + /* If we are inside a MULTI/EXEC and the list is empty the only thing + * we can do is treating it as a timeout (even with timeout 0). */ + if (c->flags & REDIS_MULTI) { + addReply(c,shared.nullmultibulk); + return; + } + /* If the list is empty or the key does not exists we must block */ - timeout = strtol(c->argv[c->argc-1]->ptr,NULL,10); + timeout = lltimeout; if (timeout > 0) timeout += time(NULL); blockForKeys(c,c->argv+1,c->argc-2,timeout); } @@ -836,3 +889,11 @@ void blpopCommand(redisClient *c) { void brpopCommand(redisClient *c) { blockingPopGenericCommand(c,REDIS_TAIL); } + +void brpoplpushCommand(redisClient *c) { + c->blocking_target = c->argv[2]; + c->argv[2] = c->argv[3]; + c->argc--; + + blockingPopGenericCommand(c,REDIS_TAIL); +}