update boostssl (#112)

* update boostssl * rename getDesc to detail * optimize wserror output
FISCO-BCOS · Sep 13, 2024 · 6b09c39 · 6b09c39
1 parent c57dbac
commit 6b09c39
Show file tree

Hide file tree

Showing 8 changed files with 109 additions and 31 deletions.
diff --git a/bcos-boostssl/interfaces/NodeInfoDef.h b/bcos-boostssl/interfaces/NodeInfoDef.h
@@ -28,7 +28,6 @@ namespace bcos
 {
 namespace boostssl
 {
-
 /**
  * @brief client end endpoint. Node will connect to NodeIPEndpoint.
  */
@@ -65,6 +64,14 @@ struct NodeIPEndpoint
     {
         return boost::asio::ip::tcp::endpoint(boost::asio::ip::make_address(m_host), m_port);
     }
+    std::string detail() const
+    {
+        if (m_host.empty())
+        {
+            return "";
+        }
+        return m_host + ":" + std::to_string(m_port);
+    }
 
     // Get the port associated with the endpoint.
     uint16_t port() const { return m_port; };

diff --git a/bcos-boostssl/websocket/WsError.h b/bcos-boostssl/websocket/WsError.h
@@ -18,6 +18,9 @@
  * @date 2021-10-02
  */
 #pragma once
+#include <memory>
+#include <sstream>
+#include <string>
 
 namespace bcos
 {
@@ -43,6 +46,60 @@ enum WsError
     MessageEncodeError = -4013
 };
 
+inline std::ostream& operator<<(std::ostream& _out, WsError const& error)
+{
+    switch (error)
+    {
+    case WsError::AcceptError:
+        _out << "AcceptError";
+        break;
+    case WsError::ReadError:
+        _out << "ReadError";
+        break;
+    case WsError::WriteError:
+        _out << "WriteError";
+        break;
+    case WsError::PingError:
+        _out << "PingError";
+        break;
+    case WsError::PongError:
+        _out << "PongError";
+        break;
+    case WsError::PacketError:
+        _out << "PacketError";
+        break;
+    case WsError::SessionDisconnect:
+        _out << "SessionDisconnect";
+        break;
+    case WsError::UserDisconnect:
+        _out << "UserDisconnect";
+        break;
+    case WsError::TimeOut:
+        _out << "TimeOut";
+        break;
+    case WsError::NoActiveCons:
+        _out << "NoActiveCons";
+        break;
+    case WsError::EndPointNotExist:
+        _out << "EndPointNotExist";
+        break;
+    case WsError::MessageOverflow:
+        _out << "MessageOverflow";
+        break;
+    case WsError::UndefinedException:
+        _out << "UndefinedException";
+        break;
+    case WsError::MessageEncodeError:
+        _out << "MessageEncodeError";
+        break;
+    default:
+        _out << "Unkown";
+        break;
+    }
+    return _out;
+}
+
+
 inline bool notRetryAgain(int _wsError)
 {
     return (_wsError == boostssl::ws::WsError::MessageOverflow);

diff --git a/bcos-boostssl/websocket/WsMessage.h b/bcos-boostssl/websocket/WsMessage.h
@@ -46,8 +46,6 @@ namespace boostssl
 {
 namespace ws
 {
-
-
 // the message format for ws protocol
 class WsMessage : public boostssl::MessageFace
 {

diff --git a/bcos-boostssl/websocket/WsService.cpp b/bcos-boostssl/websocket/WsService.cpp
@@ -254,7 +254,7 @@ WsService::asyncConnectToEndpoints(EndPointsPtr _peers)
 
     for (auto& peer : *_peers)
     {
-        std::string connectedEndPoint = peer.address() + ":" + std::to_string(peer.port());
+        std::string connectedEndPoint = peer.detail();
 
         /*
         WEBSOCKET_SERVICE(DEBUG) << LOG_BADGE("asyncConnect")
@@ -271,7 +271,7 @@ WsService::asyncConnectToEndpoints(EndPointsPtr _peers)
 
         auto self = std::weak_ptr<WsService>(shared_from_this());
         m_connector->connectToWsServer(host, port, m_config->disableSsl(),
-            [p, self, connectedEndPoint](boost::beast::error_code _ec,
+            [p, self, connectedEndPoint, peer](boost::beast::error_code _ec,
                 const std::string& _extErrorMsg,
                 std::shared_ptr<WsStreamDelegate> _wsStreamDelegate,
                 std::shared_ptr<std::string> _nodeId) {
@@ -290,7 +290,7 @@ WsService::asyncConnectToEndpoints(EndPointsPtr _peers)
                 }
 
                 auto session = service->newSession(_wsStreamDelegate, *_nodeId.get());
-                session->setEndPoint(connectedEndPoint);
+                session->setEndPoint(peer);
                 session->startAsClient();
             });
     }
@@ -307,7 +307,7 @@ void WsService::reconnect()
     {
         for (auto& peer : *m_reconnectedPeers)
         {
-            std::string connectedEndPoint = peer.address() + ":" + std::to_string(peer.port());
+            std::string connectedEndPoint = peer.detail();
             auto session = getSession(connectedEndPoint);
             if (session)
             {
@@ -321,9 +321,7 @@ void WsService::reconnect()
     {
         for (auto reconnectPeer : *connectPeers)
         {
-            WEBSOCKET_SERVICE(INFO) << ("reconnect")
-                                    << LOG_KV("peer", reconnectPeer.address() + ":" +
-                                                          std::to_string(reconnectPeer.port()));
+            WEBSOCKET_SERVICE(INFO) << ("reconnect") << LOG_KV("peer", reconnectPeer.detail());
         }
         asyncConnectToEndpoints(connectPeers);
     }
@@ -372,14 +370,13 @@ std::shared_ptr<WsSession> WsService::newSession(
 {
     _wsStreamDelegate->setMaxReadMsgSize(m_config->maxMsgSize());
 
-    std::string endPoint = _wsStreamDelegate->remoteEndpoint();
     auto session = m_sessionFactory->createSession(m_moduleName);
 
     session->setWsStreamDelegate(_wsStreamDelegate);
     session->setIoc(m_ioservicePool->getIOService());
     session->setThreadPool(threadPool());
     session->setMessageFactory(messageFactory());
-    session->setEndPoint(endPoint);
+    session->setEndPoint(_wsStreamDelegate->remoteEndpointInfo());
     session->setMaxWriteMsgSize(m_config->maxMsgSize());
     session->setSendMsgTimeout(m_config->sendMsgTimeout());
     session->setNodeId(_nodeId);
@@ -410,7 +407,7 @@ std::shared_ptr<WsSession> WsService::newSession(
         });
 
     WEBSOCKET_SERVICE(INFO) << LOG_BADGE("newSession") << LOG_DESC("start the session")
-                            << LOG_KV("endPoint", endPoint);
+                            << LOG_KV("endPoint", _wsStreamDelegate->remoteEndpoint());
     return session;
 }
 
@@ -459,6 +456,16 @@ std::shared_ptr<WsSession> WsService::getSession(const std::string& _endPoint)
     return nullptr;
 }
 
+bool WsService::isConnected(NodeIPEndpoint const& _nodeIPEndpoint)
+{
+    auto session = getSession(_nodeIPEndpoint.detail());
+    if (session == nullptr || !session->isConnected())
+    {
+        return false;
+    }
+    return true;
+}
+
 WsSessions WsService::sessions()
 {
     WsSessions sessions;
@@ -533,7 +540,7 @@ void WsService::onRecvMessage(
                              << LOG_DESC("receive message from server")
                              << LOG_KV("type", _msg->packetType()) << LOG_KV("seq", seq)
                              << LOG_KV("endpoint", _session->endPoint())
-                             << LOG_KV("data size", _msg->payload()->size())
+                             << LOG_KV("dataSize", _msg->payload() ? _msg->payload()->size() : 0)
                              << LOG_KV("use_count", _session.use_count());
 
     auto typeHandler = getMsgHandler(_msg->packetType());
@@ -725,4 +732,4 @@ void WsService::broadcastMessage(
     }
 
     WEBSOCKET_SERVICE(DEBUG) << LOG_BADGE("broadcastMessage");
-}
+}
diff --git a/bcos-boostssl/websocket/WsService.h b/bcos-boostssl/websocket/WsService.h
@@ -78,6 +78,8 @@ class WsService : public std::enable_shared_from_this<WsService>
     std::string genConnectError(const std::string& _error, const std::string& endpoint, bool end);
     void syncConnectToEndpoints(EndPointsPtr _peers);
 
+    virtual bool isConnected(NodeIPEndpoint const& _nodeIPEndpoint);
+
 public:
     std::shared_ptr<WsSession> newSession(
         std::shared_ptr<WsStreamDelegate> _wsStreamDelegate, std::string const& _nodeId);
@@ -195,7 +197,7 @@ class WsService : public std::enable_shared_from_this<WsService>
         return m_reconnectedPeers;
     }
 
-private:
+protected:
     bool m_running{false};
 
     int32_t m_waitConnectFinishTimeout = 30000;
@@ -208,8 +210,6 @@ class WsService : public std::enable_shared_from_this<WsService>
     // listen host port
     std::string m_listenHost = "";
     uint16_t m_listenPort = 0;
-    // nodeID
-    std::string m_nodeID;
     // Config
     std::shared_ptr<WsConfig> m_config;
 
@@ -228,7 +228,7 @@ class WsService : public std::enable_shared_from_this<WsService>
     // timer
     timer::TimerFactory::Ptr m_timerFactory = nullptr;
 
-private:
+protected:
     // mutex for m_sessions
     mutable boost::shared_mutex x_mutex;
     // all active sessions

diff --git a/bcos-boostssl/websocket/WsSession.cpp b/bcos-boostssl/websocket/WsSession.cpp
@@ -51,14 +51,15 @@ void WsSession::drop(uint32_t _reason)
     {
         WEBSOCKET_SESSION(INFO) << LOG_BADGE("drop")
                                 << LOG_DESC("the session has already been dropped")
-                                << LOG_KV("endpoint", m_endPoint) << LOG_KV("session", this);
+                                << LOG_KV("endpoint", m_endPoint.detail())
+                                << LOG_KV("session", this);
         return;
     }
 
     m_isDrop = true;
 
     WEBSOCKET_SESSION(INFO) << LOG_BADGE("drop") << LOG_KV("reason", _reason)
-                            << LOG_KV("endpoint", m_endPoint) << LOG_KV("session", this);
+                            << LOG_KV("endpoint", m_endPoint.detail()) << LOG_KV("session", this);
 
     auto self = std::weak_ptr<WsSession>(shared_from_this());
     // call callbacks
@@ -71,7 +72,7 @@ void WsSession::drop(uint32_t _reason)
                                 << LOG_KV("cb size", m_callbacks.size()) << LOG_KV("session", this);
 
         Guard lockGuard(x_callback);
-   
+
         for (auto& cbEntry : m_callbacks)
         {
             auto callback = cbEntry.second;

diff --git a/bcos-boostssl/websocket/WsSession.h b/bcos-boostssl/websocket/WsSession.h
@@ -18,6 +18,7 @@
  * @date 2021-07-28
  */
 #pragma once
+#include "../interfaces/NodeInfoDef.h"
 #include "bcos-boostssl/interfaces/MessageFace.h"
 #include <bcos-boostssl/httpserver/Common.h>
 #include <bcos-boostssl/websocket/Common.h>
@@ -82,8 +83,10 @@ class WsSession : public std::enable_shared_from_this<WsSession>
         Options _options = Options(), RespCallBack _respCallback = RespCallBack());
 
 
-    std::string endPoint() const { return m_endPoint; }
-    void setEndPoint(const std::string& _endPoint) { m_endPoint = _endPoint; }
+    std::string endPoint() const { return m_endPoint.detail(); }
+    NodeIPEndpoint const& endPointInfo() const { return m_endPoint; }
+    void setEndPoint(const NodeIPEndpoint& _endPoint) { m_endPoint = _endPoint; }
+
 
     void setConnectHandler(WsConnectHandler _connectHandler) { m_connectHandler = _connectHandler; }
     WsConnectHandler connectHandler() { return m_connectHandler; }
@@ -192,7 +195,7 @@ class WsSession : public std::enable_shared_from_this<WsSession>
     // buffer used to read message
     boost::beast::flat_buffer m_buffer;
 
-    std::string m_endPoint;
+    NodeIPEndpoint m_endPoint;
     std::string m_connectedEndPoint;
     std::string m_nodeId;
 

diff --git a/bcos-boostssl/websocket/WsStream.h b/bcos-boostssl/websocket/WsStream.h
@@ -19,6 +19,7 @@
  */
 #pragma once
 
+#include "../interfaces/NodeInfoDef.h"
 #include <bcos-boostssl/httpserver/Common.h>
 #include <bcos-boostssl/websocket/Common.h>
 #include <bcos-boostssl/websocket/WsTools.h>
@@ -180,22 +181,21 @@ class WsStream
         return std::string("");
     }
 
-    virtual std::string remoteEndpoint()
+    virtual std::string remoteEndpoint() { return remoteEndpointInfo().detail(); }
+
+    virtual NodeIPEndpoint remoteEndpointInfo()
     {
         try
         {
             auto& s = tcpStream();
             auto remoteEndpoint = s.socket().remote_endpoint();
-            auto endPoint =
-                remoteEndpoint.address().to_string() + ":" + std::to_string(remoteEndpoint.port());
-            return endPoint;
+            return NodeIPEndpoint(remoteEndpoint.address(), remoteEndpoint.port());
         }
         catch (const std::exception& e)
         {
             WEBSOCKET_STREAM(WARNING) << LOG_BADGE("remoteEndpoint") << LOG_KV("e", e.what());
         }
-
-        return std::string("");
+        return NodeIPEndpoint();
     }
 
 private:
@@ -234,6 +234,11 @@ class WsStreamDelegate
         return m_isSsl ? m_sslStream->remoteEndpoint() : m_rawStream->remoteEndpoint();
     }
 
+    NodeIPEndpoint remoteEndpointInfo()
+    {
+        return m_isSsl ? m_sslStream->remoteEndpointInfo() : m_rawStream->remoteEndpointInfo();
+    }
+
     void asyncWrite(const bcos::bytes& _buffer, bool _fin, WsStreamRWHandler _handler)
     {
         return m_isSsl ? m_sslStream->asyncWrite(_buffer, _fin, _handler) :
-Original file line number
+Diff line change
@@ Expand Up / @@ -46,8 +46,6 @@ namespace boostssl @@
     {
     namespace ws
     {
     // the message format for ws protocol
     class WsMessage : public boostssl::MessageFace
     {
@@ Expand Down @@