Discard stale WebSocket state when reconnecting

ixwebsocket/IXWebSocket.cpp

@@ -210,6 +210,12 @@ namespace ix
 
     WebSocketInitResult WebSocket::connect(int timeoutSecs)
     {
+        // Finish any in-flight send before discarding the previous connection's state.
+        std::unique_lock<std::mutex> writeLock(_writeMutex, std::defer_lock);
+        if (!waitForWriteLock(writeLock))
+        {
+            return WebSocketInitResult(false, 0, "Connection cancelled while waiting for a send");
+        }
         {
             std::lock_guard<std::mutex> lock(_configMutex);
             _ws.configure(
@@ -240,6 +246,7 @@ namespace ix
         }
 
         WebSocketInitResult status = _ws.connectToUrl(_url, headers, timeoutSecs);
+        writeLock.unlock();
         if (!status.success)
         {
             return status;
@@ -268,6 +275,11 @@ namespace ix
                                                    HttpRequestPtr request,
                                                    int sendTimeoutSecs)
     {
+        std::unique_lock<std::mutex> writeLock(_writeMutex, std::defer_lock);
+        if (!waitForWriteLock(writeLock))
+        {
+            return WebSocketInitResult(false, 0, "Connection cancelled while waiting for a send");
+        }
         {
             std::lock_guard<std::mutex> lock(_configMutex);
             _ws.configure(
@@ -276,6 +288,7 @@ namespace ix
 
         WebSocketInitResult status = _ws.connectToSocket(
             std::move(socket), timeoutSecs, enablePerMessageDeflate, request, sendTimeoutSecs);
+        writeLock.unlock();
         if (!status.success)
         {
             return status;
@@ -298,6 +311,22 @@ namespace ix
         return status;
     }
 
+    bool WebSocket::waitForWriteLock(std::unique_lock<std::mutex>& lock)
+    {
+        while (!lock.try_lock())
+        {
+            // A progress callback can call stop() while holding _writeMutex. Let the
+            // background thread exit so stop() can join it without waiting for that send.
+            std::unique_lock<std::mutex> sleepLock(_sleepMutex);
+            if (_sleepCondition.wait_for(
+                    sleepLock, std::chrono::milliseconds(10), [this] { return _stop.load(); }))
+            {
+                return false;
+            }
+        }
+        return true;
+    }
+
     bool WebSocket::isConnected() const
     {
         return getReadyState() == ReadyState::Open;

ixwebsocket/IXWebSocket.h

@@ -129,6 +129,7 @@ namespace ix
         bool isConnected() const;
         bool isClosing() const;
         void checkConnection(bool firstConnectionAttempt);
+        bool waitForWriteLock(std::unique_lock<std::mutex>& lock);
         static void invokeTrafficTrackerCallback(size_t size, bool incoming);
 
         // Server

ixwebsocket/IXWebSocketTransport.cpp

@@ -106,6 +106,7 @@ namespace ix
                                                          const WebSocketHttpHeaders& headers,
                                                          int timeoutSecs)
     {
+        resetConnectionState();
         std::lock_guard<std::mutex> lock(_socketMutex);
 
         std::string protocol, host, path, query;
@@ -176,6 +177,7 @@ namespace ix
                                                             HttpRequestPtr request,
                                                             int sendTimeoutSecs)
     {
+        resetConnectionState();
         std::lock_guard<std::mutex> lock(_socketMutex);
 
         // Server should not mask the data it sends to the client
@@ -401,6 +403,19 @@ namespace ix
         return PollResult::Succeeded;
     }
 
+    void WebSocketTransport::resetConnectionState()
+    {
+        // Frames and fragments belong to one connection and cannot be replayed on another.
+        // Take this lock before _socketMutex, matching sendOnSocket()'s lock order.
+        std::lock_guard<std::mutex> lock(_txbufMutex);
+        _txbuf.clear();
+        _rxbuf.clear();
+        _rxbufWanted = 0;
+        _chunks.clear();
+        _fragmentedMessageKind = MessageKind::MSG_TEXT;
+        _receivedMessageCompressed = false;
+    }
+
     bool WebSocketTransport::isSendBufferEmpty() const
     {
         std::lock_guard<std::mutex> lock(_txbufMutex);

ixwebsocket/IXWebSocketTransport.h

@@ -279,6 +279,7 @@ namespace ix
                          const OnMessageCallback& onMessageCallback);
 
         bool isSendBufferEmpty() const;
+        void resetConnectionState();
 
         template<class Iterator>
         void appendToSendBuffer(const std::vector<uint8_t>& header,

test/CMakeLists.txt

@@ -15,6 +15,7 @@ set (TEST_TARGET_NAMES
   IXSocketConnectTest
   IXWebSocketServerTest
   IXWebSocketTestConnectionDisconnection
+  IXWebSocketReconnectTest
   IXUrlParserTest
   # IXHttpClientTest ## FIXME httpbin.org does not seem normal
   IXUnityBuildsTest
@@ -106,6 +107,8 @@ foreach(TEST_TARGET_NAME ${TEST_TARGET_NAMES})
 
 endforeach()
 
+set_tests_properties(IXWebSocketReconnectTest PROPERTIES TIMEOUT 30)
+
 if (USE_NETWORK_FRAMEWORK)
   # Only the test credential generator needs the openssl command-line tool.
   find_program(NETWORK_TEST_OPENSSL openssl)