[websockets] Use lws_cancel() to trigger send so that it works event if the lib is compiled againt libuv and libevents

Author: c-jimenez

src/rpc/RpcBase.cpp

@@ -88,7 +88,7 @@ bool RpcBase::call(const std::string&         action,
             std::stringstream expected_id;
             expected_id << m_transaction_id;
             std::shared_ptr<RpcMessage> rpc_message;
-            auto        wait_time   = std::chrono::steady_clock().now() + timeout;
+            auto                        wait_time = std::chrono::steady_clock().now() + timeout;
             do
             {
                 // Compute timeout
@@ -179,8 +179,11 @@ void RpcBase::stop()
     // Check if already started
     if (m_rx_thread)
     {
-        // Stop reception thread
+        // Stop queues
+        m_results_queue.setEnable(false);
         m_requests_queue.setEnable(false);
+
+        // Stop reception thread
         m_rx_thread->join();
         delete m_rx_thread;
         m_rx_thread = nullptr;
@@ -313,7 +316,7 @@ bool RpcBase::decodeCall(const std::string&      unique_id,
     {
         // Add request to the queue
         auto msg = std::make_shared<RpcMessage>(unique_id, action.GetString(), rpc_frame, payload);
-        m_requests_queue.push(msg);
+        m_requests_queue.push(std::move(msg));
 
         ret = true;
     }
@@ -331,7 +334,7 @@ bool RpcBase::decodeCallResult(const std::string& unique_id, rapidjson::Document
     {
         // Add result to the queue
         auto msg = std::make_shared<RpcMessage>(unique_id, rpc_frame, payload);
-        m_results_queue.push(msg);
+        m_results_queue.push(std::move(msg));
 
         ret = true;
     }
@@ -353,7 +356,7 @@ bool RpcBase::decodeCallError(const std::string&   unique_id,
     {
         // Add error to the queue
         auto msg = std::make_shared<RpcMessage>(unique_id, rpc_frame, payload, &error, &message);
-        m_results_queue.push(msg);
+        m_results_queue.push(std::move(msg));
 
         ret = true;
     }

src/websockets/libwebsockets/LibWebsocketClient.cpp

@@ -217,7 +217,7 @@ bool LibWebsocketClient::send(const void* data, size_t size)
         ret          = m_send_msgs.push(msg);
 
         // Schedule a send
-        lws_callback_on_writable(m_wsi);
+        lws_cancel_service(m_context);
     }
 
     return ret;
@@ -372,11 +372,21 @@ int LibWebsocketClient::eventCallback(struct lws* wsi, enum lws_callback_reasons
             client->m_listener->wsClientDataReceived(in, len);
             break;
 
+        case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
+        {
+            // Triggers a send
+            if (!client->m_end && !client->m_send_msgs.empty())
+            {
+                lws_callback_on_writable(client->m_wsi);
+            }
+        }
+        break;
+
         case LWS_CALLBACK_CLIENT_WRITEABLE:
         {
             // Send data if any ready
             bool     error = false;
-            SendMsg* msg = nullptr;
+            SendMsg* msg   = nullptr;
             while (client->m_send_msgs.pop(msg, 0) && !error)
             {
                 if (lws_write(wsi, msg->payload, msg->size, LWS_WRITE_TEXT) < static_cast<int>(msg->size))

src/websockets/libwebsockets/LibWebsocketServer.cpp

@@ -414,6 +414,20 @@ int LibWebsocketServer::eventCallback(struct lws* wsi, enum lws_callback_reasons
         }
         break;
 
+        case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
+        {
+            // Trigger close or send
+            for (const auto& iter_client : server->m_clients)
+            {
+                Client* client = dynamic_cast<Client*>(iter_client.second.get());
+                if (!client->m_connected || !client->m_send_msgs.empty())
+                {
+                    lws_callback_on_writable(client->m_wsi);
+                }
+            }
+        }
+        break;
+
         case LWS_CALLBACK_SERVER_WRITEABLE:
         {
             // Get corresponding client
@@ -425,7 +439,7 @@ int LibWebsocketServer::eventCallback(struct lws* wsi, enum lws_callback_reasons
                 {
                     // Send data if any ready
                     bool     error = false;
-                    SendMsg* msg = nullptr;
+                    SendMsg* msg   = nullptr;
                     while (client->m_send_msgs.pop(msg, 0) && !error)
                     {
                         if (lws_write(client->m_wsi, msg->payload, msg->size, LWS_WRITE_TEXT) < static_cast<int>(msg->size))
@@ -511,7 +525,7 @@ bool LibWebsocketServer::Client::disconnect(bool notify_disconnected)
         }
 
         // Schedule a close
-        lws_callback_on_writable(m_wsi);
+        lws_cancel_service_pt(m_wsi);
     }
 
     // Empty message queue
@@ -543,7 +557,7 @@ bool LibWebsocketServer::Client::send(const void* data, size_t size)
         ret          = m_send_msgs.push(msg);
 
         // Schedule a send
-        lws_callback_on_writable(m_wsi);
+        lws_cancel_service_pt(m_wsi);
     }
 
     return ret;