websocket write bug fix (#88)

* added mutex to protect web socket write queue from concurrent access.

Author: RogerZhongAWS

.github/workflows/ci.yml

@@ -19,7 +19,7 @@ jobs:
       name: 'Checkout'
     - name: Install brew dependencies
       run: |
-        brew install openssl zlib cmake wget
+        brew install openssl@1.1 zlib cmake wget
     - name: Install boost
       working-directory: ${{ github.workspace }}
       run: |
@@ -54,7 +54,7 @@ jobs:
       run: |
         mkdir build
         cd build
-        cmake .. -DOPENSSL_ROOT_DIR=/usr/local/Cellar/[email protected]/1.1.1n/ -DOPENSSL_LIBRARIES=/usr/local/Cellar/[email protected]/1.1.1n/lib/
+        cmake .. -DOPENSSL_ROOT_DIR=/usr/local/Cellar/[email protected]/1.1.1q/ -DOPENSSL_LIBRARIES=/usr/local/Cellar/[email protected]/1.1.1q/lib/
         make
   ubuntu:
     runs-on: ubuntu-latest

src/TcpAdapterProxy.cpp

@@ -1536,14 +1536,19 @@ namespace aws { namespace iot { namespace securedtunneling {
             BOOST_LOG_SEV(log, trace) << "Put data " << data_to_send->size() << " bytes into the web_socket_outgoing_message_queue for service id: " << service_id;
             tcp_connection::pointer socket_connection = get_tcp_connection(tac, service_id);
             data_message temp = std::make_pair(data_to_send, socket_connection->after_send_message);
+
+            const std::lock_guard<std::mutex> lock(tac.web_socket_outgoing_message_queue_mutex);
             tac.web_socket_outgoing_message_queue.push(temp);
             // Are we already writing?
             if(tac.web_socket_outgoing_message_queue.size() > 1)
+            {
                 return;
+            }
         }
 
         // We are not currently writing, so send this immediately
         data_message message_to_send = tac.web_socket_outgoing_message_queue.front();
+
         tac.wss->async_write(message_to_send.first->data(), [=, &tac](boost::system::error_code const &ec, std::size_t const bytes_sent)
         {
             if (ec)
@@ -1552,12 +1557,14 @@ namespace aws { namespace iot { namespace securedtunneling {
             }
             BOOST_LOG_SEV(log, trace) << "Sent " << bytes_sent << " bytes over websocket for service id: " << service_id;
             std::function<void()> capture_after_send_message = message_to_send.second;
-            tac.web_socket_outgoing_message_queue.pop();
 
             if(capture_after_send_message)
             {
                 capture_after_send_message();
             }
+
+            const std::lock_guard<std::mutex> lock(tac.web_socket_outgoing_message_queue_mutex);
+            tac.web_socket_outgoing_message_queue.pop();
             if(tac.web_socket_outgoing_message_queue.empty())
             {
                 BOOST_LOG_SEV(log, trace) << "web_socket_outgoing_message_queue is empty, no more messages to send.";

src/TcpAdapterProxy.h

@@ -87,7 +87,8 @@ namespace aws { namespace iot { namespace securedtunneling {
                 bind_address_actual{ },
                 is_web_socket_reading{ false },
                 is_service_ids_received{ false },
-                web_socket_outgoing_message_queue{}
+                web_socket_outgoing_message_queue{},
+                web_socket_outgoing_message_queue_mutex{}
             { }
 
             boost::asio::io_context                                 io_ctx;
@@ -120,6 +121,7 @@ namespace aws { namespace iot { namespace securedtunneling {
             bool                                                    is_web_socket_reading;
             bool                                                    is_service_ids_received;
             std::queue<data_message>                                web_socket_outgoing_message_queue;
+            std::mutex web_socket_outgoing_message_queue_mutex;
         };
 
         //simple re-usable structure for a basic retry strategy's state