async.c 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717
  1. /*
  2. * Copyright (c) 2009-2011, Salvatore Sanfilippo <antirez at gmail dot com>
  3. * Copyright (c) 2010-2011, Pieter Noordhuis <pcnoordhuis at gmail dot com>
  4. *
  5. * All rights reserved.
  6. *
  7. * Redistribution and use in source and binary forms, with or without
  8. * modification, are permitted provided that the following conditions are met:
  9. *
  10. * * Redistributions of source code must retain the above copyright notice,
  11. * this list of conditions and the following disclaimer.
  12. * * Redistributions in binary form must reproduce the above copyright
  13. * notice, this list of conditions and the following disclaimer in the
  14. * documentation and/or other materials provided with the distribution.
  15. * * Neither the name of Redis nor the names of its contributors may be used
  16. * to endorse or promote products derived from this software without
  17. * specific prior written permission.
  18. *
  19. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  20. * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  21. * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  22. * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  23. * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  24. * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  25. * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  26. * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  27. * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  28. * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  29. * POSSIBILITY OF SUCH DAMAGE.
  30. */
  31. #include "fmacros.h"
  32. #include "alloc.h"
  33. #include <stdlib.h>
  34. #include <string.h>
  35. #include <strings.h>
  36. #include <assert.h>
  37. #include <ctype.h>
  38. #include <errno.h>
  39. #include "async.h"
  40. #include "net.h"
  41. #include "dict.c"
  42. #include "sds.h"
  43. #define _EL_ADD_READ(ctx) do { \
  44. if ((ctx)->ev.addRead) (ctx)->ev.addRead((ctx)->ev.data); \
  45. } while(0)
  46. #define _EL_DEL_READ(ctx) do { \
  47. if ((ctx)->ev.delRead) (ctx)->ev.delRead((ctx)->ev.data); \
  48. } while(0)
  49. #define _EL_ADD_WRITE(ctx) do { \
  50. if ((ctx)->ev.addWrite) (ctx)->ev.addWrite((ctx)->ev.data); \
  51. } while(0)
  52. #define _EL_DEL_WRITE(ctx) do { \
  53. if ((ctx)->ev.delWrite) (ctx)->ev.delWrite((ctx)->ev.data); \
  54. } while(0)
  55. #define _EL_CLEANUP(ctx) do { \
  56. if ((ctx)->ev.cleanup) (ctx)->ev.cleanup((ctx)->ev.data); \
  57. } while(0);
  58. /* Forward declaration of function in hiredis.c */
  59. int __redisAppendCommand(redisContext *c, const char *cmd, size_t len);
  60. /* Functions managing dictionary of callbacks for pub/sub. */
  61. static unsigned int callbackHash(const void *key) {
  62. return dictGenHashFunction((const unsigned char *)key,
  63. sdslen((const sds)key));
  64. }
  65. static void *callbackValDup(void *privdata, const void *src) {
  66. ((void) privdata);
  67. redisCallback *dup = hi_malloc(sizeof(*dup));
  68. memcpy(dup,src,sizeof(*dup));
  69. return dup;
  70. }
  71. static int callbackKeyCompare(void *privdata, const void *key1, const void *key2) {
  72. int l1, l2;
  73. ((void) privdata);
  74. l1 = sdslen((const sds)key1);
  75. l2 = sdslen((const sds)key2);
  76. if (l1 != l2) return 0;
  77. return memcmp(key1,key2,l1) == 0;
  78. }
  79. static void callbackKeyDestructor(void *privdata, void *key) {
  80. ((void) privdata);
  81. sdsfree((sds)key);
  82. }
  83. static void callbackValDestructor(void *privdata, void *val) {
  84. ((void) privdata);
  85. free(val);
  86. }
  87. static dictType callbackDict = {
  88. callbackHash,
  89. NULL,
  90. callbackValDup,
  91. callbackKeyCompare,
  92. callbackKeyDestructor,
  93. callbackValDestructor
  94. };
  95. static redisAsyncContext *redisAsyncInitialize(redisContext *c) {
  96. redisAsyncContext *ac;
  97. ac = realloc(c,sizeof(redisAsyncContext));
  98. if (ac == NULL)
  99. return NULL;
  100. c = &(ac->c);
  101. /* The regular connect functions will always set the flag REDIS_CONNECTED.
  102. * For the async API, we want to wait until the first write event is
  103. * received up before setting this flag, so reset it here. */
  104. c->flags &= ~REDIS_CONNECTED;
  105. ac->err = 0;
  106. ac->errstr = NULL;
  107. ac->data = NULL;
  108. ac->ev.data = NULL;
  109. ac->ev.addRead = NULL;
  110. ac->ev.delRead = NULL;
  111. ac->ev.addWrite = NULL;
  112. ac->ev.delWrite = NULL;
  113. ac->ev.cleanup = NULL;
  114. ac->onConnect = NULL;
  115. ac->onDisconnect = NULL;
  116. ac->replies.head = NULL;
  117. ac->replies.tail = NULL;
  118. ac->sub.invalid.head = NULL;
  119. ac->sub.invalid.tail = NULL;
  120. ac->sub.channels = dictCreate(&callbackDict,NULL);
  121. ac->sub.patterns = dictCreate(&callbackDict,NULL);
  122. return ac;
  123. }
  124. /* We want the error field to be accessible directly instead of requiring
  125. * an indirection to the redisContext struct. */
  126. static void __redisAsyncCopyError(redisAsyncContext *ac) {
  127. if (!ac)
  128. return;
  129. redisContext *c = &(ac->c);
  130. ac->err = c->err;
  131. ac->errstr = c->errstr;
  132. }
  133. redisAsyncContext *redisAsyncConnect(const char *ip, int port) {
  134. redisContext *c;
  135. redisAsyncContext *ac;
  136. c = redisConnectNonBlock(ip,port);
  137. if (c == NULL)
  138. return NULL;
  139. ac = redisAsyncInitialize(c);
  140. if (ac == NULL) {
  141. redisFree(c);
  142. return NULL;
  143. }
  144. __redisAsyncCopyError(ac);
  145. return ac;
  146. }
  147. redisAsyncContext *redisAsyncConnectBind(const char *ip, int port,
  148. const char *source_addr) {
  149. redisContext *c = redisConnectBindNonBlock(ip,port,source_addr);
  150. redisAsyncContext *ac = redisAsyncInitialize(c);
  151. __redisAsyncCopyError(ac);
  152. return ac;
  153. }
  154. redisAsyncContext *redisAsyncConnectBindWithReuse(const char *ip, int port,
  155. const char *source_addr) {
  156. redisContext *c = redisConnectBindNonBlockWithReuse(ip,port,source_addr);
  157. redisAsyncContext *ac = redisAsyncInitialize(c);
  158. __redisAsyncCopyError(ac);
  159. return ac;
  160. }
  161. redisAsyncContext *redisAsyncConnectUnix(const char *path) {
  162. redisContext *c;
  163. redisAsyncContext *ac;
  164. c = redisConnectUnixNonBlock(path);
  165. if (c == NULL)
  166. return NULL;
  167. ac = redisAsyncInitialize(c);
  168. if (ac == NULL) {
  169. redisFree(c);
  170. return NULL;
  171. }
  172. __redisAsyncCopyError(ac);
  173. return ac;
  174. }
  175. int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn) {
  176. if (ac->onConnect == NULL) {
  177. ac->onConnect = fn;
  178. /* The common way to detect an established connection is to wait for
  179. * the first write event to be fired. This assumes the related event
  180. * library functions are already set. */
  181. _EL_ADD_WRITE(ac);
  182. return REDIS_OK;
  183. }
  184. return REDIS_ERR;
  185. }
  186. int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn) {
  187. if (ac->onDisconnect == NULL) {
  188. ac->onDisconnect = fn;
  189. return REDIS_OK;
  190. }
  191. return REDIS_ERR;
  192. }
  193. /* Helper functions to push/shift callbacks */
  194. static int __redisPushCallback(redisCallbackList *list, redisCallback *source) {
  195. redisCallback *cb;
  196. /* Copy callback from stack to heap */
  197. cb = malloc(sizeof(*cb));
  198. if (cb == NULL)
  199. return REDIS_ERR_OOM;
  200. if (source != NULL) {
  201. memcpy(cb,source,sizeof(*cb));
  202. cb->next = NULL;
  203. }
  204. /* Store callback in list */
  205. if (list->head == NULL)
  206. list->head = cb;
  207. if (list->tail != NULL)
  208. list->tail->next = cb;
  209. list->tail = cb;
  210. return REDIS_OK;
  211. }
  212. static int __redisShiftCallback(redisCallbackList *list, redisCallback *target) {
  213. redisCallback *cb = list->head;
  214. if (cb != NULL) {
  215. list->head = cb->next;
  216. if (cb == list->tail)
  217. list->tail = NULL;
  218. /* Copy callback from heap to stack */
  219. if (target != NULL)
  220. memcpy(target,cb,sizeof(*cb));
  221. free(cb);
  222. return REDIS_OK;
  223. }
  224. return REDIS_ERR;
  225. }
  226. static void __redisRunCallback(redisAsyncContext *ac, redisCallback *cb, redisReply *reply) {
  227. redisContext *c = &(ac->c);
  228. if (cb->fn != NULL) {
  229. c->flags |= REDIS_IN_CALLBACK;
  230. cb->fn(ac,reply,cb->privdata);
  231. c->flags &= ~REDIS_IN_CALLBACK;
  232. }
  233. }
  234. /* Helper function to free the context. */
  235. static void __redisAsyncFree(redisAsyncContext *ac) {
  236. redisContext *c = &(ac->c);
  237. redisCallback cb;
  238. dictIterator *it;
  239. dictEntry *de;
  240. /* Execute pending callbacks with NULL reply. */
  241. while (__redisShiftCallback(&ac->replies,&cb) == REDIS_OK)
  242. __redisRunCallback(ac,&cb,NULL);
  243. /* Execute callbacks for invalid commands */
  244. while (__redisShiftCallback(&ac->sub.invalid,&cb) == REDIS_OK)
  245. __redisRunCallback(ac,&cb,NULL);
  246. /* Run subscription callbacks callbacks with NULL reply */
  247. it = dictGetIterator(ac->sub.channels);
  248. while ((de = dictNext(it)) != NULL)
  249. __redisRunCallback(ac,dictGetEntryVal(de),NULL);
  250. dictReleaseIterator(it);
  251. dictRelease(ac->sub.channels);
  252. it = dictGetIterator(ac->sub.patterns);
  253. while ((de = dictNext(it)) != NULL)
  254. __redisRunCallback(ac,dictGetEntryVal(de),NULL);
  255. dictReleaseIterator(it);
  256. dictRelease(ac->sub.patterns);
  257. /* Signal event lib to clean up */
  258. _EL_CLEANUP(ac);
  259. /* Execute disconnect callback. When redisAsyncFree() initiated destroying
  260. * this context, the status will always be REDIS_OK. */
  261. if (ac->onDisconnect && (c->flags & REDIS_CONNECTED)) {
  262. if (c->flags & REDIS_FREEING) {
  263. ac->onDisconnect(ac,REDIS_OK);
  264. } else {
  265. ac->onDisconnect(ac,(ac->err == 0) ? REDIS_OK : REDIS_ERR);
  266. }
  267. }
  268. /* Cleanup self */
  269. redisFree(c);
  270. }
  271. /* Free the async context. When this function is called from a callback,
  272. * control needs to be returned to redisProcessCallbacks() before actual
  273. * free'ing. To do so, a flag is set on the context which is picked up by
  274. * redisProcessCallbacks(). Otherwise, the context is immediately free'd. */
  275. void redisAsyncFree(redisAsyncContext *ac) {
  276. redisContext *c = &(ac->c);
  277. c->flags |= REDIS_FREEING;
  278. if (!(c->flags & REDIS_IN_CALLBACK))
  279. __redisAsyncFree(ac);
  280. }
  281. /* Helper function to make the disconnect happen and clean up. */
  282. static void __redisAsyncDisconnect(redisAsyncContext *ac) {
  283. redisContext *c = &(ac->c);
  284. /* Make sure error is accessible if there is any */
  285. __redisAsyncCopyError(ac);
  286. if (ac->err == 0) {
  287. /* For clean disconnects, there should be no pending callbacks. */
  288. int ret = __redisShiftCallback(&ac->replies,NULL);
  289. assert(ret == REDIS_ERR);
  290. } else {
  291. /* Disconnection is caused by an error, make sure that pending
  292. * callbacks cannot call new commands. */
  293. c->flags |= REDIS_DISCONNECTING;
  294. }
  295. /* For non-clean disconnects, __redisAsyncFree() will execute pending
  296. * callbacks with a NULL-reply. */
  297. __redisAsyncFree(ac);
  298. }
  299. /* Tries to do a clean disconnect from Redis, meaning it stops new commands
  300. * from being issued, but tries to flush the output buffer and execute
  301. * callbacks for all remaining replies. When this function is called from a
  302. * callback, there might be more replies and we can safely defer disconnecting
  303. * to redisProcessCallbacks(). Otherwise, we can only disconnect immediately
  304. * when there are no pending callbacks. */
  305. void redisAsyncDisconnect(redisAsyncContext *ac) {
  306. redisContext *c = &(ac->c);
  307. c->flags |= REDIS_DISCONNECTING;
  308. if (!(c->flags & REDIS_IN_CALLBACK) && ac->replies.head == NULL)
  309. __redisAsyncDisconnect(ac);
  310. }
  311. static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, redisCallback *dstcb) {
  312. redisContext *c = &(ac->c);
  313. dict *callbacks;
  314. redisCallback *cb;
  315. dictEntry *de;
  316. int pvariant;
  317. char *stype;
  318. sds sname;
  319. /* Custom reply functions are not supported for pub/sub. This will fail
  320. * very hard when they are used... */
  321. if (reply->type == REDIS_REPLY_ARRAY) {
  322. assert(reply->elements >= 2);
  323. assert(reply->element[0]->type == REDIS_REPLY_STRING);
  324. stype = reply->element[0]->str;
  325. pvariant = (tolower(stype[0]) == 'p') ? 1 : 0;
  326. if (pvariant)
  327. callbacks = ac->sub.patterns;
  328. else
  329. callbacks = ac->sub.channels;
  330. /* Locate the right callback */
  331. assert(reply->element[1]->type == REDIS_REPLY_STRING);
  332. sname = sdsnewlen(reply->element[1]->str,reply->element[1]->len);
  333. de = dictFind(callbacks,sname);
  334. if (de != NULL) {
  335. cb = dictGetEntryVal(de);
  336. /* If this is an subscribe reply decrease pending counter. */
  337. if (strcasecmp(stype+pvariant,"subscribe") == 0) {
  338. cb->pending_subs -= 1;
  339. }
  340. memcpy(dstcb,cb,sizeof(*dstcb));
  341. /* If this is an unsubscribe message, remove it. */
  342. if (strcasecmp(stype+pvariant,"unsubscribe") == 0) {
  343. if (cb->pending_subs == 0)
  344. dictDelete(callbacks,sname);
  345. /* If this was the last unsubscribe message, revert to
  346. * non-subscribe mode. */
  347. assert(reply->element[2]->type == REDIS_REPLY_INTEGER);
  348. /* Unset subscribed flag only when no pipelined pending subscribe. */
  349. if (reply->element[2]->integer == 0
  350. && dictSize(ac->sub.channels) == 0
  351. && dictSize(ac->sub.patterns) == 0)
  352. c->flags &= ~REDIS_SUBSCRIBED;
  353. }
  354. }
  355. sdsfree(sname);
  356. } else {
  357. /* Shift callback for invalid commands. */
  358. __redisShiftCallback(&ac->sub.invalid,dstcb);
  359. }
  360. return REDIS_OK;
  361. }
  362. void redisProcessCallbacks(redisAsyncContext *ac) {
  363. redisContext *c = &(ac->c);
  364. redisCallback cb = {NULL, NULL, 0, NULL};
  365. void *reply = NULL;
  366. int status;
  367. while((status = redisGetReply(c,&reply)) == REDIS_OK) {
  368. if (reply == NULL) {
  369. /* When the connection is being disconnected and there are
  370. * no more replies, this is the cue to really disconnect. */
  371. if (c->flags & REDIS_DISCONNECTING && sdslen(c->obuf) == 0
  372. && ac->replies.head == NULL) {
  373. __redisAsyncDisconnect(ac);
  374. return;
  375. }
  376. /* If monitor mode, repush callback */
  377. if(c->flags & REDIS_MONITORING) {
  378. __redisPushCallback(&ac->replies,&cb);
  379. }
  380. /* When the connection is not being disconnected, simply stop
  381. * trying to get replies and wait for the next loop tick. */
  382. break;
  383. }
  384. /* Even if the context is subscribed, pending regular callbacks will
  385. * get a reply before pub/sub messages arrive. */
  386. if (__redisShiftCallback(&ac->replies,&cb) != REDIS_OK) {
  387. /*
  388. * A spontaneous reply in a not-subscribed context can be the error
  389. * reply that is sent when a new connection exceeds the maximum
  390. * number of allowed connections on the server side.
  391. *
  392. * This is seen as an error instead of a regular reply because the
  393. * server closes the connection after sending it.
  394. *
  395. * To prevent the error from being overwritten by an EOF error the
  396. * connection is closed here. See issue #43.
  397. *
  398. * Another possibility is that the server is loading its dataset.
  399. * In this case we also want to close the connection, and have the
  400. * user wait until the server is ready to take our request.
  401. */
  402. if (((redisReply*)reply)->type == REDIS_REPLY_ERROR) {
  403. c->err = REDIS_ERR_OTHER;
  404. snprintf(c->errstr,sizeof(c->errstr),"%s",((redisReply*)reply)->str);
  405. c->reader->fn->freeObject(reply);
  406. __redisAsyncDisconnect(ac);
  407. return;
  408. }
  409. /* No more regular callbacks and no errors, the context *must* be subscribed or monitoring. */
  410. assert((c->flags & REDIS_SUBSCRIBED || c->flags & REDIS_MONITORING));
  411. if(c->flags & REDIS_SUBSCRIBED)
  412. __redisGetSubscribeCallback(ac,reply,&cb);
  413. }
  414. if (cb.fn != NULL) {
  415. __redisRunCallback(ac,&cb,reply);
  416. c->reader->fn->freeObject(reply);
  417. /* Proceed with free'ing when redisAsyncFree() was called. */
  418. if (c->flags & REDIS_FREEING) {
  419. __redisAsyncFree(ac);
  420. return;
  421. }
  422. } else {
  423. /* No callback for this reply. This can either be a NULL callback,
  424. * or there were no callbacks to begin with. Either way, don't
  425. * abort with an error, but simply ignore it because the client
  426. * doesn't know what the server will spit out over the wire. */
  427. c->reader->fn->freeObject(reply);
  428. }
  429. }
  430. /* Disconnect when there was an error reading the reply */
  431. if (status != REDIS_OK)
  432. __redisAsyncDisconnect(ac);
  433. }
  434. /* Internal helper function to detect socket status the first time a read or
  435. * write event fires. When connecting was not successful, the connect callback
  436. * is called with a REDIS_ERR status and the context is free'd. */
  437. static int __redisAsyncHandleConnect(redisAsyncContext *ac) {
  438. redisContext *c = &(ac->c);
  439. if (redisCheckSocketError(c) == REDIS_ERR) {
  440. /* Try again later when connect(2) is still in progress. */
  441. if (errno == EINPROGRESS)
  442. return REDIS_OK;
  443. if (ac->onConnect) ac->onConnect(ac,REDIS_ERR);
  444. __redisAsyncDisconnect(ac);
  445. return REDIS_ERR;
  446. }
  447. /* Mark context as connected. */
  448. c->flags |= REDIS_CONNECTED;
  449. if (ac->onConnect) ac->onConnect(ac,REDIS_OK);
  450. return REDIS_OK;
  451. }
  452. /* This function should be called when the socket is readable.
  453. * It processes all replies that can be read and executes their callbacks.
  454. */
  455. void redisAsyncHandleRead(redisAsyncContext *ac) {
  456. redisContext *c = &(ac->c);
  457. if (!(c->flags & REDIS_CONNECTED)) {
  458. /* Abort connect was not successful. */
  459. if (__redisAsyncHandleConnect(ac) != REDIS_OK)
  460. return;
  461. /* Try again later when the context is still not connected. */
  462. if (!(c->flags & REDIS_CONNECTED))
  463. return;
  464. }
  465. if (redisBufferRead(c) == REDIS_ERR) {
  466. __redisAsyncDisconnect(ac);
  467. } else {
  468. /* Always re-schedule reads */
  469. _EL_ADD_READ(ac);
  470. redisProcessCallbacks(ac);
  471. }
  472. }
  473. void redisAsyncHandleWrite(redisAsyncContext *ac) {
  474. redisContext *c = &(ac->c);
  475. int done = 0;
  476. if (!(c->flags & REDIS_CONNECTED)) {
  477. /* Abort connect was not successful. */
  478. if (__redisAsyncHandleConnect(ac) != REDIS_OK)
  479. return;
  480. /* Try again later when the context is still not connected. */
  481. if (!(c->flags & REDIS_CONNECTED))
  482. return;
  483. }
  484. if (redisBufferWrite(c,&done) == REDIS_ERR) {
  485. __redisAsyncDisconnect(ac);
  486. } else {
  487. /* Continue writing when not done, stop writing otherwise */
  488. if (!done)
  489. _EL_ADD_WRITE(ac);
  490. else
  491. _EL_DEL_WRITE(ac);
  492. /* Always schedule reads after writes */
  493. _EL_ADD_READ(ac);
  494. }
  495. }
  496. /* Sets a pointer to the first argument and its length starting at p. Returns
  497. * the number of bytes to skip to get to the following argument. */
  498. static const char *nextArgument(const char *start, const char **str, size_t *len) {
  499. const char *p = start;
  500. if (p[0] != '$') {
  501. p = strchr(p,'$');
  502. if (p == NULL) return NULL;
  503. }
  504. *len = (int)strtol(p+1,NULL,10);
  505. p = strchr(p,'\r');
  506. assert(p);
  507. *str = p+2;
  508. return p+2+(*len)+2;
  509. }
  510. /* Helper function for the redisAsyncCommand* family of functions. Writes a
  511. * formatted command to the output buffer and registers the provided callback
  512. * function with the context. */
  513. static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *cmd, size_t len) {
  514. redisContext *c = &(ac->c);
  515. redisCallback cb;
  516. struct dict *cbdict;
  517. dictEntry *de;
  518. redisCallback *existcb;
  519. int pvariant, hasnext;
  520. const char *cstr, *astr;
  521. size_t clen, alen;
  522. const char *p;
  523. sds sname;
  524. int ret;
  525. /* Don't accept new commands when the connection is about to be closed. */
  526. if (c->flags & (REDIS_DISCONNECTING | REDIS_FREEING)) return REDIS_ERR;
  527. /* Setup callback */
  528. cb.fn = fn;
  529. cb.privdata = privdata;
  530. cb.pending_subs = 1;
  531. /* Find out which command will be appended. */
  532. p = nextArgument(cmd,&cstr,&clen);
  533. assert(p != NULL);
  534. hasnext = (p[0] == '$');
  535. pvariant = (tolower(cstr[0]) == 'p') ? 1 : 0;
  536. cstr += pvariant;
  537. clen -= pvariant;
  538. if (hasnext && strncasecmp(cstr,"subscribe\r\n",11) == 0) {
  539. c->flags |= REDIS_SUBSCRIBED;
  540. /* Add every channel/pattern to the list of subscription callbacks. */
  541. while ((p = nextArgument(p,&astr,&alen)) != NULL) {
  542. sname = sdsnewlen(astr,alen);
  543. if (pvariant)
  544. cbdict = ac->sub.patterns;
  545. else
  546. cbdict = ac->sub.channels;
  547. de = dictFind(cbdict,sname);
  548. if (de != NULL) {
  549. existcb = dictGetEntryVal(de);
  550. cb.pending_subs = existcb->pending_subs + 1;
  551. }
  552. ret = dictReplace(cbdict,sname,&cb);
  553. if (ret == 0) sdsfree(sname);
  554. }
  555. } else if (strncasecmp(cstr,"unsubscribe\r\n",13) == 0) {
  556. /* It is only useful to call (P)UNSUBSCRIBE when the context is
  557. * subscribed to one or more channels or patterns. */
  558. if (!(c->flags & REDIS_SUBSCRIBED)) return REDIS_ERR;
  559. /* (P)UNSUBSCRIBE does not have its own response: every channel or
  560. * pattern that is unsubscribed will receive a message. This means we
  561. * should not append a callback function for this command. */
  562. } else if(strncasecmp(cstr,"monitor\r\n",9) == 0) {
  563. /* Set monitor flag and push callback */
  564. c->flags |= REDIS_MONITORING;
  565. __redisPushCallback(&ac->replies,&cb);
  566. } else {
  567. if (c->flags & REDIS_SUBSCRIBED)
  568. /* This will likely result in an error reply, but it needs to be
  569. * received and passed to the callback. */
  570. __redisPushCallback(&ac->sub.invalid,&cb);
  571. else
  572. __redisPushCallback(&ac->replies,&cb);
  573. }
  574. __redisAppendCommand(c,cmd,len);
  575. /* Always schedule a write when the write buffer is non-empty */
  576. _EL_ADD_WRITE(ac);
  577. return REDIS_OK;
  578. }
  579. int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap) {
  580. char *cmd;
  581. int len;
  582. int status;
  583. len = redisvFormatCommand(&cmd,format,ap);
  584. /* We don't want to pass -1 or -2 to future functions as a length. */
  585. if (len < 0)
  586. return REDIS_ERR;
  587. status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
  588. free(cmd);
  589. return status;
  590. }
  591. int redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, ...) {
  592. va_list ap;
  593. int status;
  594. va_start(ap,format);
  595. status = redisvAsyncCommand(ac,fn,privdata,format,ap);
  596. va_end(ap);
  597. return status;
  598. }
  599. int redisAsyncCommandArgv(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, int argc, const char **argv, const size_t *argvlen) {
  600. sds cmd;
  601. int len;
  602. int status;
  603. len = redisFormatSdsCommandArgv(&cmd,argc,argv,argvlen);
  604. if (len < 0)
  605. return REDIS_ERR;
  606. status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
  607. sdsfree(cmd);
  608. return status;
  609. }
  610. int redisAsyncFormattedCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *cmd, size_t len) {
  611. int status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
  612. return status;
  613. }