async.c 24 KB

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