1

server/wsproxy/app.cc

@@ -393,8 +393,8 @@ void App::ProcessClientMsg(f8::MsgHdr& hdr)
     } else {
         auto down_wp = DownStreamMgr::Instance()->GetGameClientBySocket(hdr.socket_handle);
         if (auto down = down_wp.lock(); !down_wp.expired()) {
-            if (down->conn) {
-                down->conn->ForwardClientMsg(hdr);
+            if (!down->conn.expired()) {
+                down->conn.lock()->ForwardClientMsg(hdr);
             }
         }
     }

server/wsproxy/downstream.h

@@ -5,6 +5,6 @@ class DownStream
 {
  public:
     int socket_handle = a8::INVALID_SOCKET_HANDLE;
-    UpStream* conn = nullptr;
+    std::weak_ptr<UpStream> conn;
 
 };

server/wsproxy/downstreammgr.cc

@@ -43,9 +43,9 @@ void DownStreamMgr::OnClientDisconnect(int socket_handle)
 {
     auto down_wp = GetGameClientBySocket(socket_handle);
     if (auto down = down_wp.lock(); !down_wp.expired()) {
-        if (down->conn) {
+        if (!down->conn.expired()) {
             ss::SS_WSP_SocketDisconnect msg;
-            down->conn->SendMsg(socket_handle, msg);
+            down->conn.lock()->SendMsg(socket_handle, msg);
         }
         socket_hash_.erase(socket_handle);
     }
@@ -57,7 +57,7 @@ void DownStreamMgr::OnUpStreamDisconnect(int instance_id)
 {
     std::list<std::shared_ptr<DownStream>> delete_client;
     for (auto& pair : socket_hash_) {
-        if (pair.second->conn && pair.second->conn->instance_id == instance_id) {
+        if (!pair.second->conn.expired() && pair.second->conn.lock()->instance_id == instance_id) {
             delete_client.push_back(pair.second);
         }
     }
@@ -81,8 +81,8 @@ std::weak_ptr<DownStream> DownStreamMgr::GetGameClientBySocket(int sockhandle)
 
 void DownStreamMgr::BindUpStream(int socket_handle, int conn_instance_id)
 {
-    UpStream* conn = UpStreamMgr::Instance()->GetConnById(conn_instance_id);
-    if (conn) {
+    std::weak_ptr<UpStream> conn = UpStreamMgr::Instance()->GetConnById(conn_instance_id);
+    if (!conn.expired()) {
         auto down_wp = GetGameClientBySocket(socket_handle);
         if (auto down = down_wp.lock(); !down_wp.expired()) {
             down->conn = conn;

server/wsproxy/mastermgr.cc

@@ -27,7 +27,7 @@ public:
     f8::TimerWp timer_wp;
     long long req_tick = 0;
 
-    UpStream* conn = nullptr;
+    std::weak_ptr<UpStream> conn;
 };
 
 void MasterMgr::Init()
@@ -58,13 +58,13 @@ void MasterMgr::_SS_MS_ResponseTargetServer(f8::MsgHdr& hdr, const ss::SS_MS_Res
     auto req = GetRequestByContextId(msg.context_id());
     if (req) {
         if (msg.error_code() == 0) {
-            UpStream* conn = UpStreamMgr::Instance()->RecreateUpStream
+            std::weak_ptr<UpStream> conn = UpStreamMgr::Instance()->RecreateUpStream
                 (
                  msg.host(),
                  msg.port()
                  );
-            if (conn) {
-                conn->ForwardClientMsgEx(req->hdr_copy);
+            if (!conn.expired()) {
+                conn.lock()->ForwardClientMsgEx(req->hdr_copy);
                 req->conn = conn;
                 req->hdr_copy = nullptr;
                 if (!req->timer_wp.expired()) {

server/wsproxy/upstreammgr.cc

@@ -15,9 +15,9 @@ void UpStreamMgr::Init()
          [this] (const a8::Args& args)
          {
              int instance_id = args.Get<int>(0);
-             UpStream* conn = GetConnById(instance_id);
-             if (conn && conn->Connected()) {
-                 conn->SendStockMsg();
+             std::weak_ptr<UpStream> conn = GetConnById(instance_id);
+             if (!conn.expired() && conn.lock()->Connected()) {
+                 conn.lock()->SendStockMsg();
              }
          });
     f8::MsgQueue::Instance()->RegisterCallBack
@@ -32,35 +32,33 @@ void UpStreamMgr::UnInit()
 {
     for (auto& pair : id_hash_) {
         pair.second->UnInit();
-        delete pair.second;
     }
 }
 
-UpStream* UpStreamMgr::GetConnByKey(const std::string& key)
+std::weak_ptr<UpStream> UpStreamMgr::GetConnByKey(const std::string& key)
 {
     auto itr = key_hash_.find(key);
     return itr != key_hash_.end() ? itr->second : nullptr;
 }
 
-UpStream* UpStreamMgr::GetConnById(int instance_id)
+std::weak_ptr<UpStream> UpStreamMgr::GetConnById(int instance_id)
 {
     auto itr = id_hash_.find(instance_id);
     return itr != id_hash_.end() ? itr->second : nullptr;
 }
 
-UpStream* UpStreamMgr::RecreateUpStream(const std::string& host, int port)
+std::weak_ptr<UpStream> UpStreamMgr::RecreateUpStream(const std::string& host, int port)
 {
     std::string key = host + ":" + a8::XValue(port).GetString();
-    UpStream* conn = GetConnByKey(key);
-    if (conn) {
-        return conn;
+    if (!GetConnByKey(key).expired()) {
+        return GetConnByKey(key);
     }
-    while (GetConnById(++curr_id_)) {};
+    while (GetConnById(++curr_id_).expired()) {};
     int instance_id = curr_id_;
     std::string remote_ip = host;
     int remote_port = port;
 
-    conn = new UpStream();
+    std::shared_ptr<UpStream> conn = std::make_shared<UpStream>();
     conn->Init(instance_id, remote_ip, remote_port);
     id_hash_[conn->instance_id] = conn;
     key_hash_[key] = conn;

server/wsproxy/upstreammgr.h

@@ -14,12 +14,12 @@ class UpStreamMgr : public a8::Singleton<UpStreamMgr>
     void Init();
     void UnInit();
 
-    UpStream* GetConnByKey(const std::string& key);
-    UpStream* GetConnById(int instance_id);
-    UpStream* RecreateUpStream(const std::string& host, int port);
+    std::weak_ptr<UpStream> GetConnByKey(const std::string& key);
+    std::weak_ptr<UpStream> GetConnById(int instance_id);
+    std::weak_ptr<UpStream> RecreateUpStream(const std::string& host, int port);
 
  private:
     unsigned short curr_id_ = 1000;
-    std::map<std::string, UpStream*> key_hash_;
-    std::map<int, UpStream*> id_hash_;
+    std::map<std::string, std::shared_ptr<UpStream>> key_hash_;
+    std::map<int, std::shared_ptr<UpStream>> id_hash_;
 };