Fix race condition in the SSE code

Author: mathieucarbou

src/AsyncEventSource.cpp

@@ -199,16 +199,16 @@ AsyncEventSourceClient::~AsyncEventSourceClient() {
 }
 
 bool AsyncEventSourceClient::_queueMessage(const char *message, size_t len) {
+#ifdef ESP32
+  // Protect message queue access (size checks and modifications) which is not thread-safe.
+  std::lock_guard<std::recursive_mutex> lock(_lockmq);
+#endif
+
   if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) {
     async_ws_log_w("Event message queue overflow: discard message");
     return false;
   }
 
-#ifdef ESP32
-  // length() is not thread-safe, thus acquiring the lock before this call..
-  std::lock_guard<std::recursive_mutex> lock(_lockmq);
-#endif
-
   if (_client) {
     _messageQueue.emplace_back(message, len);
   } else {
@@ -230,16 +230,16 @@ bool AsyncEventSourceClient::_queueMessage(const char *message, size_t len) {
 }
 
 bool AsyncEventSourceClient::_queueMessage(AsyncEvent_SharedData_t &&msg) {
-  if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) {
-    async_ws_log_w("Event message queue overflow: discard message");
-    return false;
-  }
-
 #ifdef ESP32
   // length() is not thread-safe, thus acquiring the lock before this call..
   std::lock_guard<std::recursive_mutex> lock(_lockmq);
 #endif
 
+  if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) {
+    async_ws_log_w("Event message queue overflow: discard message");
+    return false;
+  }
+
   if (_client) {
     _messageQueue.emplace_back(std::move(msg));
   } else {
@@ -288,11 +288,11 @@ void AsyncEventSourceClient::_onAck(size_t len __attribute__((unused)), uint32_t
 }
 
 void AsyncEventSourceClient::_onPoll() {
-  if (_messageQueue.size()) {
 #ifdef ESP32
-    // Same here, acquiring the lock early
-    std::lock_guard<std::recursive_mutex> lock(_lockmq);
+  // Acquire the lock early before checking queue size, same as _queueMessage
+  std::lock_guard<std::recursive_mutex> lock(_lockmq);
 #endif
+  if (_messageQueue.size()) {
     _runQueue();
   }
 }
@@ -379,12 +379,12 @@ void AsyncEventSource::_addClient(AsyncEventSourceClient *client) {
 }
 
 void AsyncEventSource::_handleDisconnect(AsyncEventSourceClient *client) {
-  if (_disconnectcb) {
-    _disconnectcb(client);
-  }
 #ifdef ESP32
   std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
 #endif
+  if (_disconnectcb) {
+    _disconnectcb(client);
+  }
   for (auto i = _clients.begin(); i != _clients.end(); ++i) {
     if (i->get() == client) {
       _clients.erase(i);

src/AsyncEventSource.h

@@ -205,6 +205,9 @@ class AsyncEventSourceClient {
     return _lastId;
   }
   size_t packetsWaiting() const {
+#ifdef ESP32
+    std::lock_guard<std::recursive_mutex> lock(_lockmq);
+#endif
     return _messageQueue.size();
   };