1

server/bin/monitor.sh

@@ -1,11 +0,0 @@
-path_arr=$(echo $PWD|tr '/' '\n')
-path_arr=(${path_arr})
-GAME_ID=${path_arr[-5]}
-
-if echo $GAME_ID | grep -q '[^0-9]'
-then
-    echo 'game_id参数必须为数字'
-    exit 1
-fi
-
-python ../../third_party/tools/scripts/server/monitor.py wsproxy${GAME_ID} '-n1 -i1'

server/bin/wsproxy/2006_dev_config/zone1/node1/kcp_conf.json

@@ -0,0 +1,14 @@
+{
+  "open": 0,
+
+  "sndwnd": 128,
+  "rcvwnd": 128,
+
+  "nodelay": 2,
+  "interval": 10,
+  "resend": 2,
+  "nc": 1,
+
+  "rx_minrto": 10,
+  "fastresend": 1
+}

server/bin/wsproxy/2006_dev_config/zone1/node1/masterserver.cluster.json

@@ -0,0 +1,7 @@
+[
+  {
+    "instance_id": 1,
+    "ip": "127.0.0.1",
+    "port": 7621
+  }
+]

server/bin/wsproxy/2006_dev_config/zone1/node1/wsproxy.cluster.json

@@ -0,0 +1,16 @@
+[
+  {
+    "instance_id" : 1,
+    "listen_ip": "127.0.0.1",
+    "listen_port": 7601,
+    "listen_udp_host": "192.144.200.110",
+    "listen_udp_port": 17601
+  },
+  {
+    "instance_id" : 2,
+    "listen_ip": "127.0.0.1",
+    "listen_port": 7602,
+    "listen_udp_host": "192.144.200.110",
+    "listen_udp_port": 17602
+  }
+]

server/bin/wsproxy/2006_test_config/zone1/node1/kcp_conf.json

@@ -0,0 +1,14 @@
+{
+  "open": 0,
+
+  "sndwnd": 128,
+  "rcvwnd": 128,
+
+  "nodelay": 2,
+  "interval": 10,
+  "resend": 2,
+  "nc": 1,
+
+  "rx_minrto": 10,
+  "fastresend": 1
+}

server/bin/wsproxy/2006_test_config/zone1/node1/masterserver.cluster.json

@@ -0,0 +1,7 @@
+[
+  {
+    "instance_id": 1,
+    "ip": "127.0.0.1",
+    "port": 7621
+  }
+]

server/bin/wsproxy/2006_test_config/zone1/node1/wsproxy.cluster.json

@@ -0,0 +1,16 @@
+[
+  {
+    "instance_id" : 1,
+    "listen_ip": "127.0.0.1",
+    "listen_port": 7601,
+    "listen_udp_host": "192.144.200.110",
+    "listen_udp_port": 17601
+  },
+  {
+    "instance_id" : 2,
+    "listen_ip": "127.0.0.1",
+    "listen_port": 7602,
+    "listen_udp_host": "192.144.200.110",
+    "listen_udp_port": 17602
+  }
+]

server/bin/wsproxy/9007_test_config/zone1/node1/kcp_conf.json

@@ -0,0 +1,14 @@
+{
+  "open": 0,
+
+  "sndwnd": 128,
+  "rcvwnd": 128,
+
+  "nodelay": 2,
+  "interval": 10,
+  "resend": 2,
+  "nc": 1,
+
+  "rx_minrto": 10,
+  "fastresend": 1
+}

server/bin/wsproxy/9007_test_config/zone1/node1/masterserver.cluster.json

@@ -0,0 +1,7 @@
+[
+  {
+    "instance_id": 1,
+    "ip": "127.0.0.1",
+    "port": 7627
+  }
+]

server/bin/wsproxy/9007_test_config/zone1/node1/wsproxy.cluster.json

@@ -0,0 +1,9 @@
+[
+  {
+    "instance_id" : 1,
+    "listen_ip": "127.0.0.1",
+    "listen_port": 7607,
+    "listen_udp_host": "192.144.200.110",
+    "listen_udp_port": 17607
+  }
+]

server/bin/wsproxy/9008_test_config/zone1/node1/kcp_conf.json

@@ -0,0 +1,14 @@
+{
+  "open": 0,
+
+  "sndwnd": 128,
+  "rcvwnd": 128,
+
+  "nodelay": 2,
+  "interval": 10,
+  "resend": 2,
+  "nc": 1,
+
+  "rx_minrto": 10,
+  "fastresend": 1
+}

server/bin/wsproxy/9008_test_config/zone1/node1/masterserver.cluster.json

@@ -0,0 +1,7 @@
+[
+  {
+    "instance_id": 1,
+    "ip": "127.0.0.1",
+    "port": 7628
+  }
+]

server/bin/wsproxy/9008_test_config/zone1/node1/wsproxy.cluster.json

@@ -0,0 +1,9 @@
+[
+  {
+    "instance_id" : 1,
+    "listen_ip": "127.0.0.1",
+    "listen_port": 7608,
+    "listen_udp_host": "192.144.200.110",
+    "listen_udp_port": 17608
+  }
+]

server/bin/wsproxy/9009_test_config/zone1/node1/kcp_conf.json

@@ -0,0 +1,14 @@
+{
+  "open": 0,
+
+  "sndwnd": 128,
+  "rcvwnd": 128,
+
+  "nodelay": 2,
+  "interval": 10,
+  "resend": 2,
+  "nc": 1,
+
+  "rx_minrto": 10,
+  "fastresend": 1
+}

server/bin/wsproxy/9009_test_config/zone1/node1/masterserver.cluster.json

@@ -0,0 +1,7 @@
+[
+  {
+    "instance_id": 1,
+    "ip": "127.0.0.1",
+    "port": 7629
+  }
+]

server/bin/wsproxy/9009_test_config/zone1/node1/wsproxy.cluster.json

@@ -0,0 +1,9 @@
+[
+  {
+    "instance_id" : 1,
+    "listen_ip": "127.0.0.1",
+    "listen_port": 7609,
+    "listen_udp_host": "192.144.200.110",
+    "listen_udp_port": 17609
+  }
+]

server/bin/wsproxy/bin/wsproxy2006

@@ -0,0 +1 @@
+wsproxy

server/bin/wsproxy/bin/wsproxy9007

@@ -0,0 +1 @@
+wsproxy

server/bin/wsproxy/bin/wsproxy9008

@@ -0,0 +1 @@
+wsproxy

server/bin/wsproxy/bin/wsproxy9009

@@ -0,0 +1 @@
+wsproxy

server/tools/protobuild/ss_msgid.proto

@@ -9,6 +9,11 @@ enum SSMessageId_e
     _SS_WSP_RequestTargetServer = 11;
     _SS_MS_ResponseTargetServer = 12;
     _SS_ForceCloseSocket = 13;
+    _SS_WSP_HttpTunnelRequest = 14;
+    _SS_MS_HttpTunnelResponse = 15;
+    _SS_HttpTunnelRequest = 16;
+    _SS_HttpTunnelResponse = 17;
+    _SS_BindUpStreamSocket = 20;
 
     _SS_CMKcpHandshake = 99;
     _SS_CMPing = 101;

server/tools/protobuild/ss_proto.proto

@@ -78,6 +78,39 @@ message SS_MS_ResponseTargetServer
     optional int32 port = 5;
 }
 
+message SS_WSP_HttpTunnelRequest
+{
+    optional int64 context_id = 1;
+    optional string url = 7;
+    optional string query_str = 8;
+}
+
+message SS_MS_HttpTunnelResponse
+{
+    optional int32 error_code = 1;
+    optional string error_msg = 2;
+    optional int64 context_id = 3;
+    optional string host = 4;
+    optional int32 port = 5;
+}
+
+message SS_HttpTunnelRequest
+{
+    optional int64 context_id = 1;
+    optional int32 socket_handle = 2;
+    optional string url = 3;
+    optional string query_str = 4;
+}
+
+message SS_HttpTunnelResponse
+{
+    optional int32 error_code = 1;
+    optional string error_msg = 2;
+    optional int64 context_id = 3;
+    optional int32 socket_handle = 4;
+    optional string response = 5;
+}
+
 message SS_SMRpcError
 {
     optional int32 error_code = 1;
@@ -99,3 +132,7 @@ message SS_Pong
 message SS_ForceCloseSocket
 {
 }
+
+message SS_BindUpStreamSocket
+{
+}

server/wsproxy/CMakeLists.txt

@@ -18,11 +18,12 @@ else()
   message("debug mode")
 endif()
 
-set(CMAKE_CXX_FLAGS_RELEASE "-O3 -Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DNDEBUG")
+set(CMAKE_CXX_FLAGS_RELEASE "-O3 -Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DMMCHK=1 -DNDEBUG")
 if (${ASAN})
-  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -fsanitize=address -fno-omit-frame-pointer")
+  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DMMCHK=1 -DDEBUG -fsanitize=address -fno-omit-frame-pointer -fsanitize-recover=address")
+    message("asan mode")
 else()
-  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG")
+  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DMMCHK=1 -DDEBUG -fsanitize-recover=address")
 endif()
 
 include_directories(
@@ -74,11 +75,11 @@ aux_source_directory(.
 )
 
 set(EXECUTABLE_OUTPUT_PATH
-  ${PROJECT_BINARY_DIR}/../bin
+  ${PROJECT_BINARY_DIR}/../bin/wsproxy/bin
 )
 
 add_executable(
-  wsproxy${GAME_ID} ${SRC_LIST}
+  wsproxy ${SRC_LIST}
 )
 
 add_custom_target(script_pb_protocol ALL)
@@ -86,10 +87,10 @@ add_custom_command(TARGET script_pb_protocol
   PRE_BUILD
   COMMAND python ../tools/scripts/construct/build_pb.py
 )
-add_dependencies(wsproxy${GAME_ID} script_pb_protocol)
+add_dependencies(wsproxy script_pb_protocol)
 
 target_link_libraries(
-  wsproxy${GAME_ID}
+  wsproxy
   pthread
   mysqlclient
   protobuf
@@ -106,12 +107,12 @@ target_link_libraries(
 
 if (CMAKE_BUILD_TYPE STREQUAL "Debug")
   target_link_libraries(
-    wsproxy${GAME_ID}
+    wsproxy
     behaviac_gcc_debug
     )
 else()
   target_link_libraries(
-    wsproxy${GAME_ID}
+    wsproxy
     tcmalloc
     behaviac_gcc_release
     )

server/wsproxy/GCListener.cc

@@ -22,6 +22,9 @@ public:
 
     virtual void DecodeUserPacket(char* buf, int& offset, unsigned int buflen) override
     {
+        #if 1
+        is_activite = true;
+        #endif
         //packagelen + msgid + magiccode + msgbody
         //2 + 2 + 4+ xx + \0 + xx
         bool warning = false;
@@ -31,14 +34,15 @@ public:
                 if (buflen - offset < sizeof(f8::PackHead) + p->packlen) {
                     break;
                 }
-                App::Instance()->AddSocketMsg(SF_Client,
-                                              socket_handle,
-                                              saddr,
-                                              p->msgid,
-                                              p->seqid,
-                                              &buf[offset + sizeof(f8::PackHead)],
-                                              p->packlen,
-                                              ST_Tcp);
+                f8::App::Instance()->AddSocketMsg
+                    (SF_Client,
+                     socket_handle,
+                     saddr,
+                     p->msgid,
+                     p->seqid,
+                     &buf[offset + sizeof(f8::PackHead)],
+                     p->packlen,
+                     ST_Tcp);
                 offset += sizeof(f8::PackHead) + p->packlen;
             } else {
                 warning = true;
@@ -107,11 +111,6 @@ public:
 
 };
 
-static void CreateGameClientSocket(a8::TcpSession **p)
-{
-    *p = new GCClientSession();
-}
-
 static void GSListeneron_error(a8::TcpListener*, int type, int errorid)
 {
     f8::UdpLog::Instance()->Debug("GCListeneron_error %d %d", {type, errorid});
@@ -119,13 +118,14 @@ static void GSListeneron_error(a8::TcpListener*, int type, int errorid)
 
 void GCListener::Init()
 {
-    tcp_listener_ = new a8::TcpListener();
-    tcp_listener_->on_create_client_socket = CreateGameClientSocket;
+    tcp_listener_ = std::make_shared<a8::TcpListener>();
+    tcp_listener_->RegisterSessionClass<GCClientSession>(1024 * 64 * 2);
     tcp_listener_->on_error = GSListeneron_error;
 
     tcp_listener_->bind_address = "0.0.0.0";
     tcp_listener_->bind_port = JsonDataMgr::Instance()->GetConf()->At("listen_port")->AsXValue();
     tcp_listener_->Open();
+    a8::XPrintf("listener_port:%d\n", {tcp_listener_->bind_port});
     f8::MsgQueue::Instance()->RegisterCallBack
         (
          IM_HandleRedirect,
@@ -148,7 +148,6 @@ void GCListener::Init()
 
 void GCListener::UnInit()
 {
-    delete tcp_listener_;
     tcp_listener_ = nullptr;
 }
 

server/wsproxy/GCListener.h

@@ -29,7 +29,7 @@ class GCListener : public a8::Singleton<GCListener>
     template <typename T>
     void SendMsgEx(unsigned short socket_handle, int msgid, T& msg)
     {
-        f8::Net_SendMsg(tcp_listener_, socket_handle, 0, msgid, msg);
+        f8::Net_SendMsg(tcp_listener_.get(), socket_handle, 0, msgid, msg);
     }
 
     void SendBuf(unsigned short sockhandle, char* buf, int buflen);
@@ -43,6 +43,6 @@ class GCListener : public a8::Singleton<GCListener>
     int GetSocketCount();
 
 private:
-    a8::TcpListener *tcp_listener_ = nullptr;
+    std::shared_ptr<a8::TcpListener> tcp_listener_;
     std::map<int, std::tuple<std::string, std::string>> websocket_url_hash_;
 };

server/wsproxy/app.cc

@@ -1,10 +1,6 @@
 #include "precompile.h"
 
-#include <signal.h>
 #include <string.h>
-#include <unistd.h>
-#include <mutex>
-#include <condition_variable>
 
 #include <a8/xtimer.h>
 #include <a8/uuid.h>
@@ -12,6 +8,7 @@
 
 #include <f8/netmsghandler.h>
 #include <f8/udplog.h>
+#include <f8/app.h>
 #include <f8/msgqueue.h>
 
 #include "app.h"
@@ -31,49 +28,34 @@
 #include "ss_msgid.pb.h"
 #include "ss_proto.pb.h"
 
-struct MsgNode
-{
-    SocketFrom_e sockfrom;
-    int sockhandle;
-    unsigned short msgid;
-    unsigned int seqid;
-    long ip_saddr;
-    char* buf;
-    int buflen;
-    int tag;
-    MsgNode* next;
-};
-
 struct UdpMsgNode
 {
     a8::UdpPacket* pkt;
     UdpMsgNode* next;
 };
 
-const char* const PROJ_LOG_ROOT_FMT = "/data/logs/%s/logs";
-const char* const PROJ_LOG_FILENAME_FMT = "log_$pid_%Y%m%d.log";
-
 static void SavePerfLog()
 {
-    f8::UdpLog::Instance()->Info("max_run_delay_time:%d max_timer_idle:%d "
-                                 "in_data_size:%d out_data_size:%d msgnode_size:%d udp_msgnode_size:%d "
-                                 "read_count:%d max_login_time:%d "
-                                 "max_join_time:%d tcp_count:%d udp_count:%d down_stream_count:%d",
-                                 {
-                                     App::Instance()->GetPerf().max_run_delay_time,
-                                     App::Instance()->GetPerf().max_timer_idle,
-                                     App::Instance()->GetPerf().in_data_size,
-                                     App::Instance()->GetPerf().out_data_size,
-                                     App::Instance()->GetMsgNodeSize(),
-                                     App::Instance()->GetUdpMsgNodeSize(),
-                                     App::Instance()->GetPerf().read_count,
-                                     App::Instance()->GetPerf().max_login_time,
-                                     App::Instance()->GetPerf().max_join_time,
-                                     GCListener::Instance()->GetSocketCount(),
-                                     LongSessionMgr::Instance()->GetLongSessionCount(),
-                                     DownStreamMgr::Instance()->GetDownStreamCount()
-                                  });
-    if (App::Instance()->HasFlag(2)) {
+    f8::UdpLog::Instance()->Info
+        ("max_run_delay_time:%d max_timer_idle:%d "
+         "in_data_size:%d out_data_size:%d msgnode_size:%d udp_msgnode_size:%d "
+         "read_count:%d max_login_time:%d "
+         "max_join_time:%d tcp_count:%d udp_count:%d down_stream_count:%d",
+         {
+             App::Instance()->GetPerf().max_run_delay_time,
+                 App::Instance()->GetPerf().max_timer_idle,
+                 App::Instance()->GetPerf().in_data_size,
+                 App::Instance()->GetPerf().out_data_size,
+                 f8::App::Instance()->GetMsgNodeSize(),
+                 App::Instance()->GetUdpMsgNodeSize(),
+                 App::Instance()->GetPerf().read_count,
+                 App::Instance()->GetPerf().max_login_time,
+                 App::Instance()->GetPerf().max_join_time,
+                 GCListener::Instance()->GetSocketCount(),
+                 LongSessionMgr::Instance()->GetLongSessionCount(),
+                 DownStreamMgr::Instance()->GetDownStreamCount()
+                 });
+    if (f8::App::Instance()->HasFlag(2)) {
         a8::XPrintf("mainloop_time:%d netmsg_time:%d send_node_num:%d sent_bytes_num:%d\n",
                     {
                      App::Instance()->GetPerf().max_run_delay_time,
@@ -88,45 +70,17 @@ static void SavePerfLog()
     App::Instance()->GetPerf().max_join_time = 0;
 }
 
-bool App::Init(int argc, char* argv[])
+const std::string App::GetPkgName()
 {
-    signal(SIGPIPE, SIG_IGN);
-    this->argc_ = argc;
-    this->argv_ = argv;
+    return a8::Format("game%d_wsproxy", {App::Instance()->GetGameId()});
+}
 
-    if (!ParseOpt()) {
-        terminated_ = true;
-        if (node_id_ <= 0) {
-            a8::XPrintf("gameserver启动失败,缺少-n参数\n", {});
-        } else if (node_id_ > MAX_NODE_ID) {
-            a8::XPrintf("gameserver启动失败,-n参数不能大于%d\n", {MAX_NODE_ID});
-        } else if (instance_id_ <= 0) {
-            a8::XPrintf("gameserver启动失败,缺少-i参数\n", {});
-        } else if (instance_id_ > MAX_INSTANCE_ID) {
-            a8::XPrintf("gameserver启动失败,-i参数不能大于%d\n", {MAX_INSTANCE_ID});
-        }
-        return false;
-    }
-    a8::XPrintf("wsproxy starting node_id:%d instance_id:%d pid:%d\n",
-                {
-                    node_id_,
-                        instance_id_,
-                        getpid()
-                });
+void App::Init()
+{
+    udp_msg_mutex_ = std::make_shared<std::mutex>();
 
-    uuid_ = std::make_shared<a8::uuid::SnowFlake>();
-    loop_mutex_ = new std::mutex();
-    loop_cond_ = new std::condition_variable();
-    msg_mutex_ = new std::mutex();
-    udp_msg_mutex_ = new std::mutex();
-
-    srand(time(nullptr));
-    InitLog();
-    f8::MsgQueue::Instance()->Init();
     HandlerMgr::Instance()->Init();
-    f8::Timer::Instance()->Init();
     JsonDataMgr::Instance()->Init();
-    uuid_->SetMachineId((node_id_ - 1) * MAX_NODE_ID + instance_id_);
     DownStreamMgr::Instance()->Init();
     MasterMgr::Instance()->Init();
     UpStreamMgr::Instance()->Init();
@@ -135,8 +89,8 @@ bool App::Init(int argc, char* argv[])
 
     f8::UdpLog::Instance()->Info("wsproxy starting instance_id:%d pid:%d",
                                  {
-                                  instance_id_,
-                                  getpid(),
+                                  f8::App::Instance()->GetInstanceId(),
+                                  f8::App::Instance()->GetPid(),
                                  });
     {
         int perf_log_time = 1000 * 60 * 5;
@@ -152,120 +106,49 @@ bool App::Init(int argc, char* argv[])
                  }
              });
     }
-    if (HasFlag(1)) {
+    if (f8::App::Instance()->HasFlag(1)) {
         f8::Timer::Instance()->SetTimeout
             (
              1000 * 60,
              [] (int event, const a8::Args* args)
              {
                  if (a8::TIMER_EXEC_EVENT == event) {
-                     App::Instance()->terminated_ = true;
-                     App::Instance()->NotifyLoopCond();
+                     f8::App::Instance()->Terminated();
+                     f8::App::Instance()->NotifyLoopCond();
                  }
              }
              );
     }
-    if (HasFlag(4)) {
-        f8::Timer::Instance()->SetTimeout
-            (
-             1000 * 30,
-             [] (int event, const a8::Args* args)
-             {
-                 if (a8::TIMER_EXEC_EVENT == event) {
-                     App::Instance()->shutdowned_ = true;
-                     a8::XPrintf("shutdowned\n", {});
-                 }
-             }
-             );
-    }
-    return true;
 }
 
 void App::UnInit()
 {
-    a8::XPrintf("wsproxy terminating instance_id:%d pid:%d\n", {instance_id_, getpid()});
+    a8::XPrintf("wsproxy terminating instance_id:%d pid:%d\n",
+                {f8::App::Instance()->GetInstanceId(),
+                 f8::App::Instance()->GetPid()});
     GCListener::Instance()->UnInit();
     LongSessionMgr::Instance()->UnInit();
     MasterMgr::Instance()->UnInit();
     UpStreamMgr::Instance()->UnInit();
     DownStreamMgr::Instance()->UnInit();
     JsonDataMgr::Instance()->UnInit();
-    f8::Timer::Instance()->UnInit();
-    f8::MsgQueue::Instance()->UnInit();
     HandlerMgr::Instance()->UnInit();
-    UnInitLog();
 
-    FreeSocketMsgQueue();
     FreeUdpMsgQueue();
-    delete msg_mutex_;
-    msg_mutex_ = nullptr;
-    delete udp_msg_mutex_;
     udp_msg_mutex_ = nullptr;
-    delete loop_cond_;
-    loop_cond_ = nullptr;
-    delete loop_mutex_;
-    loop_mutex_ = nullptr;
-    a8::XPrintf("wsproxy terminated instance_id:%d pid:%d\n", {instance_id_, getpid()});
+    a8::XPrintf("wsproxy terminated instance_id:%d pid:%d\n",
+                {f8::App::Instance()->GetInstanceId(),
+                 f8::App::Instance()->GetPid()});
 }
 
-int App::Run()
+void App::Update(int delta_time)
 {
-    int ret = 0;
-    f8::UdpLog::Instance()->Info("wsproxy running", {});
-    while (!terminated_) {
-        a8::tick_t begin_tick = a8::XGetTickCount();
-        QuickExecute();
-        SlowerExecute();
-        a8::tick_t end_tick = a8::XGetTickCount();
-        if (end_tick - begin_tick > GetPerf().max_run_delay_time) {
-            GetPerf().max_run_delay_time = end_tick - begin_tick;
-        }
-        Schedule();
-    }
-    return ret;
-}
-
-void App::AddSocketMsg(SocketFrom_e sockfrom,
-                       int sockhandle,
-                       long ip_saddr,
-                       unsigned short msgid,
-                       unsigned int seqid,
-                       const char *msgbody,
-                       int bodylen,
-                       int tag)
-{
-    MsgNode *p = (MsgNode*)malloc(sizeof(MsgNode));
-    memset(p, 0, sizeof(MsgNode));
-    p->sockfrom = sockfrom;
-    p->ip_saddr = ip_saddr;
-    p->sockhandle = sockhandle;
-    p->msgid = msgid;
-    p->seqid = seqid;
-    p->buf = nullptr;
-    p->buflen = bodylen;
-    p->tag = tag;
-    if (bodylen > 0) {
-        p->buf = (char*)malloc(bodylen);
-        memmove(p->buf, msgbody, bodylen);
-    }
-    msg_mutex_->lock();
-    if (bot_node_) {
-        bot_node_->next = p;
-        bot_node_ = p;
-    } else {
-        top_node_ = p;
-        bot_node_ = p;
-    }
-    ++msgnode_size_;
-    msg_mutex_->unlock();
-    NotifyLoopCond();
+    QuickExecute();
+    SlowerExecute();
 }
 
 void App::QuickExecute()
 {
-    f8::Timer::Instance()->Update();
-    f8::MsgQueue::Instance()->Update();
-    DispatchMsg();
     DispatchUdpMsg();
     LongSessionMgr::Instance()->Update();
 }
@@ -274,45 +157,8 @@ void App::SlowerExecute()
 {
 }
 
-void App::NotifyLoopCond()
-{
-    std::unique_lock<std::mutex> lk(*loop_mutex_);
-    loop_cond_->notify_all();
-}
-
-void App::Schedule()
-{
-    std::unique_lock<std::mutex> lk(*loop_mutex_);
-    if (!HasTask()) {
-        #if 1
-        int sleep_time = 1;
-        loop_cond_->wait_for(lk, std::chrono::milliseconds(sleep_time));
-        #else
-        int sleep_time = f8::Timer::Instance()->GetIdleableMillSeconds();
-        loop_cond_->wait_for(lk, std::chrono::milliseconds(sleep_time));
-        if (sleep_time > perf.max_timer_idle) {
-            perf.max_timer_idle = sleep_time;
-        }
-        #endif
-    }
-}
-
 bool App::HasTask()
 {
-    {
-        if (!work_node_) {
-            msg_mutex_->lock();
-            if (!work_node_ && top_node_) {
-                work_node_ = top_node_;
-                top_node_ = nullptr;
-                bot_node_ = nullptr;
-            }
-            msg_mutex_->unlock();
-        }
-        if (work_node_) {
-            return true;
-        }
-    }
     {
         if (!udp_work_node_) {
             udp_msg_mutex_->lock();
@@ -330,73 +176,18 @@ bool App::HasTask()
     return false;
 }
 
-void App::DispatchMsg()
+void App::ProcessClientMsg(f8::MsgHdr* hdr)
 {
-    long long starttick = a8::XGetTickCount();
-    if (!work_node_ && top_node_) {
-        msg_mutex_->lock();
-        work_node_ = top_node_;
-        top_node_ = nullptr;
-        bot_node_ = nullptr;
-        working_msgnode_size_ = msgnode_size_;
-        msg_mutex_->unlock();
-    }
-
-    f8::MsgHdr hdr;
-    while (work_node_) {
-        MsgNode *pdelnode = work_node_;
-        work_node_ = pdelnode->next;
-        hdr.msgid = pdelnode->msgid;
-        hdr.seqid = pdelnode->seqid;
-        hdr.socket_handle = pdelnode->sockhandle;
-        hdr.buf = pdelnode->buf;
-        hdr.buflen = pdelnode->buflen;
-        hdr.offset = 0;
-        hdr.ip_saddr = pdelnode->ip_saddr;
-        switch (pdelnode->sockfrom) {
-        case SF_Client:
-            {
-                ProcessClientMsg(hdr, pdelnode->tag);
-            }
-            break;
-        case SF_TargetServer:
-            {
-                ProcessTargetServerMsg(hdr, pdelnode->tag);
-            }
-            break;
-        case SF_MasterServer:
-            {
-                ProcessMasterServerMsg(hdr, pdelnode->tag);
-            }
-            break;
-        }
-        if (pdelnode->buf) {
-            free(pdelnode->buf);
-        }
-        free(pdelnode);
-        working_msgnode_size_--;
-        if (a8::XGetTickCount() - starttick > 200) {
-            break;
-        }
-    }//end while
-
-    if (!work_node_) {
-        working_msgnode_size_ = 0;
-    }
-}
-
-void App::ProcessClientMsg(f8::MsgHdr& hdr, int tag)
-{
-    if (hdr.msgid == ss::_SS_CMLogin ||
-        hdr.msgid == ss::_SS_CMReconnect ||
-        hdr.msgid == ss::_SS_CMKcpHandshake) {
-        auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr.socket_handle);
+    if (hdr->msgid == ss::_SS_CMLogin ||
+        hdr->msgid == ss::_SS_CMReconnect ||
+        hdr->msgid == ss::_SS_CMKcpHandshake) {
+        auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr->socket_handle);
         if (down_wp.expired()) {
-            switch (hdr.msgid) {
+            switch (hdr->msgid) {
             case ss::_SS_CMLogin:
                 {
                     ss::SS_CMLogin msg;
-                    bool ok = msg.ParseFromArray(hdr.buf + hdr.offset, hdr.buflen - hdr.offset);
+                    bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
                     if (ok) {
                         MasterMgr::Instance()->RequestTargetServer
                             (hdr,
@@ -412,7 +203,7 @@ void App::ProcessClientMsg(f8::MsgHdr& hdr, int tag)
             case ss::_SS_CMReconnect:
                 {
                     ss::SS_CMReconnect msg;
-                    bool ok = msg.ParseFromArray(hdr.buf + hdr.offset, hdr.buflen - hdr.offset);
+                    bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
                     if (ok) {
                         MasterMgr::Instance()->RequestTargetServer
                             (hdr,
@@ -428,7 +219,7 @@ void App::ProcessClientMsg(f8::MsgHdr& hdr, int tag)
             case ss::_SS_CMKcpHandshake:
                 {
                     ss::SS_CMKcpHandshake msg;
-                    bool ok = msg.ParseFromArray(hdr.buf + hdr.offset, hdr.buflen - hdr.offset);
+                    bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
                     if (ok) {
                         LongSessionMgr::Instance()->_SS_CMKcpHandshake(hdr, msg);
                     }
@@ -442,17 +233,17 @@ void App::ProcessClientMsg(f8::MsgHdr& hdr, int tag)
             }
         }
     } else {
-        auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr.socket_handle);
+        auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr->socket_handle);
         if (auto down = down_wp.lock(); !down_wp.expired()) {
-            down->ProcCMMsg(hdr, tag);
+            down->ProcCMMsg(hdr);
         }
     }
 }
 
-void App::ProcessMasterServerMsg(f8::MsgHdr& hdr, int tag)
+void App::ProcessMasterServerMsg(f8::MsgHdr* hdr)
 {
     f8::NetMsgHandler* handler = f8::GetNetMsgHandler(&HandlerMgr::Instance()->msmsghandler,
-                                                      hdr.msgid);
+                                                      hdr->msgid);
     if (handler) {
         switch (handler->handlerid) {
         case HID_MasterMgr:
@@ -462,104 +253,32 @@ void App::ProcessMasterServerMsg(f8::MsgHdr& hdr, int tag)
     }
 }
 
-void App::ProcessTargetServerMsg(f8::MsgHdr& hdr, int tag)
+void App::ProcessTargetServerMsg(f8::MsgHdr* hdr)
 {
-    if (hdr.msgid == ss::_SS_ForceCloseSocket) {
-        GCListener::Instance()->ForceCloseClient(hdr.socket_handle);
+    if (hdr->msgid == ss::_SS_ForceCloseSocket) {
+        GCListener::Instance()->ForceCloseClient(hdr->socket_handle);
         return;
     }
-    if (hdr.msgid == ss::_SS_CMLogin || hdr.msgid == ss::_SS_CMReconnect) {
-        DownStreamMgr::Instance()->BindUpStream(hdr.socket_handle, hdr.ip_saddr);
-        GCListener::Instance()->MarkClient(hdr.socket_handle, true);
+    if (hdr->msgid == ss::_SS_CMLogin ||
+        hdr->msgid == ss::_SS_CMReconnect ||
+        hdr->msgid == ss::_SS_BindUpStreamSocket) {
+        DownStreamMgr::Instance()->BindUpStream(hdr->socket_handle, hdr->ip_saddr);
+        GCListener::Instance()->MarkClient(hdr->socket_handle, true);
     }
-    auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr.socket_handle);
+    if (hdr->msgid == ss::_SS_HttpTunnelResponse) {
+        ss::SS_HttpTunnelResponse msg;
+        bool ok = msg.ParseFromArray(hdr->buf + hdr->offset, hdr->buflen - hdr->offset);
+        if (ok) {
+            MasterMgr::Instance()->_SS_HttpTunnelResponse(hdr, msg);
+        }
+        return;
+    }
+    auto down_wp = DownStreamMgr::Instance()->GetDownStream(hdr->socket_handle);
     if (!down_wp.expired()) {
         down_wp.lock()->ForwardUpStreamMsg(hdr);
     }
 }
 
-void App::InitLog()
-{
-    std::string filename_fmt = PROJ_LOG_FILENAME_FMT;
-    a8::ReplaceString(filename_fmt, "$pid", a8::XValue(getpid()));
-
-    std::string proj_root_dir = a8::Format(PROJ_ROOT_FMT, {a8::Format(PROJ_NAME_FMT,{GAME_ID})});
-    std::string proj_log_root_dir = a8::Format(PROJ_LOG_ROOT_FMT, {a8::Format(PROJ_NAME_FMT, {GAME_ID})});
-    std::string log_file_name = a8::Format(PROJ_LOG_ROOT_FMT,
-                                           {a8::Format(PROJ_NAME_FMT, {GAME_ID})}) + "/" + filename_fmt;
-
-    a8::MkDir(proj_root_dir);
-    a8::MkDir(proj_log_root_dir);
-    f8::UdpLog::Instance()->SetLogFileName(log_file_name);
-    f8::UdpLog::Instance()->Init();
-    f8::UdpLog::Instance()->Info("proj_root_dir:%s", {proj_root_dir});
-    f8::UdpLog::Instance()->Info("proj_log_root_dir:%s", {proj_log_root_dir});
-    f8::UdpLog::Instance()->Info("log_file_name:%s", {log_file_name});
-}
-
-void App::UnInitLog()
-{
-    f8::UdpLog::Instance()->UnInit();
-}
-
-bool App::ParseOpt()
-{
-    int ch = 0;
-    while ((ch = getopt(argc_, argv_, "n:i:f:")) != -1) {
-        switch (ch) {
-        case 'n':
-            {
-                node_id_ = a8::XValue(optarg);
-            }
-            break;
-        case 'i':
-            {
-                instance_id_ = a8::XValue(optarg);
-            }
-            break;
-        case 'f':
-            {
-                std::vector<std::string> strings;
-                a8::Split(optarg, strings, ',');
-                for (auto& str : strings) {
-                    flags_.insert(a8::XValue(str).GetInt());
-                }
-            }
-            break;
-        }
-    }
-    return instance_id_ > 0 && node_id_ > 0;
-}
-
-bool App::HasFlag(int flag)
-{
-    return flags_.find(flag) != flags_.end();
-}
-
-void App::FreeSocketMsgQueue()
-{
-    msg_mutex_->lock();
-    if (!work_node_) {
-        work_node_ = top_node_;
-        top_node_ = nullptr;
-        bot_node_ = nullptr;
-    }
-    while (work_node_) {
-        MsgNode* pdelnode = work_node_;
-        work_node_ = work_node_->next;
-        if (pdelnode->buf) {
-            free(pdelnode->buf);
-        }
-        free(pdelnode);
-        if (!work_node_) {
-            work_node_ = top_node_;
-            top_node_ = nullptr;
-            bot_node_ = nullptr;
-        }
-    }
-    msg_mutex_->unlock();
-}
-
 void App::FreeUdpMsgQueue()
 {
     udp_msg_mutex_->lock();
@@ -602,7 +321,7 @@ void App::AddUdpMsg(a8::UdpPacket* pkt)
     }
     ++udp_msgnode_size_;
     udp_msg_mutex_->unlock();
-    NotifyLoopCond();
+    f8::App::Instance()->NotifyLoopCond();
 }
 
 void App::DispatchUdpMsg()
@@ -639,7 +358,42 @@ void App::DispatchUdpMsg()
     }
 }
 
-long long App::NewUuid()
+int App::GetGameId() const
 {
-    return uuid_->Generate();
+#ifdef DEBUG
+    std::string game_id = a8::XValue((f8::App::Instance()->GetArgv())[0]).GetString();
+    std::vector<std::string> strings;
+    a8::Split(game_id, strings, '/');
+    game_id = strings.at(strings.size() - 1);
+    a8::ReplaceString(game_id, "wsproxy", "");
+    return a8::XValue(game_id);
+#else
+    return GAME_ID;
+#endif
+}
+
+void App::DispatchSocketMsg(f8::MsgHdr* hdr)
+{
+    switch (hdr->sockfrom) {
+    case SF_Client:
+        {
+            ProcessClientMsg(hdr);
+        }
+        break;
+    case SF_TargetServer:
+        {
+            ProcessTargetServerMsg(hdr);
+        }
+        break;
+    case SF_MasterServer:
+        {
+            ProcessMasterServerMsg(hdr);
+        }
+        break;
+    default:
+        {
+            A8_ABORT();
+        }
+        break;
+    }
 }

server/wsproxy/app.h

@@ -2,18 +2,16 @@
 
 #include <a8/singleton.h>
 
+#include <f8/app.h>
+
 namespace a8
 {
     struct UdpPacket;
-    namespace uuid
-    {
-        class SnowFlake;
-    }
 }
 
 struct MsgNode;
 struct UdpMsgNode;
-class App : public a8::Singleton<App>
+class App : public a8::Singleton<App>, public f8::UserApp
 {
 private:
     App() {};
@@ -21,79 +19,41 @@ private:
 
 public:
 
-    bool Init(int argc, char* argv[]);
-    void UnInit();
-
-    int Run();
-
-    void AddSocketMsg(SocketFrom_e sockfrom,
-                      int sockhandle,
-                      long ip_saddr,
-                      unsigned short msgid,
-                      unsigned int seqid,
-                      const char *msgbody,
-                      int bodylen,
-                      int tag = ST_Tcp);
+    virtual const std::string GetPkgName() override;
+    virtual void Init() override;
+    virtual void UnInit() override;
+    virtual void Update(int delta_time) override;
+    virtual bool HasTask() override;
+    virtual void DispatchSocketMsg(f8::MsgHdr* hdr) override;
 
     void AddUdpMsg(a8::UdpPacket* pkt);
 
-    void NotifyLoopCond();
-    bool HasFlag(int flag);
-    long long NewUuid();
-    int GetNodeId() { return node_id_; }
-    int GetInstanceId() { return instance_id_; }
     PerfMonitor& GetPerf() { return perf_; }
-    int GetMsgNodeSize() { return msgnode_size_;}
     int GetUdpMsgNodeSize() { return udp_msgnode_size_;}
 
+    int GetGameId() const;
+
 private:
     void QuickExecute();
     void SlowerExecute();
-    void Schedule();
-    bool HasTask();
 
-    void DispatchMsg();
     void DispatchUdpMsg();
 
-    void ProcessClientMsg(f8::MsgHdr& hdr, int tag);
-    void ProcessMasterServerMsg(f8::MsgHdr& hdr, int tag);
-    void ProcessTargetServerMsg(f8::MsgHdr& hdr, int tag);
+    void ProcessClientMsg(f8::MsgHdr* hdr);
+    void ProcessMasterServerMsg(f8::MsgHdr* hdr);
+    void ProcessTargetServerMsg(f8::MsgHdr* hdr);
 
-    void InitLog();
-    void UnInitLog();
-
-    bool ParseOpt();
-    void FreeSocketMsgQueue();
     void FreeUdpMsgQueue();
 
 private:
-    int argc_ = 0;
-    char** argv_ = nullptr;
     PerfMonitor perf_;
-    volatile bool terminated_ = false;
-    volatile bool shutdowned_ = false;
 
-    int node_id_ = 0;
-    int instance_id_ = 0;
-    std::set<int> flags_;
-
-    std::shared_ptr<a8::uuid::SnowFlake> uuid_;
-    std::mutex *loop_mutex_ = nullptr;
-    std::condition_variable *loop_cond_ = nullptr;
-
-    std::mutex* msg_mutex_ = nullptr;
-    MsgNode* top_node_ = nullptr;
-    MsgNode* bot_node_ = nullptr;
-    MsgNode* work_node_ = nullptr;
-
-    std::mutex* udp_msg_mutex_ = nullptr;
+    std::shared_ptr<std::mutex> udp_msg_mutex_;
     UdpMsgNode* udp_top_node_ = nullptr;
     UdpMsgNode* udp_bot_node_ = nullptr;
     UdpMsgNode* udp_work_node_ = nullptr;
 
-    int msgnode_size_ = 0 ;
     int udp_msgnode_size_ = 0 ;
-    int working_msgnode_size_ = 0;
     int udp_working_msgnode_size_ = 0;
 
 };

server/wsproxy/constant.h

@@ -20,7 +20,7 @@ enum InnerMesssage_e
     IM_ExecGM,
     IM_UpStreamDisconnect,
     IM_UpStreamConnect,
-    IM_HandleRedirect
+    IM_HandleRedirect,
 };
 
 //网络处理对象

server/wsproxy/downstream.cc

@@ -23,21 +23,21 @@ void DownStream::ReBindUpStream(std::weak_ptr<UpStream> up)
     up_ = up;
 }
 
-void DownStream::ForwardUpStreamMsg(f8::MsgHdr& hdr)
+void DownStream::ForwardUpStreamMsg(f8::MsgHdr* hdr)
 {
-    char* buff = (char*)malloc(sizeof(f8::PackHead) + hdr.buflen);
+    char* buff = (char*)malloc(sizeof(f8::PackHead) + hdr->buflen);
     f8::PackHead* head = (f8::PackHead*)buff;
-    head->packlen = hdr.buflen;
-    head->msgid = hdr.msgid;
-    head->seqid = hdr.seqid;
+    head->packlen = hdr->buflen;
+    head->msgid = hdr->msgid;
+    head->seqid = hdr->seqid;
     head->magic_code = f8::MAGIC_CODE;
-    head->ext_len = hdr.buflen >> 16;
-    if (hdr.buflen > 0) {
-        memmove(buff + sizeof(f8::PackHead), hdr.buf, hdr.buflen);
+    head->ext_len = hdr->buflen >> 16;
+    if (hdr->buflen > 0) {
+        memmove(buff + sizeof(f8::PackHead), hdr->buf, hdr->buflen);
     }
 
     if (auto long_session = long_session_wp_.lock(); !long_session_wp_.expired()) {
-        if (hdr.msgid == ss::_SS_CMPing) {
+        if (hdr->msgid == ss::_SS_CMPing) {
             ss::SS_SMPing msg;
             msg.set_source(1);
             {
@@ -46,10 +46,10 @@ void DownStream::ForwardUpStreamMsg(f8::MsgHdr& hdr)
                 buff = (char*)malloc(sizeof(f8::PackHead) + msg.ByteSize());
                 f8::PackHead* head = (f8::PackHead*)buff;
                 head->packlen = msg.ByteSize();
-                head->msgid = hdr.msgid;
-                head->seqid = hdr.seqid;
+                head->msgid = hdr->msgid;
+                head->seqid = hdr->seqid;
                 head->magic_code = f8::MAGIC_CODE;
-                head->ext_len = hdr.buflen >> 16;
+                head->ext_len = hdr->buflen >> 16;
                 msg.SerializeToArray(buff + sizeof(f8::PackHead), head->packlen);
                 long_session->GetKcpSession()->SendClientMsg(buff, sizeof(f8::PackHead) + head->packlen);
             }
@@ -57,7 +57,7 @@ void DownStream::ForwardUpStreamMsg(f8::MsgHdr& hdr)
             long_session->GetKcpSession()->SendClientMsg(buff, sizeof(f8::PackHead) + head->packlen);
         }
     } else {
-        GCListener::Instance()->SendBuf(hdr.socket_handle, buff, sizeof(f8::PackHead) + head->packlen);
+        GCListener::Instance()->SendBuf(hdr->socket_handle, buff, sizeof(f8::PackHead) + head->packlen);
     }
     free(buff);
 }
@@ -73,9 +73,9 @@ void DownStream::OnClose()
     }
 }
 
-void DownStream::ProcCMMsg(f8::MsgHdr& hdr, int tag)
+void DownStream::ProcCMMsg(f8::MsgHdr* hdr)
 {
-    if (hdr.msgid == ss::_SS_CMPing && IsLongSession() && tag == ST_Tcp) {
+    if (hdr->msgid == ss::_SS_CMPing && IsLongSession() && hdr->tag == ST_Tcp) {
         ss::SS_SMPing msg;
         GCListener::Instance()->SendMsgEx(socket_handle_, ss::_SS_CMPing, msg);
         if (!long_session_wp_.expired()) {

server/wsproxy/downstream.h

@@ -12,9 +12,9 @@ class DownStream
     std::weak_ptr<UpStream> GetUpStream() const { return up_; }
     void ReBindUpStream(std::weak_ptr<UpStream> up);
     bool IsLongSession() { return is_long_session_; }
-    void ProcCMMsg(f8::MsgHdr& hdr, int tag);
+    void ProcCMMsg(f8::MsgHdr* hdr);
 
-    void ForwardUpStreamMsg(f8::MsgHdr& hdr);
+    void ForwardUpStreamMsg(f8::MsgHdr* hdr);
     void OnClose();
 
 private:

server/wsproxy/downstreammgr.cc

@@ -47,7 +47,7 @@ void DownStreamMgr::OnClientDisconnect(int socket_handle)
         socket_hash_.erase(socket_handle);
     }
     RemovePendingAccount(socket_handle);
-    MasterMgr::Instance()->RemoveRequest(socket_handle);
+    MasterMgr::Instance()->OnDownStreamDisconnect(socket_handle);
 }
 
 void DownStreamMgr::OnUpStreamDisconnect(int instance_id)

server/wsproxy/handlermgr.cc

@@ -3,6 +3,7 @@
 #include <a8/mutable_xobject.h>
 #include <f8/msgqueue.h>
 #include <f8/jsonhttprequest.h>
+#include <f8/app.h>
 
 #include "handlermgr.h"
 
@@ -13,36 +14,53 @@
 
 #include "ss_proto.pb.h"
 
-static void _GMOpsSelfChecking(std::shared_ptr<f8::JsonHttpRequest> request)
+static void _GMOpsSelfChecking(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
 {
-    request->resp_xobj->SetVal("errcode", 0);
-    request->resp_xobj->SetVal("errmsg", "");
-    request->resp_xobj->SetVal("healthy", 1);
-    request->resp_xobj->SetVal("max_rundelay", 10);
+    request->GetResp()->SetVal("errcode", 0);
+    request->GetResp()->SetVal("errmsg", "");
+    request->GetResp()->SetVal("healthy", 1);
+    request->GetResp()->SetVal("max_rundelay", 10);
 }
 
-static void _GMOpsGetNodeId(std::shared_ptr<f8::JsonHttpRequest> request)
+static void _GMOpsGetNodeId(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
 {
-    request->resp_xobj->SetVal("errcode", 0);
-    request->resp_xobj->SetVal("errmsg", "");
-    request->resp_xobj->SetVal("node_id", App::Instance()->GetNodeId());
+    request->GetResp()->SetVal("errcode", 0);
+    request->GetResp()->SetVal("errmsg", "");
+    request->GetResp()->SetVal("node_id", f8::App::Instance()->GetNodeId());
 }
 
-static void _GMOpsSetKcpSwitch(std::shared_ptr<f8::JsonHttpRequest> request)
+static void _GMOpsSetKcpSwitch(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
 {
-    request->resp_xobj->SetVal("errcode", 0);
-    request->resp_xobj->SetVal("errmsg", "");
-    if (request->params->HasKey("open")) {
-        JsonDataMgr::Instance()->SetKcpSwitch(request->params->At("open")->AsXValue().GetInt());
+    request->GetResp()->SetVal("errcode", 0);
+    request->GetResp()->SetVal("errmsg", "");
+    if (request->GetParams()->HasKey("open")) {
+        JsonDataMgr::Instance()->SetKcpSwitch(request->GetParams()->At("open")->AsXValue().GetInt());
     }
-    request->resp_xobj->SetVal("is_open", JsonDataMgr::Instance()->GetKcpConf().open);
+    request->GetResp()->SetVal("is_open", JsonDataMgr::Instance()->GetKcpConf().open);
 }
 
-static void _GMOpsGetKcpSwitch(std::shared_ptr<f8::JsonHttpRequest> request)
+static void _GMOpsGetKcpSwitch(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
 {
-    request->resp_xobj->SetVal("errcode", 0);
-    request->resp_xobj->SetVal("errmsg", "");
-    request->resp_xobj->SetVal("is_open", JsonDataMgr::Instance()->GetKcpConf().open);
+    request->GetResp()->SetVal("errcode", 0);
+    request->GetResp()->SetVal("errmsg", "");
+    request->GetResp()->SetVal("is_open", JsonDataMgr::Instance()->GetKcpConf().open);
+}
+
+static void _GMOpsTerminate(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
+{
+    request->GetResp()->SetVal("errcode", 0);
+    request->GetResp()->SetVal("errmsg", "");
+    f8::App::Instance()->Terminate();
+}
+
+static void _GMHttpTunnelTeamRequest(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
+{
+    MasterMgr::Instance()->AddHttpTunnelRequest(socket_handle, request);
+}
+
+static void _GMHttpTunnelOnlineTeamRequest(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
+{
+    MasterMgr::Instance()->AddHttpTunnelRequest(socket_handle, request);
 }
 
 void HandlerMgr::Init()
@@ -52,6 +70,9 @@ void HandlerMgr::Init()
     RegisterGMMsgHandler("Ops$getNodeId", _GMOpsGetNodeId);
     RegisterGMMsgHandler("Ops$setKcpSwitch", _GMOpsSetKcpSwitch);
     RegisterGMMsgHandler("Ops$getKcpSwitch", _GMOpsGetKcpSwitch);
+    RegisterGMMsgHandler("Ops$terminate", _GMOpsTerminate);
+    RegisterGMMsgHandler("HttpTunnel$teamRequest", _GMHttpTunnelTeamRequest);
+    RegisterGMMsgHandler("HttpTunnel$onlineTeamRequest", _GMHttpTunnelOnlineTeamRequest);
     f8::MsgQueue::Instance()->RegisterCallBack
         (
          IM_ExecGM,
@@ -72,6 +93,7 @@ void HandlerMgr::UnInit()
 void HandlerMgr::RegisterNetMsgHandlers()
 {
     RegisterNetMsgHandler(&msmsghandler, &MasterMgr::_SS_MS_ResponseTargetServer);
+    RegisterNetMsgHandler(&msmsghandler, &MasterMgr::_SS_MS_HttpTunnelResponse);
 }
 
 void HandlerMgr::ProcGMMsg(unsigned long saddr, int sockhandle,
@@ -88,25 +110,24 @@ void HandlerMgr::ProcGMMsg(unsigned long saddr, int sockhandle,
     std::string msgname = a8::Get(request, "c").GetString() + "$" + a8::Get(request, "a").GetString();
     auto itr = gmhandlers_.find(msgname);
     if (itr != gmhandlers_.end()) {
-        auto request = std::make_shared<f8::JsonHttpRequest>();
-        request->saddr = saddr;
-        request->socket_handle = sockhandle;
-        request->query_str = querystr;
-        request->params->ReadFromUrlQueryString(querystr);
-        itr->second(request);
-
-        if (!request->pending){
-            std::string response;
-            request->resp_xobj->ToJsonStr(response);
-            GCListener::Instance()->SendText(sockhandle, a8::HttpResponse(response));
-        }
+        auto request = std::make_shared<f8::JsonHttpRequest>
+            (
+             saddr,
+             url,
+             querystr,
+             [sockhandle] (const a8::Args& args)
+             {
+                 std::string data = args.Get<std::string>(0);
+                 GCListener::Instance()->SendText(sockhandle, data);
+             });
+        itr->second(sockhandle, request);
     } else {
         GCListener::Instance()->SendText(sockhandle, a8::HttpResponse("{}"));
     }
 }
 
 void HandlerMgr::RegisterGMMsgHandler(const std::string& msgname,
-                                      void (*handler)(std::shared_ptr<f8::JsonHttpRequest>))
+                                      void (*handler)(int, std::shared_ptr<f8::JsonHttpRequest>))
 {
     gmhandlers_[msgname] = handler;
 }

server/wsproxy/handlermgr.h

@@ -30,7 +30,7 @@ class HandlerMgr : public a8::Singleton<HandlerMgr>
  private:
     void RegisterNetMsgHandlers();
     void RegisterGMMsgHandler(const std::string& msgname,
-                              void (*)(std::shared_ptr<f8::JsonHttpRequest>));
+                              void (*)(int, std::shared_ptr<f8::JsonHttpRequest>));
 
-    std::map<std::string, void (*)(std::shared_ptr<f8::JsonHttpRequest>)> gmhandlers_;
+    std::map<std::string, void (*)(int, std::shared_ptr<f8::JsonHttpRequest>)> gmhandlers_;
 };

server/wsproxy/jsondatamgr.cc

@@ -11,35 +11,36 @@
 void JsonDataMgr::Init()
 {
     if (!f8::IsOnlineEnv()) {
-        work_path_ = a8::Format
-            ("../../../conf_test/game%d/%s",
-             {
-              GAME_ID,
-              f8::IsTestEnv() ? "wsproxy.test" : "wsproxy.dev"
-             });
+        if (f8::IsTestEnv()) {
+            work_path_ = a8::Format("../%d_test_config", {App::Instance()->GetGameId()});
+        } else {
+            work_path_ = a8::Format("../%d_dev_config", {App::Instance()->GetGameId()});
+        }
     }
 
     std::string wsproxy_cluster_json_file;
     std::string master_cluster_json_file;
     std::string kcp_conf_json_file;
     wsproxy_cluster_json_file = a8::Format
-        ("%s/node%d/game%d.wsproxy.cluster.json",
+        ("%s/zone%d/node%d/wsproxy.cluster.json",
          {
           work_path_,
-          App::Instance()->GetNodeId(),
-          GAME_ID
+          f8::App::Instance()->GetZoneId(),
+          f8::App::Instance()->GetNodeId()
          });
     master_cluster_json_file = a8::Format
-        ("%s/node%d/game%d.masterserver.cluster.json",
+        ("%s/zone%d/node%d/masterserver.cluster.json",
          {
           work_path_,
-          App::Instance()->GetNodeId(),
-          GAME_ID
+          f8::App::Instance()->GetZoneId(),
+          f8::App::Instance()->GetNodeId(),
          });
     kcp_conf_json_file = a8::Format
-        ("%s/kcp_conf.json",
+        ("%s/zone%d/node%d/kcp_conf.json",
          {
           work_path_,
+          f8::App::Instance()->GetZoneId(),
+          f8::App::Instance()->GetNodeId(),
          });
 
     wsproxy_cluster_json_.ReadFromFile(wsproxy_cluster_json_file);
@@ -91,7 +92,7 @@ std::shared_ptr<a8::XObject> JsonDataMgr::GetConf()
 {
     for (int i = 0; i < wsproxy_cluster_json_.Size(); ++i) {
         std::shared_ptr<a8::XObject> conf = wsproxy_cluster_json_.At(i);
-        if (conf->At("instance_id")->AsXValue().GetInt() == App::Instance()->GetInstanceId()) {
+        if (conf->At("instance_id")->AsXValue().GetInt() == f8::App::Instance()->GetInstanceId()) {
             return conf;
         }
     }

server/wsproxy/kcpsession.cc

@@ -37,7 +37,7 @@ KcpSession::~KcpSession()
 void KcpSession::Init(int socket_handle, int secret_key_place)
 {
     socket_handle_ = socket_handle;
-    secret_key_ = App::Instance()->NewUuid();
+    secret_key_ = f8::App::Instance()->NewNodeUuid();
     kcp_ = ikcp_create(socket_handle_, (void*)this);
     const KcpConf& kcp_conf = JsonDataMgr::Instance()->GetKcpConf();
     ikcp_wndsize(kcp_, kcp_conf.sndwnd, kcp_conf.rcvwnd);
@@ -131,15 +131,16 @@ void KcpSession::DecodeUserPacketOld(char* buf, int& offset, unsigned int buflen
             if (buflen - offset < sizeof(f8::PackHead) + p->packlen + GetSecretKeyLen()) {
                 break;
             }
-            App::Instance()->AddSocketMsg(SF_Client,
-                                          socket_handle_,
-                                          0,
-                                          //saddr,
-                                          p->msgid,
-                                          p->seqid,
-                                          &buf[offset + sizeof(f8::PackHead) + GetSecretKeyLen()],
-                                          p->packlen,
-                                          ST_Udp);
+            f8::App::Instance()->AddSocketMsg
+                (SF_Client,
+                 socket_handle_,
+                 0,
+                 //saddr,
+                 p->msgid,
+                 p->seqid,
+                 &buf[offset + sizeof(f8::PackHead) + GetSecretKeyLen()],
+                 p->packlen,
+                 ST_Udp);
             offset += sizeof(f8::PackHead) + p->packlen + GetSecretKeyLen();
         } else {
             warning = true;
@@ -162,15 +163,16 @@ void KcpSession::DecodeUserPacketNew(char* buf, int& offset, unsigned int buflen
             if (buflen - offset < sizeof(f8::PackHead) + p->packlen) {
                 break;
             }
-            App::Instance()->AddSocketMsg(SF_Client,
-                                          socket_handle_,
-                                          0,
-                                          //saddr,
-                                          p->msgid,
-                                          p->seqid,
-                                          &buf[offset + sizeof(f8::PackHead)],
-                                          p->packlen,
-                                          ST_Udp);
+            f8::App::Instance()->AddSocketMsg
+                (SF_Client,
+                 socket_handle_,
+                 0,
+                 //saddr,
+                 p->msgid,
+                 p->seqid,
+                 &buf[offset + sizeof(f8::PackHead)],
+                 p->packlen,
+                 ST_Udp);
             offset += sizeof(f8::PackHead) + p->packlen;
         } else {
             warning = true;

server/wsproxy/longsession.cc

@@ -8,10 +8,10 @@
 #include "ss_msgid.pb.h"
 #include "ss_proto.pb.h"
 
-void LongSession::Init(f8::MsgHdr& hdr, const ss::SS_CMKcpHandshake& msg)
+void LongSession::Init(f8::MsgHdr* hdr, const ss::SS_CMKcpHandshake& msg)
 {
     kcp_session_ = std::make_shared<KcpSession>();
-    kcp_session_->Init(hdr.socket_handle, msg.secret_key_place());
+    kcp_session_->Init(hdr->socket_handle, msg.secret_key_place());
 }
 
 void LongSession::UnInit()

server/wsproxy/longsession.h

@@ -11,7 +11,7 @@ class LongSession
 
  public:
 
-    void Init(f8::MsgHdr& hdr, const ss::SS_CMKcpHandshake& msg);
+    void Init(f8::MsgHdr* hdr, const ss::SS_CMKcpHandshake& msg);
     void UnInit();
     void Update(long long tick);
 

server/wsproxy/longsessionmgr.cc

@@ -53,17 +53,17 @@ void LongSessionMgr::Update()
     }
 }
 
-void LongSessionMgr::_SS_CMKcpHandshake(f8::MsgHdr& hdr, const ss::SS_CMKcpHandshake& msg)
+void LongSessionMgr::_SS_CMKcpHandshake(f8::MsgHdr* hdr, const ss::SS_CMKcpHandshake& msg)
 {
     ss::SS_SMKcpHandshake respmsg;
     respmsg.set_errcode(0);
     if (!JsonDataMgr::Instance()->GetKcpConf().open) {
         respmsg.set_errcode(1);
         respmsg.set_errmsg("not support kcp");
-        GCListener::Instance()->SendMsgEx(hdr.socket_handle, ss::_SS_CMKcpHandshake, respmsg);
+        GCListener::Instance()->SendMsgEx(hdr->socket_handle, ss::_SS_CMKcpHandshake, respmsg);
         return;
     }
-    if (GetSession(hdr.socket_handle)) {
+    if (GetSession(hdr->socket_handle)) {
 #ifdef DEBUG
         abort();
 #endif
@@ -77,9 +77,9 @@ void LongSessionMgr::_SS_CMKcpHandshake(f8::MsgHdr& hdr, const ss::SS_CMKcpHands
     respmsg.set_secret_key(session->GetKcpSession()->GetSecretKeyDataPtr(), KcpSession::GetSecretKeyLen());
     respmsg.set_remote_host(JsonDataMgr::Instance()->GetUdpHost());
     respmsg.set_remote_port(JsonDataMgr::Instance()->GetUdpPort());
-    GCListener::Instance()->SendMsgEx(hdr.socket_handle, ss::_SS_CMKcpHandshake, respmsg);
+    GCListener::Instance()->SendMsgEx(hdr->socket_handle, ss::_SS_CMKcpHandshake, respmsg);
     {
-        int socket_handle = hdr.socket_handle;
+        int socket_handle = hdr->socket_handle;
         f8::Timer::Instance()->SetTimeout
             (
              1000 * 30,

server/wsproxy/longsessionmgr.h

@@ -26,7 +26,7 @@ class LongSessionMgr : public a8::Singleton<LongSessionMgr>
     void UnInit();
     void Update();
 
-    void _SS_CMKcpHandshake(f8::MsgHdr& hdr, const ss::SS_CMKcpHandshake& msg);
+    void _SS_CMKcpHandshake(f8::MsgHdr* hdr, const ss::SS_CMKcpHandshake& msg);
     void ProcUdpPacket(a8::UdpPacket* pkt);
     std::shared_ptr<LongSession> GetSession(int socket_handle);
     std::shared_ptr<a8::UdpListener> GetUdpListener() { return udp_listener_; }

server/wsproxy/main.cc

@@ -1,12 +1,10 @@
 #include "precompile.h"
+
+#include <f8/app.h>
+
 #include "app.h"
 
 int main(int argc, char* argv[])
 {
-    int exitcode = 0;
-    if (App::Instance()->Init(argc, argv)) {
-        exitcode = App::Instance()->Run();
-        App::Instance()->UnInit();
-    }
-    return exitcode;
+    return f8::App::Instance()->Run(argc, argv, App::Instance());
 }

server/wsproxy/master.cc

@@ -24,9 +24,7 @@ void Master::Init(int instance_id, const std::string& remote_ip, int remote_port
     recv_bufflen_ = 0;
     last_pong_tick = a8::XGetTickCount();
     recv_buff_ = (char*) malloc(PACK_MAX * 2);
-    tcp_client_ = std::make_shared<a8::TcpClient>();
-    tcp_client_->remote_address = remote_ip;
-    tcp_client_->remote_port = remote_port;
+    tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
     tcp_client_->on_error = std::bind(&Master::on_error, this, std::placeholders::_1, std::placeholders::_2);
     tcp_client_->on_connect = std::bind(&Master::on_connect, this, std::placeholders::_1);
     tcp_client_->on_disconnect = std::bind(&Master::on_disconnect, this, std::placeholders::_1);
@@ -77,8 +75,8 @@ void Master::on_error(a8::TcpClient* sender, int errorId)
     f8::UdpLog::Instance()->Error("Master errorid=%d remote_ip:%s remote_port:%d",
                                   {
                                    errorId,
-                                   sender->remote_address,
-                                   sender->remote_port
+                                   sender->GetRemoteAddress(),
+                                   sender->GetRemotePort()
                                   });
 }
 
@@ -116,13 +114,15 @@ void Master::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
             if (recv_bufflen_ - offset < sizeof(f8::PackHead) + p->packlen) {
                 break;
             }
-            App::Instance()->AddSocketMsg(SF_MasterServer,
-                                          0,
-                                          instance_id,
-                                          p->msgid,
-                                          p->seqid,
-                                          &recv_buff_[offset + sizeof(f8::PackHead)],
-                                          p->packlen);
+            f8::App::Instance()->AddSocketMsg
+                (SF_MasterServer,
+                 0,
+                 instance_id,
+                 p->msgid,
+                 p->seqid,
+                 &recv_buff_[offset + sizeof(f8::PackHead)],
+                 p->packlen,
+                 0);
             offset += sizeof(f8::PackHead) + p->packlen;
         } else {
             warning = true;

server/wsproxy/mastermgr.cc

@@ -3,10 +3,13 @@
 #include <unistd.h>
 
 #include <a8/openssl.h>
+#include <a8/mutable_xobject.h>
 
 #include <f8/timer.h>
 #include <f8/netmsghandler.h>
 #include <f8/protoutils.h>
+#include <f8/jsonhttprequest.h>
+#include <f8/utils.h>
 
 #include "mastermgr.h"
 #include "master.h"
@@ -18,6 +21,14 @@
 #include "downstreammgr.h"
 #include "GCListener.h"
 
+struct HttpTunnelRequest
+{
+    int socket_handle = 0;
+    long long context_id = 0;
+    f8::TimerWp timer_wp;
+    std::shared_ptr<f8::JsonHttpRequest> request;
+};
+
 class RequestTarget
 {
 public:
@@ -54,10 +65,19 @@ void MasterMgr::UnInit()
     mastersvr_hash_.clear();
 }
 
-void MasterMgr::_SS_MS_ResponseTargetServer(f8::MsgHdr& hdr, const ss::SS_MS_ResponseTargetServer& msg)
+void MasterMgr::_SS_MS_ResponseTargetServer(f8::MsgHdr* hdr, const ss::SS_MS_ResponseTargetServer& msg)
 {
     auto req = GetRequestByContextId(msg.context_id());
     if (req) {
+#ifdef DEBUG
+        a8::XPrintf("error_code:%d error_msg:%s host:%s port:%d\n",
+                    {
+                        msg.error_code(),
+                            msg.error_msg(),
+                            msg.host(),
+                            msg.port()
+                    });
+#endif
         if (msg.error_code() == 0) {
             std::weak_ptr<UpStream> conn = UpStreamMgr::Instance()->RecreateUpStream
                 (
@@ -66,6 +86,7 @@ void MasterMgr::_SS_MS_ResponseTargetServer(f8::MsgHdr& hdr, const ss::SS_MS_Res
                  );
             if (!conn.expired()) {
                 conn.lock()->ForwardClientMsgEx(req->hdr_copy);
+                f8::MsgHdr::Destroy(req->hdr_copy);
                 req->conn = conn;
                 req->hdr_copy = nullptr;
                 if (!req->timer_wp.expired()) {
@@ -85,13 +106,70 @@ void MasterMgr::_SS_MS_ResponseTargetServer(f8::MsgHdr& hdr, const ss::SS_MS_Res
     }
 }
 
+void MasterMgr::_SS_MS_HttpTunnelResponse(f8::MsgHdr* hdr, const ss::SS_MS_HttpTunnelResponse& msg)
+{
+#ifdef DEBUG
+    a8::XPrintf("SS_MS_HttpTunnelResponse %s\n" , {f8::PbToJson(&msg)});
+#endif
+    auto req = GetHttpTunnelRequestByContextId(msg.context_id());
+    if (req) {
+        if (msg.error_code() == 0) {
+            std::weak_ptr<UpStream> conn = UpStreamMgr::Instance()->RecreateUpStream
+                (
+                 msg.host(),
+                 msg.port()
+                 );
+            if (!conn.expired()) {
+                ss::SS_HttpTunnelRequest upmsg;
+                upmsg.set_context_id(msg.context_id());
+                upmsg.set_socket_handle(req->socket_handle);
+                upmsg.set_url(req->request->GetUrl());
+                std::string query_str;
+                req->request->GetParams()->ToUrlEncodeStr(query_str);
+                upmsg.set_query_str(query_str);
+                conn.lock()->SendMsg(req->socket_handle, upmsg);
+                return;
+            } else {
+                abort();
+            }
+        } else {
+            req->request->GetResp()->SetVal("errcode", 500);
+            req->request->GetResp()->SetVal("errmsg", "server internal error");
+            req->request->Response();
+            if (!req->timer_wp.expired()) {
+                f8::Timer::Instance()->Delete(req->timer_wp);
+            }
+            RemoveHttpTunnelRequest(req->socket_handle);
+        }
+    }
+}
+
+void MasterMgr::_SS_HttpTunnelResponse(f8::MsgHdr* hdr, const ss::SS_HttpTunnelResponse& msg)
+{
+#ifdef DEBUG
+    a8::XPrintf("SS_HttpTunnelResponse %s\n" , {f8::PbToJson(&msg)});
+#endif
+    auto req = GetHttpTunnelRequestByContextId(msg.context_id());
+    if (req) {
+#ifdef DEBUG
+        a8::XPrintf("SS_HttpTunnelResponse2 %s\n" , {f8::PbToJson(&msg)});
+#endif
+        req->request->GetResp()->ReadFromJsonString(msg.response());
+        req->request->Response();
+        if (!req->timer_wp.expired()) {
+            f8::Timer::Instance()->Delete(req->timer_wp);
+        }
+        RemoveHttpTunnelRequest(req->socket_handle);
+    }
+}
+
 std::shared_ptr<Master> MasterMgr::GetConnById(int instance_id)
 {
     auto itr = mastersvr_hash_.find(instance_id);
     return itr != mastersvr_hash_.end() ? itr->second : nullptr;
 }
 
-void MasterMgr::RequestTargetServer(f8::MsgHdr& hdr,
+void MasterMgr::RequestTargetServer(f8::MsgHdr* hdr,
                                     const std::string& team_id,
                                     const std::string& account_id,
                                     const std::string& session_id,
@@ -99,7 +177,16 @@ void MasterMgr::RequestTargetServer(f8::MsgHdr& hdr,
                                     int is_reconnect,
                                     int proto_version)
 {
-    if (GetRequestBySocket(hdr.socket_handle)) {
+#ifdef DEBUG
+    a8::XPrintf("RequestTargetServer team_id:%s account_id:%s server_info:%s is_reconnect:%d\n",
+                {
+                    team_id,
+                        account_id,
+                        server_info,
+                        is_reconnect
+                });
+#endif
+    if (GetRequestBySocket(hdr->socket_handle)) {
         return;
     }
     unsigned int code = 0;
@@ -110,7 +197,7 @@ void MasterMgr::RequestTargetServer(f8::MsgHdr& hdr,
         std::string data = a8::Format("!%s_%s_%d_%d",
                                       {
                                        account_id,
-                                       App::Instance()->NewUuid(),
+                                       f8::App::Instance()->NewNodeUuid(),
                                        getpid(),
                                        rand()
                                       });
@@ -122,10 +209,10 @@ void MasterMgr::RequestTargetServer(f8::MsgHdr& hdr,
         ++curr_context_id_;
         auto req = std::make_shared<RequestTarget>();
         req->context_id = curr_context_id_;
-        req->socket_handle = hdr.socket_handle;
+        req->socket_handle = hdr->socket_handle;
         req->account_id = account_id;
         req->req_tick = a8::XGetTickCount();
-        req->hdr_copy = hdr.Clone();
+        req->hdr_copy = hdr->Clone();
 
         ss::SS_WSP_RequestTargetServer msg;
         msg.set_context_id(curr_context_id_);
@@ -139,14 +226,14 @@ void MasterMgr::RequestTargetServer(f8::MsgHdr& hdr,
         std::string url;
         std::string query_str;
         {
-            GCListener::Instance()->GetWebSocketUrl(hdr.socket_handle, url, query_str);
+            GCListener::Instance()->GetWebSocketUrl(hdr->socket_handle, url, query_str);
         }
         msg.set_url(url);
         msg.set_query_str(query_str);
 #endif
         svr->SendMsg(msg);
 
-        pending_socket_hash_[hdr.socket_handle] = req;
+        pending_socket_hash_[hdr->socket_handle] = req;
         assert(pending_context_hash_.find(curr_context_id_) == pending_context_hash_.end());
         pending_context_hash_[curr_context_id_] = req;
         req->timer_wp = f8::Timer::Instance()->SetTimeoutWp
@@ -154,37 +241,58 @@ void MasterMgr::RequestTargetServer(f8::MsgHdr& hdr,
              [req] (int event, const a8::Args* args)
              {
                  if (a8::TIMER_EXEC_EVENT == event) {
-                     MasterMgr::Instance()->RemoveRequest
-                         (
-                          req->socket_handle
-                          );
+                     MasterMgr::Instance()->RemoveRequest(req->socket_handle);
                      long long req_handle_time = a8::XGetTickCount() - req->req_tick;
                      if (req_handle_time > App::Instance()->GetPerf().max_login_time) {
                          App::Instance()->GetPerf().max_login_time = req_handle_time;
                      }
-                     DownStreamMgr::Instance()->AddPendingAccount(req->account_id, req->socket_handle, req->req_tick);
+                     DownStreamMgr::Instance()->AddPendingAccount
+                         (req->account_id,
+                          req->socket_handle,
+                          req->req_tick);
                  } else if (ALLOC_TARGET_SERVER_SUCCESS_TIMER_EVENT == event) {
                      long long req_handle_time = a8::XGetTickCount() - req->req_tick;
                      if (req_handle_time > App::Instance()->GetPerf().max_login_time) {
                          App::Instance()->GetPerf().max_login_time = req_handle_time;
                      }
-                     DownStreamMgr::Instance()->AddPendingAccount(req->account_id, req->socket_handle, req->req_tick);
+                     DownStreamMgr::Instance()->AddPendingAccount
+                         (req->account_id,
+                          req->socket_handle,
+                          req->req_tick);
                  }
              }
              );
     }
 }
 
+void MasterMgr::OnDownStreamDisconnect(int socket_handle)
+{
+    {
+        auto req = GetHttpTunnelRequestBySocket(socket_handle);
+        if (req) {
+            if (!req->timer_wp.expired()) {
+                f8::Timer::Instance()->Delete(req->timer_wp);
+            }
+        }
+    }
+    RemoveRequest(socket_handle);
+}
+
 void MasterMgr::RemoveRequest(int socket_handle)
 {
-    auto req = GetRequestBySocket(socket_handle);
-    if (req) {
-        if (req->hdr_copy) {
-            f8::MsgHdr::Destroy(req->hdr_copy);
-            req->hdr_copy = nullptr;
+    {
+        auto req = GetRequestBySocket(socket_handle);
+        if (req) {
+            if (req->hdr_copy) {
+                f8::MsgHdr::Destroy(req->hdr_copy);
+                req->hdr_copy = nullptr;
+            }
+            pending_context_hash_.erase(req->context_id);
+            pending_socket_hash_.erase(socket_handle);
         }
-        pending_context_hash_.erase(req->context_id);
-        pending_socket_hash_.erase(socket_handle);
+    }
+    {
+        RemoveHttpTunnelRequest(socket_handle);
     }
 }
 
@@ -199,3 +307,66 @@ std::shared_ptr<RequestTarget> MasterMgr::GetRequestByContextId(long long contex
     auto itr = pending_context_hash_.find(context_id);
     return itr != pending_context_hash_.end() ? itr->second : nullptr;
 }
+
+void MasterMgr::AddHttpTunnelRequest(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request)
+{
+    if (!request->GetParams()->IsObject() ||
+        request->GetParams()->Get("team_uuid", "").GetString() == "") {
+        return;
+    }
+    std::string team_uuid = request->GetParams()->Get("team_uuid", "").GetString();
+    unsigned int code = a8::openssl::Crc32((unsigned char*)team_uuid.data(), team_uuid.size());
+    std::shared_ptr<Master> svr = GetConnById(code % mastersvr_hash_.size() + 1);
+    if (svr) {
+        ++curr_context_id_;
+
+        ss::SS_WSP_HttpTunnelRequest msg;
+        msg.set_context_id(curr_context_id_);
+        msg.set_url(request->GetUrl());
+        std::string query_str;
+        request->GetParams()->ToUrlEncodeStr(query_str);
+        msg.set_query_str(query_str);
+        svr->SendMsg(msg);
+
+        auto req = std::make_shared<HttpTunnelRequest>();
+        req->socket_handle = socket_handle;
+        req->context_id = curr_context_id_;
+        req->request = request;
+
+        pending_http_tunnel_socket_hash_[socket_handle] = req;
+        pending_http_tunnel_context_hash_[curr_context_id_] = req;
+        req->timer_wp = f8::Timer::Instance()->SetTimeoutWp
+            (1000 * 10,
+             [req] (int event, const a8::Args* args)
+             {
+                 if (a8::TIMER_EXEC_EVENT == event) {
+                     req->request->GetResp()->SetVal("errcode", 0);
+                     req->request->GetResp()->SetVal("errmsg", "");
+                     req->request->Response();
+                     MasterMgr::Instance()->RemoveHttpTunnelRequest(req->socket_handle);
+                 }
+             }
+             );
+    }
+}
+
+void MasterMgr::RemoveHttpTunnelRequest(int socket_handle)
+{
+    auto req = GetHttpTunnelRequestBySocket(socket_handle);
+    if (req) {
+        pending_http_tunnel_socket_hash_.erase(req->socket_handle);
+        pending_http_tunnel_context_hash_.erase(req->context_id);
+    }
+}
+
+std::shared_ptr<HttpTunnelRequest> MasterMgr::GetHttpTunnelRequestBySocket(int socket_handle)
+{
+    auto itr = pending_http_tunnel_socket_hash_.find(socket_handle);
+    return itr != pending_http_tunnel_socket_hash_.end() ? itr->second : nullptr;
+}
+
+std::shared_ptr<HttpTunnelRequest> MasterMgr::GetHttpTunnelRequestByContextId(long long context_id)
+{
+    auto itr = pending_http_tunnel_context_hash_.find(context_id);
+    return itr != pending_http_tunnel_context_hash_.end() ? itr->second : nullptr;
+}

server/wsproxy/mastermgr.h

@@ -8,8 +8,11 @@ namespace f8
 namespace ss
 {
     class SS_MS_ResponseTargetServer;
+    class SS_MS_HttpTunnelResponse;
+    class SS_HttpTunnelResponse;
 }
 
+struct HttpTunnelRequest;
 class RequestTarget;
 class Master;
 class MasterMgr : public a8::Singleton<MasterMgr>
@@ -26,25 +29,35 @@ class MasterMgr : public a8::Singleton<MasterMgr>
     void Init();
     void UnInit();
 
-    void _SS_MS_ResponseTargetServer(f8::MsgHdr& hdr, const ss::SS_MS_ResponseTargetServer& msg);
-    void RequestTargetServer(f8::MsgHdr& hdr,
+    void _SS_MS_ResponseTargetServer(f8::MsgHdr* hdr, const ss::SS_MS_ResponseTargetServer& msg);
+    void _SS_MS_HttpTunnelResponse(f8::MsgHdr* hdr, const ss::SS_MS_HttpTunnelResponse& msg);
+    void _SS_HttpTunnelResponse(f8::MsgHdr* hdr, const ss::SS_HttpTunnelResponse& msg);
+    void RequestTargetServer(f8::MsgHdr* hdr,
                              const std::string& team_id,
                              const std::string& account_id,
                              const std::string& session_id,
                              const std::string& server_info,
                              int is_reconnect,
                              int proto_version);
-    void RemoveRequest(int socket_handle);
+    void OnDownStreamDisconnect(int socket_handle);
+    void AddHttpTunnelRequest(int socket_handle, std::shared_ptr<f8::JsonHttpRequest> request);
 
  private:
+    void RemoveRequest(int socket_handle);
+    std::shared_ptr<HttpTunnelRequest> GetHttpTunnelRequestBySocket(int socket_handle);
+    std::shared_ptr<HttpTunnelRequest> GetHttpTunnelRequestByContextId(long long conext_id);
     std::shared_ptr<RequestTarget> GetRequestBySocket(int socket_handle);
     std::shared_ptr<RequestTarget> GetRequestByContextId(long long context_id);
     std::shared_ptr<Master> GetConnById(int instance_id);
+    void RemoveHttpTunnelRequest(int socket_handle);
 
  private:
     long long curr_context_id_ = 0;
     std::map<int, std::shared_ptr<Master>> mastersvr_hash_;
+
     std::map<int, std::shared_ptr<RequestTarget>> pending_socket_hash_;
     std::map<long long, std::shared_ptr<RequestTarget>> pending_context_hash_;
 
+    std::map<int, std::shared_ptr<HttpTunnelRequest>> pending_http_tunnel_socket_hash_;
+    std::map<long long, std::shared_ptr<HttpTunnelRequest>> pending_http_tunnel_context_hash_;
 };

server/wsproxy/upstream.cc

@@ -28,9 +28,7 @@ void UpStream::Init(int instance_id, const std::string& remote_ip, int remote_po
     recv_bufflen_ = 0;
     last_pong_tick = a8::XGetTickCount();
     recv_buff_ = (char*) malloc(PACK_MAX * 2);
-    tcp_client_ = std::make_shared<a8::TcpClient>();
-    tcp_client_->remote_address = remote_ip;
-    tcp_client_->remote_port = remote_port;
+    tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
     tcp_client_->on_error = std::bind(&UpStream::on_error, this, std::placeholders::_1, std::placeholders::_2);
     tcp_client_->on_connect = std::bind(&UpStream::on_connect, this, std::placeholders::_1);
     tcp_client_->on_disconnect = std::bind(&UpStream::on_disconnect, this, std::placeholders::_1);
@@ -101,7 +99,7 @@ void UpStream::SendStockMsg()
             delete pdelnode->msg;
         }
         if (pdelnode->hdr) {
-            ForwardClientMsg(*pdelnode->hdr);
+            ForwardClientMsg(pdelnode->hdr);
             f8::MsgHdr::Destroy(pdelnode->hdr);
             pdelnode->hdr = nullptr;
         }
@@ -109,23 +107,23 @@ void UpStream::SendStockMsg()
     }
 }
 
-void UpStream::ForwardClientMsg(f8::MsgHdr& hdr)
+void UpStream::ForwardClientMsg(f8::MsgHdr* hdr)
 {
-    char* buff = (char*)malloc(sizeof(f8::WSProxyPackHead_C) + hdr.buflen);
+    char* buff = (char*)malloc(sizeof(f8::WSProxyPackHead_C) + hdr->buflen);
     memset(buff, 0, sizeof(f8::WSProxyPackHead_C));
     f8::WSProxyPackHead_C* head = (f8::WSProxyPackHead_C*)buff;
-    head->packlen = hdr.buflen;
-    head->msgid = hdr.msgid;
-    head->seqid = hdr.seqid;
+    head->packlen = hdr->buflen;
+    head->msgid = hdr->msgid;
+    head->seqid = hdr->seqid;
     head->magic_code = f8::MAGIC_CODE;
     #if 0
     head->rpc_error_code = 0;
     #endif
-    head->socket_handle = hdr.socket_handle;
-    head->ip_saddr = hdr.ip_saddr;
+    head->socket_handle = hdr->socket_handle;
+    head->ip_saddr = hdr->ip_saddr;
 
-    if (hdr.buflen > 0) {
-        memmove(buff + sizeof(f8::WSProxyPackHead_C), hdr.buf, hdr.buflen);
+    if (hdr->buflen > 0) {
+        memmove(buff + sizeof(f8::WSProxyPackHead_C), hdr->buf, hdr->buflen);
     }
 
     tcp_client_->SendBuff(buff, sizeof(f8::WSProxyPackHead_C) + head->packlen);
@@ -138,11 +136,7 @@ void UpStream::ForwardClientMsgEx(f8::MsgHdr* hdr)
         if (top_node_) {
             SendStockMsg();
         }
-        ForwardClientMsg(*hdr);
-        if (hdr->buf) {
-            free((char*)hdr->buf);
-        }
-        free(hdr);
+        ForwardClientMsg(hdr);
     } else {
         AddStockMsg(hdr->socket_handle, 0, nullptr, hdr);
     }
@@ -153,8 +147,8 @@ void UpStream::on_error(a8::TcpClient* sender, int errorId)
     f8::UdpLog::Instance()->Error("target server errorid=%d remote_ip:%s remote_port:%d",
                                   {
                                    errorId,
-                                   sender->remote_address,
-                                   sender->remote_port
+                                   sender->GetRemoteAddress(),
+                                   sender->GetRemotePort()
                                   });
 }
 
@@ -163,8 +157,8 @@ void UpStream::on_connect(a8::TcpClient* sender)
     recv_bufflen_ = 0;
     f8::UdpLog::Instance()->Info("target server connected remote_ip:%s remote_port:%d",
                                  {
-                                   sender->remote_address,
-                                   sender->remote_port
+                                  sender->GetRemoteAddress(),
+                                  sender->GetRemotePort()
                                  });
     f8::MsgQueue::Instance()->PostMsg
         (
@@ -185,8 +179,8 @@ void UpStream::on_disconnect(a8::TcpClient* sender)
                                  "remote_ip:%s remote_port:%d",
                                  {
                                   instance_id,
-                                  sender->remote_address,
-                                  sender->remote_port
+                                  sender->GetRemoteAddress(),
+                                  sender->GetRemotePort()
                                  });
     f8::MsgQueue::Instance()->PostMsg
         (
@@ -223,13 +217,15 @@ void UpStream::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
             if (recv_bufflen_ - offset < sizeof(f8::WSProxyPackHead_S) + real_len) {
                 break;
             }
-            App::Instance()->AddSocketMsg(SF_TargetServer,
-                                          p->socket_handle,
-                                          instance_id,
-                                          p->msgid,
-                                          p->seqid,
-                                          &recv_buff_[offset + sizeof(f8::WSProxyPackHead_S)],
-                                          real_len);
+            f8::App::Instance()->AddSocketMsg
+                (SF_TargetServer,
+                 p->socket_handle,
+                 instance_id,
+                 p->msgid,
+                 p->seqid,
+                 &recv_buff_[offset + sizeof(f8::WSProxyPackHead_S)],
+                 real_len,
+                 0);
             offset += sizeof(f8::WSProxyPackHead_S) + real_len;
         } else {
             warning = true;
@@ -266,13 +262,17 @@ void UpStream::CheckAlive()
 }
 
 void UpStream::AddStockMsg(unsigned short socket_handle, int msgid, ::google::protobuf::Message* msg,
-                             f8::MsgHdr* hdr)
+                           f8::MsgHdr* hdr)
 {
     UpStreamMsgNode* node = new UpStreamMsgNode();
     node->socket_handle = socket_handle;
     node->msgid = msgid;
     node->msg = msg;
-    node->hdr = hdr;
+    if (hdr) {
+        node->hdr = hdr->Clone();
+    } else {
+        node->hdr = nullptr;
+    }
     if (bot_node_) {
         bot_node_->next_node = node;
         bot_node_ = node;

server/wsproxy/upstream.h

@@ -50,7 +50,7 @@ class UpStream
     }
 
     void SendStockMsg();
-    void ForwardClientMsg(f8::MsgHdr& hdr);
+    void ForwardClientMsg(f8::MsgHdr* hdr);
     void ForwardClientMsgEx(f8::MsgHdr* hdr);
 
  private:

server/wsproxy/upstreammgr.cc

@@ -4,6 +4,7 @@
 
 #include "upstreammgr.h"
 #include "upstream.h"
+#include "downstreammgr.h"
 
 void UpStreamMgr::Init()
 {
@@ -23,6 +24,8 @@ void UpStreamMgr::Init()
          IM_UpStreamDisconnect,
          [this] (const a8::Args& args)
          {
+             int instance_id = args.Get<int>(0);
+             DownStreamMgr::Instance()->OnUpStreamDisconnect(instance_id);
          });
 }
 

third_party/a8

@@ -1 +1 @@
-Subproject commit 1e577389c8a2870db9ddbf18577bfca24def049b
+Subproject commit c99c33e2e97058bd812dc4d9d931b94757d280bd

third_party/f8

@@ -1 +1 @@
-Subproject commit 243bbe515ef4a01089f9a6cf608c93d4097018de
+Subproject commit 9d627653179015b73f4f21c285b065a50e98a59e