Merge pull request #152 from c-jimenez/dev/rpc_server_optim

c-jimenez · web-flow · commit d9b7f903f63b · 2023-09-27T10:47:28.000+02:00
[websocket/rpc] Add server callback when connection fails + disable queues on disconnect to speed-up disconnection management
diff --git a/examples/common/DefaultCentralSystemEventsHandler.cpp b/examples/common/DefaultCentralSystemEventsHandler.cpp
@@ -66,6 +66,12 @@ bool DefaultCentralSystemEventsHandler::acceptConnection(const char* ip_address)
     return true;
 }
 
+/** @copydoc void ICentralSystemEventsHandler::clientFailedToConnect(const char*) */
+void DefaultCentralSystemEventsHandler::clientFailedToConnect(const char* ip_address)
+{
+    cout << "Client [" << ip_address << "] failed to connect" << endl;
+}
+
 /** @copydoc bool ICentralSystemEventsHandler::checkCredentials(const std::string&, const std::string&) */
 bool DefaultCentralSystemEventsHandler::checkCredentials(const std::string& chargepoint_id, const std::string& password)
 {
diff --git a/examples/common/DefaultCentralSystemEventsHandler.h b/examples/common/DefaultCentralSystemEventsHandler.h
@@ -48,6 +48,9 @@ class DefaultCentralSystemEventsHandler : public ocpp::centralsystem::ICentralSy
     /** @copydoc bool ICentralSystemEventsHandler::acceptConnection(const char*) */
     bool acceptConnection(const char* ip_address) override;
 
+    /** @copydoc void ICentralSystemEventsHandler::clientFailedToConnect(const char*) */
+    void clientFailedToConnect(const char* ip_address) override;
+
     /** @copydoc bool ICentralSystemEventsHandler::checkCredentials(const std::string&, const std::string&) */
     bool checkCredentials(const std::string& chargepoint_id, const std::string& password) override;
 
diff --git a/examples/common/DefaultLocalControllerEventsHandler.cpp b/examples/common/DefaultLocalControllerEventsHandler.cpp
@@ -48,6 +48,12 @@ bool DefaultLocalControllerEventsHandler::acceptConnection(const char* ip_addres
     return true;
 }
 
+/** @copydoc void ILocalControllerEventsHandler::clientFailedToConnect(const char*) */
+void DefaultLocalControllerEventsHandler::clientFailedToConnect(const char* ip_address)
+{
+    cout << "Client [" << ip_address << "] failed to connect" << endl;
+}
+
 /** @copydoc bool ILocalControllerEventsHandler::checkCredentials(const std::string&, const std::string&) */
 bool DefaultLocalControllerEventsHandler::checkCredentials(const std::string& chargepoint_id, const std::string& password)
 {
diff --git a/examples/common/DefaultLocalControllerEventsHandler.h b/examples/common/DefaultLocalControllerEventsHandler.h
@@ -46,6 +46,9 @@ class DefaultLocalControllerEventsHandler : public ocpp::localcontroller::ILocal
     /** @copydoc bool ILocalControllerEventsHandler::acceptConnection(const char*) */
     bool acceptConnection(const char* ip_address) override;
 
+    /** @copydoc void ILocalControllerEventsHandler::clientFailedToConnect(const char*) */
+    void clientFailedToConnect(const char* ip_address) override;
+
     /** @copydoc bool ILocalControllerEventsHandler::checkCredentials(const std::string&, const std::string&) */
     bool checkCredentials(const std::string& chargepoint_id, const std::string& password) override;
 
diff --git a/src/centralsystem/CentralSystem.cpp b/src/centralsystem/CentralSystem.cpp
@@ -300,6 +300,14 @@ void CentralSystem::rpcClientConnected(const std::string& chargepoint_id, std::s
     m_events_handler.chargePointConnected(chargepoint);
 }
 
+/** @copydoc void RpcServer::IListener::rpcClientFailedToConnect(const char*) */
+void CentralSystem::rpcClientFailedToConnect(const char* ip_address)
+{
+    // Notify failure => no additional processing is done here
+    // to keep this callback has fast as possible
+    return m_events_handler.clientFailedToConnect(ip_address);
+}
+
 /** @copydoc void RpcServer::IListener::rpcServerError() */
 void CentralSystem::rpcServerError()
 {
diff --git a/src/centralsystem/CentralSystem.h b/src/centralsystem/CentralSystem.h
@@ -78,6 +78,9 @@ class CentralSystem : public ICentralSystem, public ocpp::rpc::RpcServer::IListe
     /** @copydoc void RpcServer::IListener::rpcClientConnected(const std::string&, std::shared_ptr<Client>) */
     void rpcClientConnected(const std::string& chargepoint_id, std::shared_ptr<ocpp::rpc::RpcServer::Client> client) override;
 
+    /** @copydoc void RpcServer::IListener::rpcClientFailedToConnect(const char*) */
+    void rpcClientFailedToConnect(const char* ip_address) override;
+
     /** @copydoc void RpcServer::IListener::rpcServerError() */
     void rpcServerError() override;
 
diff --git a/src/centralsystem/interface/ICentralSystemEventsHandler.h b/src/centralsystem/interface/ICentralSystemEventsHandler.h
@@ -40,6 +40,12 @@ class ICentralSystemEventsHandler
      */
     virtual bool acceptConnection(const char* ip_address) = 0;
 
+    /**
+     * @brief Called when connection fails to established
+     * @param ip_address IP address of the client
+     */
+    virtual void clientFailedToConnect(const char* ip_address) = 0;
+
     /**
      * @brief Called to check the charge point credentials for HTTP basic authentication
      * @param chargepoint_id Charge Point identifier
diff --git a/src/localcontroller/LocalController.cpp b/src/localcontroller/LocalController.cpp
@@ -313,6 +313,14 @@ void LocalController::rpcClientConnected(const std::string& chargepoint_id, std:
     m_events_handler.chargePointConnected(chargepoint);
 }
 
+/** @copydoc void RpcServer::IListener::rpcClientFailedToConnect(const char*) */
+void LocalController::rpcClientFailedToConnect(const char* ip_address)
+{
+    // Notify failure => no additional processing is done here
+    // to keep this callback has fast as possible
+    return m_events_handler.clientFailedToConnect(ip_address);
+}
+
 /** @copydoc void RpcServer::IListener::rpcServerError() */
 void LocalController::rpcServerError()
 {
diff --git a/src/localcontroller/LocalController.h b/src/localcontroller/LocalController.h
@@ -78,6 +78,9 @@ class LocalController : public ILocalController, public ocpp::rpc::RpcServer::IL
     /** @copydoc void RpcServer::IListener::rpcClientConnected(const std::string&, std::shared_ptr<Client>) */
     void rpcClientConnected(const std::string& chargepoint_id, std::shared_ptr<ocpp::rpc::RpcServer::Client> client) override;
 
+    /** @copydoc void RpcServer::IListener::rpcClientFailedToConnect(const char*) */
+    void rpcClientFailedToConnect(const char* ip_address) override;
+
     /** @copydoc void RpcServer::IListener::rpcServerError() */
     void rpcServerError() override;
 
diff --git a/src/localcontroller/interface/ILocalControllerEventsHandler.h b/src/localcontroller/interface/ILocalControllerEventsHandler.h
@@ -40,6 +40,12 @@ class ILocalControllerEventsHandler
      */
     virtual bool acceptConnection(const char* ip_address) = 0;
 
+    /**
+     * @brief Called when connection fails to established
+     * @param ip_address IP address of the client
+     */
+    virtual void clientFailedToConnect(const char* ip_address) = 0;
+
     /**
      * @brief Called to check the charge point credentials for HTTP basic authentication
      * @param chargepoint_id Charge Point identifier
diff --git a/src/rpc/RpcBase.cpp b/src/rpc/RpcBase.cpp
@@ -272,6 +272,14 @@ void RpcBase::stop()
     }
 }
 
+/** @brief Process the websocket disconnection event */
+void RpcBase::processDisconnected()
+{
+    // Disable queues
+    m_requests_queue.setEnable(false);
+    m_results_queue.setEnable(false);
+}
+
 /** @brief Process received data */
 void RpcBase::processReceivedData(const void* data, size_t size)
 {
diff --git a/src/rpc/RpcBase.h b/src/rpc/RpcBase.h
@@ -124,6 +124,8 @@ class RpcBase : public IRpc
     void start();
     /** @brief Stop RPC operations */
     void stop();
+    /** @brief Process the websocket disconnection event */
+    void processDisconnected();
     /** @brief Process received data */
     void processReceivedData(const void* data, size_t size);
     /** @brief Get the RPC listener */
diff --git a/src/rpc/RpcServer.cpp b/src/rpc/RpcServer.cpp
@@ -123,6 +123,12 @@ void RpcServer::wsClientConnected(const char* uri, std::shared_ptr<ocpp::websock
     m_listener->rpcClientConnected(chargepoint_id, rpc_client);
 }
 
+/** @copydoc void IWebsocketServer::IListener::wsClientFailedToConnect(const char*) */
+void RpcServer::wsClientFailedToConnect(const char* ip_address)
+{
+    m_listener->rpcClientFailedToConnect(ip_address);
+}
+
 /** @copydoc void IWebsocketServer::IListener::wsServerError() */
 void RpcServer::wsServerError()
 {
@@ -173,6 +179,10 @@ bool RpcServer::Client::isConnected() const
 /** @brief void IWebsocketServer::IClient::IListener::wsClientDisconnected() */
 void RpcServer::Client::wsClientDisconnected()
 {
+    // Process disconnection event
+    processDisconnected();
+
+    // Notify listener
     rpcListener()->rpcDisconnected();
 }
 
diff --git a/src/rpc/RpcServer.h b/src/rpc/RpcServer.h
@@ -78,6 +78,9 @@ class RpcServer : public ocpp::websockets::IWebsocketServer::IListener
     /** @copydoc void IWebsocketServer::IListener::wsClientConnected(const char*, std::shared_ptr<IClient>) */
     void wsClientConnected(const char* uri, std::shared_ptr<ocpp::websockets::IWebsocketServer::IClient> client) override;
 
+    /** @copydoc void IWebsocketServer::IListener::wsClientFailedToConnect(const char*) */
+    void wsClientFailedToConnect(const char* ip_address) override;
+
     /** @copydoc void IWebsocketServer::IListener::wsServerError() */
     void wsServerError() override;
 
@@ -111,6 +114,12 @@ class RpcServer : public ocpp::websockets::IWebsocketServer::IListener
          */
         virtual void rpcClientConnected(const std::string& chargepoint_id, std::shared_ptr<Client> client) = 0;
 
+        /**
+         * @brief Called when connection fails to established
+         * @param ip_address IP address of the client
+         */
+        virtual void rpcClientFailedToConnect(const char* ip_address) = 0;
+
         /** @brief Called on critical error */
         virtual void rpcServerError() = 0;
     };
diff --git a/src/websockets/IWebsocketServer.h b/src/websockets/IWebsocketServer.h
@@ -95,6 +95,12 @@ class IWebsocketServer
          */
         virtual void wsClientConnected(const char* uri, std::shared_ptr<IClient> client) = 0;
 
+        /**
+         * @brief Called when connection fails to established
+         * @param ip_address IP address of the client
+         */
+        virtual void wsClientFailedToConnect(const char* ip_address) = 0;
+
         /** @brief Called on critical error */
         virtual void wsServerError() = 0;
     };
diff --git a/src/websockets/libwebsockets/LibWebsocketServer.cpp b/src/websockets/libwebsockets/LibWebsocketServer.cpp
@@ -45,6 +45,7 @@ LibWebsocketServer::LibWebsocketServer()
       m_wsi(nullptr),
       m_retry_policy(),
       m_protocols(),
+      m_connecting_ip_address(nullptr),
       m_clients()
 {
 }
@@ -278,18 +279,33 @@ int LibWebsocketServer::eventCallback(struct lws* wsi, enum lws_callback_reasons
             struct lws_filter_network_conn_args* filter = reinterpret_cast<struct lws_filter_network_conn_args*>(user);
 
             // Get client IP address
-            char ip_address[64];
-            lws_sa46_write_numeric_address(reinterpret_cast<lws_sockaddr46*>(&filter->cli_addr), ip_address, sizeof(ip_address));
+            // => Save the current IP address so it can be retrieved
+            //    in the LWS_CALLBACK_WSI_CREATE calls which happens
+            //    during the same event loop processing
+            server->m_connecting_ip_address = new char[64u];
+            lws_sa46_write_numeric_address(reinterpret_cast<lws_sockaddr46*>(&filter->cli_addr), server->m_connecting_ip_address, 64u);
 
             // Notify user
-            if (!server->m_listener->wsAcceptConnection(ip_address))
+            if (!server->m_listener->wsAcceptConnection(server->m_connecting_ip_address))
             {
                 // Disconnect
                 ret = -1;
+
+                // Release memory
+                delete[] server->m_connecting_ip_address;
             }
         }
         break;
 
+        case LWS_CALLBACK_WSI_CREATE:
+        {
+            // Set client IP address
+            // => Must be done here to ensure that the event loop is still working
+            //    with the same client as in the LWS_CALLBACK_FILTER_NETWORK_CONNECTION call
+            lws_set_wsi_user(wsi, server->m_connecting_ip_address);
+        }
+        break;
+
         case LWS_CALLBACK_HTTP_CONFIRM_UPGRADE:
         {
             // Check selected protocol
@@ -418,8 +434,7 @@ int LibWebsocketServer::eventCallback(struct lws* wsi, enum lws_callback_reasons
         case LWS_CALLBACK_ESTABLISHED:
         {
             // Get client IP address
-            char ip_address[64];
-            lws_get_peer_simple(wsi, ip_address, sizeof(ip_address));
+            char* ip_address = reinterpret_cast<char*>(lws_wsi_user(wsi));
 
             // Instanciate a new client
             std::shared_ptr<IClient> client(new Client(wsi, ip_address));
@@ -455,10 +470,30 @@ int LibWebsocketServer::eventCallback(struct lws* wsi, enum lws_callback_reasons
                 {
                     client->m_listener->wsClientDisconnected();
                 }
+            }
+        }
+        break;
 
+        case LWS_CALLBACK_WSI_DESTROY:
+        {
+            // Get client IP address
+            char* ip_address = reinterpret_cast<char*>(lws_wsi_user(wsi));
+
+            // Get corresponding client
+            auto iter_client = server->m_clients.find(wsi);
+            if (iter_client != server->m_clients.end())
+            {
                 // Remove client
                 server->m_clients.erase(iter_client);
             }
+            else
+            {
+                // Connection failed to be established
+                server->m_listener->wsClientFailedToConnect(ip_address);
+            }
+
+            // Release memory
+            delete[] ip_address;
         }
         break;
 
diff --git a/src/websockets/libwebsockets/LibWebsocketServer.h b/src/websockets/libwebsockets/LibWebsocketServer.h
@@ -180,6 +180,8 @@ class LibWebsocketServer : public IWebsocketServer
     lws_retry_bo_t m_retry_policy;
     /** @brief Protocols */
     std::array<struct lws_protocols, 2u> m_protocols;
+    /** @brief IP address of the currently connecting client */
+    char* m_connecting_ip_address;
 
     /** @brief Connected clients */
     std::unordered_map<struct lws*, std::shared_ptr<IClient>> m_clients;

Original file line number	Diff line number	Diff line change
`@@ -66,6 +66,12 @@ bool DefaultCentralSystemEventsHandler::acceptConnection(const char* ip_address)`
`66`	`66`	`return true;`
`67`	`67`	`}`
`68`	`68`
	`69`	`+/** @copydoc void ICentralSystemEventsHandler::clientFailedToConnect(const char) /`
	`70`	`+void DefaultCentralSystemEventsHandler::clientFailedToConnect(const char* ip_address)`
	`71`	`+{`
	`72`	`+ cout << "Client [" << ip_address << "] failed to connect" << endl;`
	`73`	`+}`
	`74`	`+`
`69`	`75`	`/** @copydoc bool ICentralSystemEventsHandler::checkCredentials(const std::string&, const std::string&) */`
`70`	`76`	`bool DefaultCentralSystemEventsHandler::checkCredentials(const std::string& chargepoint_id, const std::string& password)`
`71`	`77`	`{`
Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,12 @@ bool DefaultLocalControllerEventsHandler::acceptConnection(const char* ip_addres`
`48`	`48`	`return true;`
`49`	`49`	`}`
`50`	`50`
	`51`	`+/** @copydoc void ILocalControllerEventsHandler::clientFailedToConnect(const char) /`
	`52`	`+void DefaultLocalControllerEventsHandler::clientFailedToConnect(const char* ip_address)`
	`53`	`+{`
	`54`	`+ cout << "Client [" << ip_address << "] failed to connect" << endl;`
	`55`	`+}`
	`56`	`+`
`51`	`57`	`/** @copydoc bool ILocalControllerEventsHandler::checkCredentials(const std::string&, const std::string&) */`
`52`	`58`	`bool DefaultLocalControllerEventsHandler::checkCredentials(const std::string& chargepoint_id, const std::string& password)`
`53`	`59`	`{`
Original file line number	Diff line number	Diff line change
`@@ -300,6 +300,14 @@ void CentralSystem::rpcClientConnected(const std::string& chargepoint_id, std::s`
`300`	`300`	`m_events_handler.chargePointConnected(chargepoint);`
`301`	`301`	`}`
`302`	`302`
	`303`	`+/** @copydoc void RpcServer::IListener::rpcClientFailedToConnect(const char) /`
	`304`	`+void CentralSystem::rpcClientFailedToConnect(const char* ip_address)`
	`305`	`+{`
	`306`	`+ // Notify failure => no additional processing is done here`
	`307`	`+ // to keep this callback has fast as possible`
	`308`	`+ return m_events_handler.clientFailedToConnect(ip_address);`
	`309`	`+}`
	`310`	`+`
`303`	`311`	`/** @copydoc void RpcServer::IListener::rpcServerError() */`
`304`	`312`	`void CentralSystem::rpcServerError()`
`305`	`313`	`{`
Original file line number	Diff line number	Diff line change
`@@ -313,6 +313,14 @@ void LocalController::rpcClientConnected(const std::string& chargepoint_id, std:`
`313`	`313`	`m_events_handler.chargePointConnected(chargepoint);`
`314`	`314`	`}`
`315`	`315`
	`316`	`+/** @copydoc void RpcServer::IListener::rpcClientFailedToConnect(const char) /`
	`317`	`+void LocalController::rpcClientFailedToConnect(const char* ip_address)`
	`318`	`+{`
	`319`	`+ // Notify failure => no additional processing is done here`
	`320`	`+ // to keep this callback has fast as possible`
	`321`	`+ return m_events_handler.clientFailedToConnect(ip_address);`
	`322`	`+}`
	`323`	`+`
`316`	`324`	`/** @copydoc void RpcServer::IListener::rpcServerError() */`
`317`	`325`	`void LocalController::rpcServerError()`
`318`	`326`	`{`