123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716 |
- /*
- * Copyright (c) 2009-2011, Salvatore Sanfilippo <antirez at gmail dot com>
- * Copyright (c) 2010-2011, Pieter Noordhuis <pcnoordhuis 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 "fmacros.h"
- #include <stdlib.h>
- #include <string.h>
- #include <strings.h>
- #include <assert.h>
- #include <ctype.h>
- #include <errno.h>
- #include "async.h"
- #include "net.h"
- #include "dict.c"
- #include "sds.h"
- #define _EL_ADD_READ(ctx) do { \
- if ((ctx)->ev.addRead) (ctx)->ev.addRead((ctx)->ev.data); \
- } while(0)
- #define _EL_DEL_READ(ctx) do { \
- if ((ctx)->ev.delRead) (ctx)->ev.delRead((ctx)->ev.data); \
- } while(0)
- #define _EL_ADD_WRITE(ctx) do { \
- if ((ctx)->ev.addWrite) (ctx)->ev.addWrite((ctx)->ev.data); \
- } while(0)
- #define _EL_DEL_WRITE(ctx) do { \
- if ((ctx)->ev.delWrite) (ctx)->ev.delWrite((ctx)->ev.data); \
- } while(0)
- #define _EL_CLEANUP(ctx) do { \
- if ((ctx)->ev.cleanup) (ctx)->ev.cleanup((ctx)->ev.data); \
- } while(0);
- /* Forward declaration of function in hiredis.c */
- int __redisAppendCommand(redisContext *c, const char *cmd, size_t len);
- /* Functions managing dictionary of callbacks for pub/sub. */
- static unsigned int callbackHash(const void *key) {
- return dictGenHashFunction((const unsigned char *)key,
- sdslen((const sds)key));
- }
- static void *callbackValDup(void *privdata, const void *src) {
- ((void) privdata);
- redisCallback *dup = malloc(sizeof(*dup));
- memcpy(dup,src,sizeof(*dup));
- return dup;
- }
- static int callbackKeyCompare(void *privdata, const void *key1, const void *key2) {
- int l1, l2;
- ((void) privdata);
- l1 = sdslen((const sds)key1);
- l2 = sdslen((const sds)key2);
- if (l1 != l2) return 0;
- return memcmp(key1,key2,l1) == 0;
- }
- static void callbackKeyDestructor(void *privdata, void *key) {
- ((void) privdata);
- sdsfree((sds)key);
- }
- static void callbackValDestructor(void *privdata, void *val) {
- ((void) privdata);
- free(val);
- }
- static dictType callbackDict = {
- callbackHash,
- NULL,
- callbackValDup,
- callbackKeyCompare,
- callbackKeyDestructor,
- callbackValDestructor
- };
- static redisAsyncContext *redisAsyncInitialize(redisContext *c) {
- redisAsyncContext *ac;
- ac = realloc(c,sizeof(redisAsyncContext));
- if (ac == NULL)
- return NULL;
- c = &(ac->c);
- /* The regular connect functions will always set the flag REDIS_CONNECTED.
- * For the async API, we want to wait until the first write event is
- * received up before setting this flag, so reset it here. */
- c->flags &= ~REDIS_CONNECTED;
- ac->err = 0;
- ac->errstr = NULL;
- ac->data = NULL;
- ac->ev.data = NULL;
- ac->ev.addRead = NULL;
- ac->ev.delRead = NULL;
- ac->ev.addWrite = NULL;
- ac->ev.delWrite = NULL;
- ac->ev.cleanup = NULL;
- ac->onConnect = NULL;
- ac->onDisconnect = NULL;
- ac->replies.head = NULL;
- ac->replies.tail = NULL;
- ac->sub.invalid.head = NULL;
- ac->sub.invalid.tail = NULL;
- ac->sub.channels = dictCreate(&callbackDict,NULL);
- ac->sub.patterns = dictCreate(&callbackDict,NULL);
- return ac;
- }
- /* We want the error field to be accessible directly instead of requiring
- * an indirection to the redisContext struct. */
- static void __redisAsyncCopyError(redisAsyncContext *ac) {
- if (!ac)
- return;
- redisContext *c = &(ac->c);
- ac->err = c->err;
- ac->errstr = c->errstr;
- }
- redisAsyncContext *redisAsyncConnect(const char *ip, int port) {
- redisContext *c;
- redisAsyncContext *ac;
- c = redisConnectNonBlock(ip,port);
- if (c == NULL)
- return NULL;
- ac = redisAsyncInitialize(c);
- if (ac == NULL) {
- redisFree(c);
- return NULL;
- }
- __redisAsyncCopyError(ac);
- return ac;
- }
- redisAsyncContext *redisAsyncConnectBind(const char *ip, int port,
- const char *source_addr) {
- redisContext *c = redisConnectBindNonBlock(ip,port,source_addr);
- redisAsyncContext *ac = redisAsyncInitialize(c);
- __redisAsyncCopyError(ac);
- return ac;
- }
- redisAsyncContext *redisAsyncConnectBindWithReuse(const char *ip, int port,
- const char *source_addr) {
- redisContext *c = redisConnectBindNonBlockWithReuse(ip,port,source_addr);
- redisAsyncContext *ac = redisAsyncInitialize(c);
- __redisAsyncCopyError(ac);
- return ac;
- }
- redisAsyncContext *redisAsyncConnectUnix(const char *path) {
- redisContext *c;
- redisAsyncContext *ac;
- c = redisConnectUnixNonBlock(path);
- if (c == NULL)
- return NULL;
- ac = redisAsyncInitialize(c);
- if (ac == NULL) {
- redisFree(c);
- return NULL;
- }
- __redisAsyncCopyError(ac);
- return ac;
- }
- int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn) {
- if (ac->onConnect == NULL) {
- ac->onConnect = fn;
- /* The common way to detect an established connection is to wait for
- * the first write event to be fired. This assumes the related event
- * library functions are already set. */
- _EL_ADD_WRITE(ac);
- return REDIS_OK;
- }
- return REDIS_ERR;
- }
- int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn) {
- if (ac->onDisconnect == NULL) {
- ac->onDisconnect = fn;
- return REDIS_OK;
- }
- return REDIS_ERR;
- }
- /* Helper functions to push/shift callbacks */
- static int __redisPushCallback(redisCallbackList *list, redisCallback *source) {
- redisCallback *cb;
- /* Copy callback from stack to heap */
- cb = malloc(sizeof(*cb));
- if (cb == NULL)
- return REDIS_ERR_OOM;
- if (source != NULL) {
- memcpy(cb,source,sizeof(*cb));
- cb->next = NULL;
- }
- /* Store callback in list */
- if (list->head == NULL)
- list->head = cb;
- if (list->tail != NULL)
- list->tail->next = cb;
- list->tail = cb;
- return REDIS_OK;
- }
- static int __redisShiftCallback(redisCallbackList *list, redisCallback *target) {
- redisCallback *cb = list->head;
- if (cb != NULL) {
- list->head = cb->next;
- if (cb == list->tail)
- list->tail = NULL;
- /* Copy callback from heap to stack */
- if (target != NULL)
- memcpy(target,cb,sizeof(*cb));
- free(cb);
- return REDIS_OK;
- }
- return REDIS_ERR;
- }
- static void __redisRunCallback(redisAsyncContext *ac, redisCallback *cb, redisReply *reply) {
- redisContext *c = &(ac->c);
- if (cb->fn != NULL) {
- c->flags |= REDIS_IN_CALLBACK;
- cb->fn(ac,reply,cb->privdata);
- c->flags &= ~REDIS_IN_CALLBACK;
- }
- }
- /* Helper function to free the context. */
- static void __redisAsyncFree(redisAsyncContext *ac) {
- redisContext *c = &(ac->c);
- redisCallback cb;
- dictIterator *it;
- dictEntry *de;
- /* Execute pending callbacks with NULL reply. */
- while (__redisShiftCallback(&ac->replies,&cb) == REDIS_OK)
- __redisRunCallback(ac,&cb,NULL);
- /* Execute callbacks for invalid commands */
- while (__redisShiftCallback(&ac->sub.invalid,&cb) == REDIS_OK)
- __redisRunCallback(ac,&cb,NULL);
- /* Run subscription callbacks callbacks with NULL reply */
- it = dictGetIterator(ac->sub.channels);
- while ((de = dictNext(it)) != NULL)
- __redisRunCallback(ac,dictGetEntryVal(de),NULL);
- dictReleaseIterator(it);
- dictRelease(ac->sub.channels);
- it = dictGetIterator(ac->sub.patterns);
- while ((de = dictNext(it)) != NULL)
- __redisRunCallback(ac,dictGetEntryVal(de),NULL);
- dictReleaseIterator(it);
- dictRelease(ac->sub.patterns);
- /* Signal event lib to clean up */
- _EL_CLEANUP(ac);
- /* Execute disconnect callback. When redisAsyncFree() initiated destroying
- * this context, the status will always be REDIS_OK. */
- if (ac->onDisconnect && (c->flags & REDIS_CONNECTED)) {
- if (c->flags & REDIS_FREEING) {
- ac->onDisconnect(ac,REDIS_OK);
- } else {
- ac->onDisconnect(ac,(ac->err == 0) ? REDIS_OK : REDIS_ERR);
- }
- }
- /* Cleanup self */
- redisFree(c);
- }
- /* Free the async context. When this function is called from a callback,
- * control needs to be returned to redisProcessCallbacks() before actual
- * free'ing. To do so, a flag is set on the context which is picked up by
- * redisProcessCallbacks(). Otherwise, the context is immediately free'd. */
- void redisAsyncFree(redisAsyncContext *ac) {
- redisContext *c = &(ac->c);
- c->flags |= REDIS_FREEING;
- if (!(c->flags & REDIS_IN_CALLBACK))
- __redisAsyncFree(ac);
- }
- /* Helper function to make the disconnect happen and clean up. */
- static void __redisAsyncDisconnect(redisAsyncContext *ac) {
- redisContext *c = &(ac->c);
- /* Make sure error is accessible if there is any */
- __redisAsyncCopyError(ac);
- if (ac->err == 0) {
- /* For clean disconnects, there should be no pending callbacks. */
- int ret = __redisShiftCallback(&ac->replies,NULL);
- assert(ret == REDIS_ERR);
- } else {
- /* Disconnection is caused by an error, make sure that pending
- * callbacks cannot call new commands. */
- c->flags |= REDIS_DISCONNECTING;
- }
- /* For non-clean disconnects, __redisAsyncFree() will execute pending
- * callbacks with a NULL-reply. */
- __redisAsyncFree(ac);
- }
- /* Tries to do a clean disconnect from Redis, meaning it stops new commands
- * from being issued, but tries to flush the output buffer and execute
- * callbacks for all remaining replies. When this function is called from a
- * callback, there might be more replies and we can safely defer disconnecting
- * to redisProcessCallbacks(). Otherwise, we can only disconnect immediately
- * when there are no pending callbacks. */
- void redisAsyncDisconnect(redisAsyncContext *ac) {
- redisContext *c = &(ac->c);
- c->flags |= REDIS_DISCONNECTING;
- if (!(c->flags & REDIS_IN_CALLBACK) && ac->replies.head == NULL)
- __redisAsyncDisconnect(ac);
- }
- static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, redisCallback *dstcb) {
- redisContext *c = &(ac->c);
- dict *callbacks;
- redisCallback *cb;
- dictEntry *de;
- int pvariant;
- char *stype;
- sds sname;
- /* Custom reply functions are not supported for pub/sub. This will fail
- * very hard when they are used... */
- if (reply->type == REDIS_REPLY_ARRAY) {
- assert(reply->elements >= 2);
- assert(reply->element[0]->type == REDIS_REPLY_STRING);
- stype = reply->element[0]->str;
- pvariant = (tolower(stype[0]) == 'p') ? 1 : 0;
- if (pvariant)
- callbacks = ac->sub.patterns;
- else
- callbacks = ac->sub.channels;
- /* Locate the right callback */
- assert(reply->element[1]->type == REDIS_REPLY_STRING);
- sname = sdsnewlen(reply->element[1]->str,reply->element[1]->len);
- de = dictFind(callbacks,sname);
- if (de != NULL) {
- cb = dictGetEntryVal(de);
- /* If this is an subscribe reply decrease pending counter. */
- if (strcasecmp(stype+pvariant,"subscribe") == 0) {
- cb->pending_subs -= 1;
- }
- memcpy(dstcb,cb,sizeof(*dstcb));
- /* If this is an unsubscribe message, remove it. */
- if (strcasecmp(stype+pvariant,"unsubscribe") == 0) {
- if (cb->pending_subs == 0)
- dictDelete(callbacks,sname);
- /* If this was the last unsubscribe message, revert to
- * non-subscribe mode. */
- assert(reply->element[2]->type == REDIS_REPLY_INTEGER);
- /* Unset subscribed flag only when no pipelined pending subscribe. */
- if (reply->element[2]->integer == 0
- && dictSize(ac->sub.channels) == 0
- && dictSize(ac->sub.patterns) == 0)
- c->flags &= ~REDIS_SUBSCRIBED;
- }
- }
- sdsfree(sname);
- } else {
- /* Shift callback for invalid commands. */
- __redisShiftCallback(&ac->sub.invalid,dstcb);
- }
- return REDIS_OK;
- }
- void redisProcessCallbacks(redisAsyncContext *ac) {
- redisContext *c = &(ac->c);
- redisCallback cb = {NULL, NULL, 0, NULL};
- void *reply = NULL;
- int status;
- while((status = redisGetReply(c,&reply)) == REDIS_OK) {
- if (reply == NULL) {
- /* When the connection is being disconnected and there are
- * no more replies, this is the cue to really disconnect. */
- if (c->flags & REDIS_DISCONNECTING && sdslen(c->obuf) == 0
- && ac->replies.head == NULL) {
- __redisAsyncDisconnect(ac);
- return;
- }
- /* If monitor mode, repush callback */
- if(c->flags & REDIS_MONITORING) {
- __redisPushCallback(&ac->replies,&cb);
- }
- /* When the connection is not being disconnected, simply stop
- * trying to get replies and wait for the next loop tick. */
- break;
- }
- /* Even if the context is subscribed, pending regular callbacks will
- * get a reply before pub/sub messages arrive. */
- if (__redisShiftCallback(&ac->replies,&cb) != REDIS_OK) {
- /*
- * A spontaneous reply in a not-subscribed context can be the error
- * reply that is sent when a new connection exceeds the maximum
- * number of allowed connections on the server side.
- *
- * This is seen as an error instead of a regular reply because the
- * server closes the connection after sending it.
- *
- * To prevent the error from being overwritten by an EOF error the
- * connection is closed here. See issue #43.
- *
- * Another possibility is that the server is loading its dataset.
- * In this case we also want to close the connection, and have the
- * user wait until the server is ready to take our request.
- */
- if (((redisReply*)reply)->type == REDIS_REPLY_ERROR) {
- c->err = REDIS_ERR_OTHER;
- snprintf(c->errstr,sizeof(c->errstr),"%s",((redisReply*)reply)->str);
- c->reader->fn->freeObject(reply);
- __redisAsyncDisconnect(ac);
- return;
- }
- /* No more regular callbacks and no errors, the context *must* be subscribed or monitoring. */
- assert((c->flags & REDIS_SUBSCRIBED || c->flags & REDIS_MONITORING));
- if(c->flags & REDIS_SUBSCRIBED)
- __redisGetSubscribeCallback(ac,reply,&cb);
- }
- if (cb.fn != NULL) {
- __redisRunCallback(ac,&cb,reply);
- c->reader->fn->freeObject(reply);
- /* Proceed with free'ing when redisAsyncFree() was called. */
- if (c->flags & REDIS_FREEING) {
- __redisAsyncFree(ac);
- return;
- }
- } else {
- /* No callback for this reply. This can either be a NULL callback,
- * or there were no callbacks to begin with. Either way, don't
- * abort with an error, but simply ignore it because the client
- * doesn't know what the server will spit out over the wire. */
- c->reader->fn->freeObject(reply);
- }
- }
- /* Disconnect when there was an error reading the reply */
- if (status != REDIS_OK)
- __redisAsyncDisconnect(ac);
- }
- /* Internal helper function to detect socket status the first time a read or
- * write event fires. When connecting was not successful, the connect callback
- * is called with a REDIS_ERR status and the context is free'd. */
- static int __redisAsyncHandleConnect(redisAsyncContext *ac) {
- int completed = 0;
- redisContext *c = &(ac->c);
- if (redisCheckConnectDone(c, &completed) == REDIS_ERR) {
- /* Error! */
- redisCheckSocketError(c);
- if (ac->onConnect) ac->onConnect(ac, REDIS_ERR);
- __redisAsyncDisconnect(ac);
- return REDIS_ERR;
- } else if (completed == 1) {
- /* connected! */
- if (ac->onConnect) ac->onConnect(ac, REDIS_OK);
- c->flags |= REDIS_CONNECTED;
- return REDIS_OK;
- } else {
- return REDIS_OK;
- }
- }
- /* This function should be called when the socket is readable.
- * It processes all replies that can be read and executes their callbacks.
- */
- void redisAsyncHandleRead(redisAsyncContext *ac) {
- redisContext *c = &(ac->c);
- if (!(c->flags & REDIS_CONNECTED)) {
- /* Abort connect was not successful. */
- if (__redisAsyncHandleConnect(ac) != REDIS_OK)
- return;
- /* Try again later when the context is still not connected. */
- if (!(c->flags & REDIS_CONNECTED))
- return;
- }
- if (redisBufferRead(c) == REDIS_ERR) {
- __redisAsyncDisconnect(ac);
- } else {
- /* Always re-schedule reads */
- _EL_ADD_READ(ac);
- redisProcessCallbacks(ac);
- }
- }
- void redisAsyncHandleWrite(redisAsyncContext *ac) {
- redisContext *c = &(ac->c);
- int done = 0;
- if (!(c->flags & REDIS_CONNECTED)) {
- /* Abort connect was not successful. */
- if (__redisAsyncHandleConnect(ac) != REDIS_OK)
- return;
- /* Try again later when the context is still not connected. */
- if (!(c->flags & REDIS_CONNECTED))
- return;
- }
- if (redisBufferWrite(c,&done) == REDIS_ERR) {
- __redisAsyncDisconnect(ac);
- } else {
- /* Continue writing when not done, stop writing otherwise */
- if (!done)
- _EL_ADD_WRITE(ac);
- else
- _EL_DEL_WRITE(ac);
- /* Always schedule reads after writes */
- _EL_ADD_READ(ac);
- }
- }
- /* Sets a pointer to the first argument and its length starting at p. Returns
- * the number of bytes to skip to get to the following argument. */
- static const char *nextArgument(const char *start, const char **str, size_t *len) {
- const char *p = start;
- if (p[0] != '$') {
- p = strchr(p,'$');
- if (p == NULL) return NULL;
- }
- *len = (int)strtol(p+1,NULL,10);
- p = strchr(p,'\r');
- assert(p);
- *str = p+2;
- return p+2+(*len)+2;
- }
- /* Helper function for the redisAsyncCommand* family of functions. Writes a
- * formatted command to the output buffer and registers the provided callback
- * function with the context. */
- static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *cmd, size_t len) {
- redisContext *c = &(ac->c);
- redisCallback cb;
- struct dict *cbdict;
- dictEntry *de;
- redisCallback *existcb;
- int pvariant, hasnext;
- const char *cstr, *astr;
- size_t clen, alen;
- const char *p;
- sds sname;
- int ret;
- /* Don't accept new commands when the connection is about to be closed. */
- if (c->flags & (REDIS_DISCONNECTING | REDIS_FREEING)) return REDIS_ERR;
- /* Setup callback */
- cb.fn = fn;
- cb.privdata = privdata;
- cb.pending_subs = 1;
- /* Find out which command will be appended. */
- p = nextArgument(cmd,&cstr,&clen);
- assert(p != NULL);
- hasnext = (p[0] == '$');
- pvariant = (tolower(cstr[0]) == 'p') ? 1 : 0;
- cstr += pvariant;
- clen -= pvariant;
- if (hasnext && strncasecmp(cstr,"subscribe\r\n",11) == 0) {
- c->flags |= REDIS_SUBSCRIBED;
- /* Add every channel/pattern to the list of subscription callbacks. */
- while ((p = nextArgument(p,&astr,&alen)) != NULL) {
- sname = sdsnewlen(astr,alen);
- if (pvariant)
- cbdict = ac->sub.patterns;
- else
- cbdict = ac->sub.channels;
- de = dictFind(cbdict,sname);
- if (de != NULL) {
- existcb = dictGetEntryVal(de);
- cb.pending_subs = existcb->pending_subs + 1;
- }
- ret = dictReplace(cbdict,sname,&cb);
- if (ret == 0) sdsfree(sname);
- }
- } else if (strncasecmp(cstr,"unsubscribe\r\n",13) == 0) {
- /* It is only useful to call (P)UNSUBSCRIBE when the context is
- * subscribed to one or more channels or patterns. */
- if (!(c->flags & REDIS_SUBSCRIBED)) return REDIS_ERR;
- /* (P)UNSUBSCRIBE does not have its own response: every channel or
- * pattern that is unsubscribed will receive a message. This means we
- * should not append a callback function for this command. */
- } else if(strncasecmp(cstr,"monitor\r\n",9) == 0) {
- /* Set monitor flag and push callback */
- c->flags |= REDIS_MONITORING;
- __redisPushCallback(&ac->replies,&cb);
- } else {
- if (c->flags & REDIS_SUBSCRIBED)
- /* This will likely result in an error reply, but it needs to be
- * received and passed to the callback. */
- __redisPushCallback(&ac->sub.invalid,&cb);
- else
- __redisPushCallback(&ac->replies,&cb);
- }
- __redisAppendCommand(c,cmd,len);
- /* Always schedule a write when the write buffer is non-empty */
- _EL_ADD_WRITE(ac);
- return REDIS_OK;
- }
- int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap) {
- char *cmd;
- int len;
- int status;
- len = redisvFormatCommand(&cmd,format,ap);
- /* We don't want to pass -1 or -2 to future functions as a length. */
- if (len < 0)
- return REDIS_ERR;
- status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
- free(cmd);
- return status;
- }
- int redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, ...) {
- va_list ap;
- int status;
- va_start(ap,format);
- status = redisvAsyncCommand(ac,fn,privdata,format,ap);
- va_end(ap);
- return status;
- }
- int redisAsyncCommandArgv(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, int argc, const char **argv, const size_t *argvlen) {
- sds cmd;
- int len;
- int status;
- len = redisFormatSdsCommandArgv(&cmd,argc,argv,argvlen);
- if (len < 0)
- return REDIS_ERR;
- status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
- sdsfree(cmd);
- return status;
- }
- int redisAsyncFormattedCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *cmd, size_t len) {
- int status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
- return status;
- }
|