[websocket] Create websocket client pool to limit number of threads when creating websocket clients

examples/quick_start_cs_lc_hybrid/main.cpp

@@ -25,6 +25,7 @@ SOFTWARE.
 #include "HybridCentralSystemEventsHandler.h"
 #include "ICentralSystem.h"
 #include "LocalControllerDemoConfig.h"
+#include "WebsocketFactory.h"
 
 #include <cstring>
 #include <filesystem>
@@ -95,6 +96,10 @@ int main(int argc, char* argv[])
     path /= "quick_start_cs_lc_hybrid.ini";
     LocalControllerDemoConfig config(path.string());
 
+    // Configure websocket pools => mandatory for local controller
+    ocpp::websockets::WebsocketFactory::setClientPoolCount(2u);
+    ocpp::websockets::WebsocketFactory::startClientPools();
+
     // Event handler
     HybridCentralSystemEventsHandler event_handler(config.stackConfig());
 

examples/quick_start_localcontroller/main.cpp

@@ -25,6 +25,7 @@ SOFTWARE.
 #include "DefaultLocalControllerEventsHandler.h"
 #include "ILocalController.h"
 #include "LocalControllerDemoConfig.h"
+#include "WebsocketFactory.h"
 
 #include <cstring>
 #include <filesystem>
@@ -98,6 +99,10 @@ int main(int argc, char* argv[])
     // Event handler
     DefaultLocalControllerEventsHandler event_handler(config.stackConfig());
 
+    // Configure websocket pools => mandatory for local controller
+    ocpp::websockets::WebsocketFactory::setClientPoolCount(2u);
+    ocpp::websockets::WebsocketFactory::startClientPools();
+
     // Instanciate local controller
     std::unique_ptr<ILocalController> local_controller = ILocalController::create(config.stackConfig(), event_handler);
     if (reset_all)

src/chargepoint/metervalues/MeterValuesManager.cpp

@@ -412,7 +412,7 @@ void MeterValuesManager::processTriggered(unsigned int connector_id)
             auto         measurands          = computeMeasurandList(meter_values, measurands_max_size);
             if (!measurands.empty())
             {
-                LOG_INFO << "Triggered mater values : " << meter_values;
+                LOG_INFO << "Triggered meter values : " << meter_values;
 
                 // Get connector
                 Connector* connector = m_connectors.getConnector(connector_id);

src/localcontroller/centralsystem/CentralSystemProxy.cpp

@@ -37,7 +37,7 @@ CentralSystemProxy::CentralSystemProxy(const std::string&
                                        const ocpp::config::ILocalControllerConfig& stack_config)
     : m_identifier(identifier),
       m_stack_config(stack_config),
-      m_websocket(ocpp::websockets::WebsocketFactory::newClient()),
+      m_websocket(ocpp::websockets::WebsocketFactory::newClientFromPool()),
       m_rpc(*m_websocket, "ocpp1.6"),
       m_messages_converter(messages_converter),
       m_msg_dispatcher(messages_validator),

src/messages/GetCompositeSchedule.cpp

@@ -78,7 +78,7 @@ bool GetCompositeScheduleConfConverter::fromJson(const rapidjson::Value&   json,
     if (json.HasMember("chargingSchedule"))
     {
         ChargingScheduleConverter charging_schedule_converter;
-        ret = charging_schedule_converter.fromJson(json[""], data.chargingSchedule.value(), error_code, error_message);
+        ret = charging_schedule_converter.fromJson(json["chargingSchedule"], data.chargingSchedule.value(), error_code, error_message);
     }
     if (!ret && error_code.empty())
     {

src/tools/helpers/Queue.h

@@ -160,7 +160,7 @@ class Queue
             // Retrieve item
             if (m_enabled)
             {
-                item = m_queue.front();
+                item = std::move(m_queue.front());
                 m_queue.pop();
                 ret = true;
             }

src/websockets/CMakeLists.txt

@@ -9,6 +9,7 @@ add_library(ws OBJECT
     Url.cpp
     WebsocketFactory.cpp
     libwebsockets/LibWebsocketClient.cpp
+    libwebsockets/LibWebsocketClientPool.cpp
     libwebsockets/LibWebsocketServer.cpp
  )
 

src/websockets/WebsocketFactory.cpp

@@ -18,17 +18,57 @@ along with OpenOCPP. If not, see <http://www.gnu.org/licenses/>.
 
 #include "WebsocketFactory.h"
 #include "LibWebsocketClient.h"
+#include "LibWebsocketClientPool.h"
 #include "LibWebsocketServer.h"
 
+#include <memory>
+#include <mutex>
+#include <vector>
+
 namespace ocpp
 {
 namespace websockets
 {
 
+/** @brief Mutex to protect the access to the client pools */
+static std::mutex s_client_pools_mutex;
+/** @brief Client pools */
+static std::vector<std::unique_ptr<LibWebsocketClientPool>> s_client_pools;
+/** @brief Indicate if the standard clients must be allocated from the pools */
+static bool s_force_clients_from_pool = false;
+
 /** @brief Instanciate a client websocket */
 IWebsocketClient* WebsocketFactory::newClient()
 {
-    return new LibWebsocketClient();
+    if (s_force_clients_from_pool)
+    {
+        return newClientFromPool();
+    }
+    else
+    {
+        return new LibWebsocketClient();
+    }
+}
+
+/** @brief Instanciate a client websocket from the pool (the pool must be started first) */
+IWebsocketClient* WebsocketFactory::newClientFromPool()
+{
+    IWebsocketClient*           ret = nullptr;
+    std::lock_guard<std::mutex> lock(s_client_pools_mutex);
+    if (!s_client_pools.empty())
+    {
+        // Look for the pool with the less associated clients
+        auto* selected_pool = &s_client_pools[0];
+        for (auto& pool : s_client_pools)
+        {
+            if (pool->getClientsCount() < (*selected_pool)->getClientsCount())
+            {
+                selected_pool = &pool;
+            }
+        }
+        ret = (*selected_pool)->newClient();
+    }
+    return ret;
 }
 
 /** @brief Instanciate a server websocket */
@@ -37,5 +77,63 @@ IWebsocketServer* WebsocketFactory::newServer()
     return new LibWebsocketServer();
 }
 
+/** @brief Set the number of client pools (can only be done once) */
+bool WebsocketFactory::setClientPoolCount(size_t count)
+{
+    bool ret = false;
+
+    std::lock_guard<std::mutex> lock(s_client_pools_mutex);
+    if (s_client_pools.empty() && (count > 0))
+    {
+        for (size_t i = 0; i < count; i++)
+        {
+            s_client_pools.emplace_back(std::make_unique<LibWebsocketClientPool>());
+        }
+        ret = true;
+    }
+
+    return ret;
+}
+
+/** @brief Start the client pools */
+bool WebsocketFactory::startClientPools()
+{
+    bool ret = false;
+
+    std::lock_guard<std::mutex> lock(s_client_pools_mutex);
+    if (!s_client_pools.empty())
+    {
+        for (auto& pool : s_client_pools)
+        {
+            ret = pool->start() && ret;
+        }
+    }
+
+    return ret;
+}
+
+/** @brief Stop the client pools */
+bool WebsocketFactory::stopClientPools()
+{
+    bool ret = false;
+
+    std::lock_guard<std::mutex> lock(s_client_pools_mutex);
+    if (!s_client_pools.empty())
+    {
+        for (auto& pool : s_client_pools)
+        {
+            ret = pool->stop() && ret;
+        }
+    }
+
+    return ret;
+}
+
+/** @brief Indicate to use the client pools even for new clients instanciated with the newClient API */
+void WebsocketFactory::forceClientPoolsUsage()
+{
+    s_force_clients_from_pool = true;
+}
+
 } // namespace websockets
 } // namespace ocpp

src/websockets/WebsocketFactory.h

@@ -33,8 +33,19 @@ class WebsocketFactory
   public:
     /** @brief Instanciate a client websocket */
     static IWebsocketClient* newClient();
+    /** @brief Instanciate a client websocket from the pool (the pool must be started first) */
+    static IWebsocketClient* newClientFromPool();
     /** @brief Instanciate a server websocket */
     static IWebsocketServer* newServer();
+
+    /** @brief Set the number of client pools (can only be done once) */
+    static bool setClientPoolCount(size_t count);
+    /** @brief Start the client pools */
+    static bool startClientPools();
+    /** @brief Stop the client pools (all client communications must be terminated first)*/
+    static bool stopClientPools();
+    /** @brief Indicate to use the client pools even for new clients instanciated with the newClient API */
+    static void forceClientPoolsUsage();
 };
 
 } // namespace websockets

src/websockets/libwebsockets/LibWebsocketClient.cpp

@@ -45,6 +45,7 @@ LibWebsocketClient::LibWebsocketClient()
       m_credentials(),
       m_connected(false),
       m_context(nullptr),
+      m_logs_context(),
       m_sched_list(),
       m_wsi(nullptr),
       m_retry_policy(),
@@ -85,13 +86,19 @@ bool LibWebsocketClient::connect(const std::string&        url,
             static const struct lws_protocols protocols[] = {
                 {"LibWebsocketClient", &LibWebsocketClient::eventCallback, 0, 0, 0, nullptr, 0}, {nullptr, nullptr, 0, 0, 0, nullptr, 0}};
 
+            // Initialize log context
+            memset(&m_logs_context, 0, sizeof(m_logs_context));
+            m_logs_context.u.emit    = LIBWEBSOCKET_LOG_OUTPUT_FN;
+            m_logs_context.lll_flags = LIBWEBSOCKET_LOG_FLAGS;
+
             // Fill context information
             struct lws_context_creation_info info;
             memset(&info, 0, sizeof info);
             info.options      = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
             info.port         = CONTEXT_PORT_NO_LISTEN;
             info.protocols    = protocols;
             info.timeout_secs = static_cast<unsigned int>(std::chrono::duration_cast<std::chrono::seconds>(connect_timeout).count());
+            info.log_cx       = &m_logs_context;
             m_credentials     = credentials;
             if (m_url.protocol() == "wss")
             {

src/websockets/libwebsockets/LibWebsocketClient.h

@@ -109,6 +109,8 @@ class LibWebsocketClient : public IWebsocketClient
 
     /** @brief Websocket context */
     struct lws_context* m_context;
+    /** @brief Websocket log context */
+    lws_log_cx_t m_logs_context;
     /** @brief Schedule list */
     lws_sorted_usec_list_t m_sched_list;
     /** @brief Related wsi */