1

server/wsproxy/GCListener.cc

@@ -52,14 +52,19 @@ public:
     virtual void OnRawHttpGet(const std::string& url, const std::string& querystr,
                               std::string& response) override
     {
-        #if 0
-        App::Instance()->AddIMMsg(IM_ExecGM,
-                                  a8::XParams()
-                                  .SetSender(socket_handle)
-                                  .SetParam1(url)
-                                  .SetParam2(querystr)
-                                  .SetParam3(saddr));
-        #endif
+        f8::MsgQueue::Instance()->PostMsg
+            (
+             IM_ExecGM,
+             a8::Args
+             (
+              {
+               (int)socket_handle,
+               (std::string)(url + ""),
+               (std::string)(querystr + ""),
+               (unsigned long)saddr
+              }
+              )
+             );
     }
 
     virtual bool HandleRedirect(const std::string& url, const std::string& querystr,

server/wsproxy/app.cc

@@ -128,7 +128,9 @@ bool App::Init(int argc, char* argv[])
             (perf_log_time,
              [] (int event, const a8::Args* args)
              {
-                 SavePerfLog();
+                 if (a8::TIMER_EXEC_EVENT == event) {
+                     SavePerfLog();
+                 }
              });
     }
     if (HasFlag(1)) {
@@ -137,8 +139,10 @@ bool App::Init(int argc, char* argv[])
              1000 * 60,
              [] (int event, const a8::Args* args)
              {
-                 App::Instance()->terminated = true;
-                 App::Instance()->NotifyLoopCond();
+                 if (a8::TIMER_EXEC_EVENT == event) {
+                     App::Instance()->terminated = true;
+                     App::Instance()->NotifyLoopCond();
+                 }
              }
              );
     }
@@ -148,8 +152,10 @@ bool App::Init(int argc, char* argv[])
              1000 * 30,
              [] (int event, const a8::Args* args)
              {
-                 App::Instance()->shutdowned = true;
-                 a8::XPrintf("shutdowned\n", {});
+                 if (a8::TIMER_EXEC_EVENT == event) {
+                     App::Instance()->shutdowned = true;
+                     a8::XPrintf("shutdowned\n", {});
+                 }
              }
              );
     }
@@ -423,49 +429,6 @@ void App::ProcessTargetServerMsg(f8::MsgHdr& hdr)
     GCListener::Instance()->ForwardUpStreamMsg(hdr);
 }
 
-#if 0
-void App::ProcessIMMsg()
-{
-    if (!im_work_node_ && im_top_node_) {
-        im_msg_mutex_->lock();
-        im_work_node_ = im_top_node_;
-        im_top_node_ = nullptr;
-        im_bot_node_ = nullptr;
-        im_msg_mutex_->unlock();
-    }
-    while (im_work_node_) {
-        IMMsgNode *pdelnode = im_work_node_;
-        switch (im_work_node_->msgid) {
-        case IM_UpStreamConnect:
-            {
-                DownStreamMgr::Instance()->OnTargetServerConnect(pdelnode->params);
-                UpStream* conn = UpStreamMgr::Instance()->GetConnById(pdelnode->params.sender);
-                if (conn && conn->Connected()) {
-                    conn->SendStockMsg();
-                }
-            }
-            break;
-        case IM_UpStreamDisconnect:
-            {
-                DownStreamMgr::Instance()->OnTargetServerDisconnect(pdelnode->params);
-            }
-            break;
-        case IM_ExecGM:
-            {
-                HandlerMgr::Instance()->ProcGMMsg(pdelnode->params.param3,
-                                                  pdelnode->params.sender,
-                                                  pdelnode->params.param1.GetString(),
-                                                  pdelnode->params.param2.GetString()
-                                                  );
-            }
-            break;
-        }
-        im_work_node_ = im_work_node_->next;
-        delete pdelnode;
-    }
-}
-#endif
-
 void App::InitLog()
 {
     std::string filename_fmt = PROJ_LOG_FILENAME_FMT;

server/wsproxy/handlermgr.cc

@@ -1,6 +1,7 @@
 #include "precompile.h"
 
 #include <a8/mutable_xobject.h>
+#include <f8/msgqueue.h>
 
 #include "handlermgr.h"
 
@@ -34,6 +35,17 @@ void HandlerMgr::Init()
     RegisterNetMsgHandlers();
     RegisterGMMsgHandler("Ops$selfChecking", _GMOpsSelfChecking);
     RegisterGMMsgHandler("Ops$getNodeId", _GMOpsGetNodeId);
+    f8::MsgQueue::Instance()->RegisterCallBack
+        (
+         IM_ExecGM,
+         [this] (const a8::Args& args)
+         {
+             int socket_handle = args.Get<int>(0);
+             std::string url = args.Get<std::string>(1);
+             std::string query_str = args.Get<std::string>(2);
+             unsigned long saddr = args.Get<unsigned long>(3);
+             ProcGMMsg(saddr, socket_handle, url, query_str);
+         });
 }
 
 void HandlerMgr::UnInit()

server/wsproxy/upstreammgr.cc

@@ -1,5 +1,7 @@
 #include "precompile.h"
 
+#include <f8/msgqueue.h>
+
 #include "upstreammgr.h"
 #include "upstream.h"
 #include "jsondatamgr.h"
@@ -7,6 +9,23 @@
 
 void UpStreamMgr::Init()
 {
+    f8::MsgQueue::Instance()->RegisterCallBack
+        (
+         IM_UpStreamConnect,
+         [this] (const a8::Args& args)
+         {
+             int instance_id = args.Get<int>(0);
+             UpStream* conn = GetConnById(instance_id);
+             if (conn && conn->Connected()) {
+                 conn->SendStockMsg();
+             }
+         });
+    f8::MsgQueue::Instance()->RegisterCallBack
+        (
+         IM_UpStreamDisconnect,
+         [this] (const a8::Args& args)
+         {
+         });
 }
 
 void UpStreamMgr::UnInit()