[Optimize]Change the way websocket connection works

1. use work_thread_ to handle websocket_client main task rather than manually lock() and unlock().
2. use SocketGuard to manage socket_fd_.

issue:#23

Author: RovicnowYoyi

debug_router/native/base/socket_guard.h

@@ -0,0 +1,47 @@
+// Copyright 2025 The Lynx Authors. All rights reserved.
+// Licensed under the Apache License Version 2.0 that can be found in the
+// LICENSE file in the root directory of this source tree.
+
+#ifndef DEBUGROUTER_NATIVE_BASE_SOCKET_UTIL_H_
+#define DEBUGROUTER_NATIVE_BASE_SOCKET_UTIL_H_
+
+#if defined(_WIN32)
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#define CLOSESOCKET closesocket
+#else
+#include <netdb.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+#define SOCKET int
+#define CLOSESOCKET close
+#endif
+
+constexpr SOCKET kInvalidSocket = 0;
+
+namespace debugrouter {
+namespace base {
+
+class SocketGuard {
+ public:
+  SOCKET Get() const { return sock_; }
+
+  explicit SocketGuard(SOCKET sock) : sock_(sock) {}
+
+  ~SocketGuard() {
+    if (sock_ != 0) {
+      CLOSESOCKET(sock_);
+    }
+  }
+  SocketGuard(const SocketGuard&) = delete;
+  SocketGuard& operator=(const SocketGuard&) = delete;
+
+ private:
+  SOCKET sock_;
+};
+
+}  // namespace base
+}  // namespace debugrouter
+
+#endif  // DEBUGROUTER_NATIVE_BASE_SOCKET_UTIL_H_

debug_router/native/net/websocket_client.cc

@@ -99,43 +99,53 @@ std::string decodeURIComponent(std::string url) {
   return result_url_.str();
 }
 
-WebSocketClient::WebSocketClient() {}
+WebSocketClient::WebSocketClient() : socket_guard_(nullptr) {}
 
-WebSocketClient::~WebSocketClient() { Disconnect(); }
+WebSocketClient::~WebSocketClient() { DisconnectInternal(); }
 
 void WebSocketClient::Init() {}
 
 bool WebSocketClient::Connect(const std::string &url) {
-  Disconnect();
+  auto self = std::static_pointer_cast<WebSocketClient>(shared_from_this());
+  work_thread_.submit([client_ptr = self, url]() {
+    client_ptr->DisconnectInternal();
+    client_ptr->ConnectInternal(url);
+  });
+  return true;
+}
 
-  mutex_.lock();
+void WebSocketClient::ConnectInternal(const std::string &url) {
   url_ = url;
   thread_ = std::make_unique<std::thread>([this]() { run(); });
-  mutex_.unlock();
-  return true;
 }
 
 void WebSocketClient::Disconnect() {
-  mutex_.lock();
-  if (socket_) {
-    CLOSESOCKET(socket_);
-    socket_ = 0;
-  }
+  auto self = std::static_pointer_cast<WebSocketClient>(shared_from_this());
+  work_thread_.submit(
+      [client_ptr = self]() { client_ptr->DisconnectInternal(); });
+}
+
+void WebSocketClient::DisconnectInternal() {
   if (thread_) {
     if (thread_->joinable()) {
       thread_->join();
       LOGI("WebSocketClient thread exit successfully.");
     }
     thread_.reset();
   }
-  mutex_.unlock();
 }
 
 core::ConnectionType WebSocketClient::GetType() {
   return core::ConnectionType::kWebSocket;
 }
 
 void WebSocketClient::Send(const std::string &data) {
+  auto self = std::static_pointer_cast<WebSocketClient>(shared_from_this());
+  work_thread_.submit(
+      [client_ptr = self, data]() { client_ptr->SendInternal(data); });
+}
+
+void WebSocketClient::SendInternal(const std::string &data) {
   const char *buf = data.data();
   size_t payloadLen = data.size();
   uint8_t prefix[14];
@@ -165,10 +175,8 @@ void WebSocketClient::Send(const std::string &data) {
   *reinterpret_cast<uint32_t *>(prefix + prefix_len) = 0;
   prefix_len += 4;
 
-  mutex_.lock();
-  send(socket_, (char *)prefix, prefix_len, 0);
-  send(socket_, buf, payloadLen, 0);
-  mutex_.unlock();
+  send(socket_guard_->Get(), (char *)prefix, prefix_len, 0);
+  send(socket_guard_->Get(), buf, payloadLen, 0);
 }
 
 void WebSocketClient::run() {
@@ -234,7 +242,7 @@ bool WebSocketClient::do_connect() {
       continue;
     }
     if (connect(sockfd, p->ai_addr, p->ai_addrlen) != -1) {
-      socket_ = sockfd;
+      socket_guard_ = std::make_unique<base::SocketGuard>(sockfd);
       break;
     }
     CLOSESOCKET(sockfd);
@@ -250,17 +258,18 @@ bool WebSocketClient::do_connect() {
            "Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw==\r\n"
            "Sec-WebSocket-Version: 13\r\n\r\n",
            path, host, port);
-  send(socket_, buf, strlen(buf), 0);
+  send(socket_guard_->Get(), buf, strlen(buf), 0);
 
   int status;
-  if (readline(socket_, buf, sizeof(buf)) < 10 ||
+  if (readline(socket_guard_->Get(), buf, sizeof(buf)) < 10 ||
       sscanf(buf, "HTTP/1.1 %d Switching Protocols\r\n", &status) != 1 ||
       status != 101) {
     LOGE("Connect Error: " << url_.c_str());
     return false;
   }
 
-  while (readline(socket_, buf, sizeof(buf)) > 0 && buf[0] != '\r') {
+  while (readline(socket_guard_->Get(), buf, sizeof(buf)) > 0 &&
+         buf[0] != '\r') {
     size_t len = strlen(buf);
     buf[len - 2] = '\0';
     LOGI(buf);
@@ -275,7 +284,8 @@ bool WebSocketClient::do_read(std::string &msg) {
   } head;
   auto self = std::static_pointer_cast<WebSocketClient>(shared_from_this());
 
-  if (recv(socket_, (char *)&head, sizeof(head), 0) != sizeof(head)) {
+  if (recv(socket_guard_->Get(), (char *)&head, sizeof(head), 0) !=
+      sizeof(head)) {
     LOGE("failed to read websocket message");
     delegate()->OnFailure(self);
     return false;
@@ -301,18 +311,18 @@ bool WebSocketClient::do_read(std::string &msg) {
 
   if (payloadLen == 126) {
     uint8_t len[2];
-    recv(socket_, (char *)&len, sizeof(len), 0);
+    recv(socket_guard_->Get(), (char *)&len, sizeof(len), 0);
     payloadLen = (len[0] << 8) | len[1];
   } else if (payloadLen == 127) {
     uint8_t len[8];
-    recv(socket_, (char *)&len, sizeof(len), 0);
+    recv(socket_guard_->Get(), (char *)&len, sizeof(len), 0);
     payloadLen = (len[4] << 24) | (len[5] << 16) | (len[6] << 8) | len[7];
   }
 
   msg.resize(payloadLen);
 
-  if (recv(socket_, const_cast<char *>(msg.data()), payloadLen, 0) !=
-      payloadLen) {
+  if (recv(socket_guard_->Get(), const_cast<char *>(msg.data()), payloadLen,
+           0) != payloadLen) {
     LOGE("failed to read websocket message");
     delegate()->OnFailure(self);
     return false;

debug_router/native/net/websocket_client.h

@@ -10,22 +10,26 @@
 #include <string>
 #include <thread>
 
+#include "debug_router/native/base/socket_guard.h"
 #include "debug_router/native/core/message_transceiver.h"
+#include "debug_router/native/socket/work_thread_executor.h"
 
 #if defined(_WIN32)
 #include <winsock2.h>
 #include <ws2tcpip.h>
-#define CLOSESOCKET closesocket
 #else
 #include <netdb.h>
 #include <sys/socket.h>
 #include <sys/types.h>
 #include <unistd.h>
 #define SOCKET int
-#define CLOSESOCKET close
 #endif
 
 namespace debugrouter {
+namespace base {
+class WorkThreadExecutor;
+class SocketGuard;
+}  // namespace base
 namespace net {
 
 class WebSocketClient : public core::MessageTransceiver {
@@ -40,14 +44,18 @@ class WebSocketClient : public core::MessageTransceiver {
   core::ConnectionType GetType() override;
 
  private:
+  void DisconnectInternal();
+  void ConnectInternal(const std::string &url);
+  void SendInternal(const std::string &data);
+
   void run();
   bool do_connect();
   bool do_read(std::string &msg);
 
   std::string url_;
-  SOCKET socket_ = 0;
+  std::unique_ptr<base::SocketGuard> socket_guard_;
+  base::WorkThreadExecutor work_thread_;
   std::unique_ptr<std::thread> thread_;
-  std::mutex mutex_;
 };
 }  // namespace net
 }  // namespace debugrouter