minor bug fixes

Author: kice

ASyncRedis.cpp

@@ -52,9 +52,9 @@ ASyncRedis::ASyncRedis(int _intval, int cache)
     cache_size = cache;
 
     loop = aeCreateEventLoop(1024 * 10);
-    evMain = std::thread([this] {
-        aeCreateTimeEvent(loop, interval, timeEventCallback, this, nullptr);
+    aeCreateTimeEvent(loop, interval, timeEventCallback, this, nullptr);
 
+    evMain = std::thread([this] {
         aeMain(loop);
         aeDeleteEventLoop(loop);
         loop = nullptr;
@@ -65,20 +65,17 @@ ASyncRedis::ASyncRedis(int _intval, int cache)
 
 ASyncRedis::~ASyncRedis()
 {
-    redisAsyncDisconnect(ac);
-
     if (loop) {
         aeStop(loop);
-        if (evMain.joinable()) {
-            evMain.join();
-        }
     }
-    loop = nullptr;
 
-    if (ac) {
+    if (evMain.joinable()) {
+        evMain.join();
+    }
+
+    if (!disconnected && ac) {
         redisAsyncFree(ac);
     }
-    ac = nullptr;
 }
 
 bool ASyncRedis::WaitForConnected()

SMASyncRedis.cpp

@@ -32,25 +32,20 @@ static inline redisReply *CopyReply(redisReply *src)
     return reply;
 }
 
-void SMASyncRedis::CmdCallback(const redisAsyncContext *c, void *r, void *privdata)
-{
-    if (privdata) {
-        CmdCBData *data = (CmdCBData *)privdata;
-        data->reply = CopyReply((redisReply *)r);
-
-        OutputDebugString(std::to_string((int)data->reply).c_str());
-
-        g_HiredisExt.RequestFrame(std::bind(&SMASyncRedis::_cmdcb, this, std::placeholders::_1), privdata);
-    }
-}
-
-void SMASyncRedis::_cmdcb(void *data)
+static void _cmdcb(void *data)
 {
-    CmdCBData *d = (CmdCBData *)data;
+    CBData *d = (CBData *)data;
     IPluginFunction *cb = d->callback;
     if (cb->IsRunnable()) {
         HandleSecurity sec(d->identity, myself->GetIdentity());
 
+        void *obj;
+        int err = handlesys->ReadHandle(d->handle, g_AsyncRedisCtxType, &sec, &obj);
+        if (err != HandleError_None || obj == nullptr) {
+            delete d;
+            return;
+        }
+
         RedisReply *reply = new RedisReply();
         reply->reply = d->reply;
 
@@ -62,7 +57,7 @@ void SMASyncRedis::_cmdcb(void *data)
             return;
         }
 
-        cb->PushCell(handle);
+        cb->PushCell(d->handle);
         cb->PushCell(qh);
         cb->PushCell(d->data);
         cb->Execute(nullptr);
@@ -74,36 +69,62 @@ void SMASyncRedis::_cmdcb(void *data)
     delete data;
 }
 
-void SMASyncRedis::ConnectCallback(int status)
+void SMASyncRedis::CmdCallback(const redisAsyncContext *c, void *r, void *privdata)
 {
-    ASyncRedis::ConnectCallback(status);
-    g_HiredisExt.RequestFrame(std::bind(&SMASyncRedis::_connectcb, this, std::placeholders::_1), (void *)status);
+    if (privdata) {
+        CBData *data = (CBData *)privdata;
+        data->reply = CopyReply((redisReply *)r);
+
+        g_HiredisExt.RequestFrame(_cmdcb, privdata);
+    }
 }
 
-void SMASyncRedis::_connectcb(void *data)
+static void _connectcb(void *data)
 {
-    int status = (int)data;
-    if (connectedCb && connectedCb->IsRunnable()) {
-        connectedCb->PushCell(handle);
-        connectedCb->PushCell(status);
-        connectedCb->PushCell(connectData);
-        connectedCb->Execute(nullptr);
+    CBData *d = (CBData *)data;
+    IPluginFunction *cb = d->callback;
+    if (cb && cb->IsRunnable()) {
+        HandleSecurity sec;
+        sec.pOwner = NULL;
+        sec.pIdentity = myself->GetIdentity();
+        void *obj;
+        int err = handlesys->ReadHandle(d->handle, g_AsyncRedisCtxType, &sec, &obj);
+        if (err != HandleError_None || obj == nullptr) {
+            delete d;
+            return;
+        }
+
+        cb->PushCell(d->handle);
+        cb->PushCell(d->status);
+        cb->PushCell(d->data);
+        cb->Execute(nullptr);
     }
+
+    delete d;
 }
 
-void SMASyncRedis::DisconnectCallback(int status)
+void SMASyncRedis::ConnectCallback(int status)
 {
-    ASyncRedis::DisconnectCallback(status);
-    g_HiredisExt.RequestFrame(std::bind(&SMASyncRedis::_disconnectcb, this, std::placeholders::_1), (void *)status);
+    ASyncRedis::ConnectCallback(status);
+
+    CBData *cbdata = new CBData();
+    cbdata->handle = handle;
+    cbdata->callback = connectedCb;
+    cbdata->data = connectData;
+    cbdata->status = status;
+
+    g_HiredisExt.RequestFrame(_connectcb, (void *)cbdata);
 }
 
-void SMASyncRedis::_disconnectcb(void *data)
+void SMASyncRedis::DisconnectCallback(int status)
 {
-    int status = (int)data;
-    if (disconnectedCb && disconnectedCb->IsRunnable()) {
-        disconnectedCb->PushCell(handle);
-        disconnectedCb->PushCell(status);
-        disconnectedCb->PushCell(disconnectData);
-        disconnectedCb->Execute(nullptr);
-    }
+    ASyncRedis::DisconnectCallback(status);
+
+    CBData *cbdata = new CBData();
+    cbdata->handle = handle;
+    cbdata->callback = disconnectedCb;
+    cbdata->data = disconnectData;
+    cbdata->status = status;
+
+    g_HiredisExt.RequestFrame(_connectcb, (void *)cbdata);
 }

SMASyncRedis.h

@@ -19,17 +19,19 @@ extern "C" {
 #include <vector>
 #include <functional>
 
+struct CBData
+{
+    Handle_t handle;
+    int status;
+    IPluginFunction *callback;
+    IdentityToken_t *identity;
+    redisReply *reply;
+    cell_t data;
+};
+
 class SMASyncRedis :public ASyncRedis
 {
 public:
-    struct CmdCBData
-    {
-        IPluginFunction *callback;
-        IdentityToken_t *identity;
-        redisReply *reply;
-        cell_t data;
-    };
-
     SMASyncRedis(int intval = 1000, int cache_size = 32) :ASyncRedis(intval, cache_size) {}
 
     IPluginFunction *connectedCb;
@@ -41,11 +43,7 @@ class SMASyncRedis :public ASyncRedis
     Handle_t handle;
 
     void CmdCallback(const redisAsyncContext *c, void *r, void *privdata);
-    void _cmdcb(void *data);
 
     void ConnectCallback(int status);
-    void _connectcb(void *data);
-
     void DisconnectCallback(int status);
-    void _disconnectcb(void *data);
 };

addons/sourcemod/extensions/hiredis.ext.dll

@@ -15,7 +15,13 @@ public void ConnectCallback(RedisAsync ac, int status, any data)
 {
     PrintToServer("ConnectCallback: %d %d", status, data);
 
+    // you can send "auth" right after calling connect
     ac.Command("auth foobared233");
+
+    for (int i = 0; i < 10; ++i) {
+        ac.Command("lpush alsit %d", i);
+    }
+
     ac.Command("set foo %s", "wtf");
     ac.CommandEx(CommandCallback, 0, "get foo");
     ac.CommandEx(CommandCallback, 0, "lrange alist 0 -1");
@@ -29,6 +35,7 @@ public void DisconnectCallback(RedisAsync ac, int status, any data)
 
 public void CommandCallback(RedisAsync ac, RedisReply reply, any data)
 {
+    // This will print only one reply if 
     PrintReply(reply);
 }
 
@@ -51,6 +58,18 @@ public Action Cmd_ASyncHiredisTest(int client, int args)
         return Plugin_Handled;
     }
 
+    RedisAsync async2 = new RedisAsync();
+    async2.SetConnectCallback(ConnectCallback, 444);
+    async2.SetDisconnectCallback(DisconnectCallback, 666);
+
+    if (!async2.Connect("127.0.0.1", 6379)) {
+        char buf[256];
+        async2.GetErrorString(buf, sizeof(buf));
+        PrintToServer("RedisAsync 2 Connect failed: %d, %s", async2.GetError(), buf);
+        return Plugin_Handled;
+    }
+    delete async2; // there will be no call back for async2
+
     return Plugin_Handled;
 }
 
@@ -68,6 +87,19 @@ public Action Cmd_HiredisTest(int client, int args)
     PrintReply(reply);
     delete reply;
 
+    for (int i = 0; i < 10; ++i) {
+        redis.AppendCommand("lpush alist %d", i);
+    }
+
+    for (int i = 0; i < 10; ++i) {
+        reply = redis.GetReply();
+        if (reply == INVALID_HANDLE) {
+            ThrowError("Replay is invalid.")
+        }
+        PrintReply(reply);
+        delete reply;
+    }
+
     redis.AppendCommand("set foo %s", "23333333");
     reply = redis.GetReply();
     if (reply == INVALID_HANDLE) {

addons/sourcemod/extensions/hiredis.ext.pdb

@@ -112,13 +112,6 @@ void HiredisExt::SDK_OnUnload()
     handlesys->RemoveType(g_AsyncRedisCtxType, myself->GetIdentity());
 }
 
-//void HiredisExt::AddCallback(IPluginFunction * callback, std::vector<std::tuple<DataType, cell_t>> params)
-//{
-//    if (callback) {
-//        cblist.enqueue({ callback, params });
-//    }
-//}
-
 void HiredisExt::RequestFrame(std::function<void(void * data)> cb, void *data)
 {
     cblist.enqueue({ cb, data });

addons/sourcemod/scripting/hiredis_test.sp

@@ -364,7 +364,8 @@ static cell_t native_redisAsyncCommandEx(IPluginContext *pContext, const cell_t
         }
     }
 
-    SMASyncRedis::CmdCBData *cbdata = new SMASyncRedis::CmdCBData();
+    CBData *cbdata = new CBData();
+    cbdata->handle = params[1];
     cbdata->callback = pContext->GetFunctionById(params[2]);
     cbdata->data = params[3];
     cbdata->identity = pContext->GetIdentity();