1 files changed, 26 insertions, 1 deletions

diff --git a/async.c b/async.c
index 1dcb04e..a425c42 100644
--- a/async.c
+++ b/async.c
@@ -38,16 +38,27 @@ void __redisAppendCommand(redisContext *c, char *cmd, size_t len);
 
 static redisAsyncContext *redisAsyncInitialize(redisContext *c) {
     redisAsyncContext *ac = realloc(c,sizeof(redisAsyncContext));
+    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->_adapter_data = NULL;
+
     ac->evAddRead = NULL;
     ac->evDelRead = NULL;
     ac->evAddWrite = NULL;
     ac->evDelWrite = NULL;
     ac->evCleanup = NULL;
+
+    ac->onConnect = NULL;
     ac->onDisconnect = NULL;
+
     ac->replies.head = NULL;
     ac->replies.tail = NULL;
     return ac;
@@ -80,6 +91,14 @@ int redisAsyncSetReplyObjectFunctions(redisAsyncContext *ac, redisReplyObjectFun
     return redisSetReplyObjectFunctions(c,fn);
 }
 
+int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn) {
+    if (ac->onConnect == NULL) {
+        ac->onConnect = fn;
+        return REDIS_OK;
+    }
+    return REDIS_ERR;
+}
+
 int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn) {
     if (ac->onDisconnect == NULL) {
         ac->onDisconnect = fn;
@@ -235,8 +254,14 @@ void redisAsyncHandleWrite(redisAsyncContext *ac) {
             if (ac->evDelWrite) ac->evDelWrite(ac->_adapter_data);
         }
 
-        /* Always schedule reads when something was written */
+        /* Always schedule reads after writes */
         if (ac->evAddRead) ac->evAddRead(ac->_adapter_data);
+
+        /* Fire onConnect when this is the first write event. */
+        if (!(c->flags & REDIS_CONNECTED)) {
+            c->flags |= REDIS_CONNECTED;
+            if (ac->onConnect) ac->onConnect(ac);
+        }
     }
 }