Fix #364 Introduce cleanup() function

This function is similar to `cleanupClients(N)` for WebSocket except that it only cleanups internal resources.

This "book-keeping" function has to be called in the loop().

This is a non-elegant attempt to fix issue #364

Author: mathieucarbou

src/AsyncEventSource.cpp

@@ -370,7 +370,21 @@ void AsyncEventSource::_addClient(AsyncEventSourceClient *client) {
 #ifdef ESP32
   std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
 #endif
-  _clients.emplace_back(client);
+
+  // find first unique_ptr with nullptr and reuse it
+  bool reused = false;
+  for (auto &c : _clients) {
+    if (c.get() == nullptr) {
+      c.reset(client);
+      reused = true;
+      break;
+    }
+  }
+
+  if (!reused) {
+    _clients.emplace_back(client);
+  }
+
   if (_connectcb) {
     _connectcb(client);
   }
@@ -387,7 +401,7 @@ void AsyncEventSource::_handleDisconnect(AsyncEventSourceClient *client) {
 #endif
   for (auto i = _clients.begin(); i != _clients.end(); ++i) {
     if (i->get() == client) {
-      _clients.erase(i);
+      i->reset();  // reset the unique_ptr but do not remove the list entry yet to keep other iterators valid
       break;
     }
   }
@@ -402,7 +416,7 @@ void AsyncEventSource::close() {
   std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
 #endif
   for (const auto &c : _clients) {
-    if (c->connected()) {
+    if (c.get() != nullptr && c->connected()) {
       /**
        * @brief: Fix self-deadlock by using recursive_mutex instead.
        * Due to c->close() shall call the callback function _onDisconnect()
@@ -413,6 +427,17 @@ void AsyncEventSource::close() {
   }
 }
 
+void AsyncEventSource::cleanup() {
+#ifdef ESP32
+  std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
+#endif
+
+  // resize the list to remove unique_ptr with nullptr
+  _clients.remove_if([](const std::unique_ptr<AsyncEventSourceClient> &c) {
+    return c.get() == nullptr;
+  });
+}
+
 // pmb fix
 size_t AsyncEventSource::avgPacketsWaiting() const {
   size_t aql = 0;
@@ -421,7 +446,7 @@ size_t AsyncEventSource::avgPacketsWaiting() const {
   std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
 #endif
   for (const auto &c : _clients) {
-    if (c->connected()) {
+    if (c.get() != nullptr && c->connected()) {
       aql += c->packetsWaiting();
       ++nConnectedClients;
     }
@@ -437,7 +462,7 @@ AsyncEventSource::SendStatus AsyncEventSource::send(const char *message, const c
   size_t hits = 0;
   size_t miss = 0;
   for (const auto &c : _clients) {
-    if (c->connected()) {
+    if (c.get() != nullptr && c->connected()) {
       if (c->write(shared_msg)) {
         ++hits;
       } else {
@@ -454,7 +479,7 @@ size_t AsyncEventSource::count() const {
 #endif
   size_t n_clients{0};
   for (const auto &i : _clients) {
-    if (i->connected()) {
+    if (i.get() != nullptr && i->connected()) {
       ++n_clients;
     }
   }
@@ -476,7 +501,7 @@ void AsyncEventSource::_adjust_inflight_window() {
   if (clientCount) {
     size_t inflight = SSE_MAX_INFLIGH / clientCount;
     for (const auto &c : _clients) {
-      if (c->connected()) {
+      if (c.get() != nullptr && c->connected()) {
         c->set_max_inflight_bytes(inflight);
       }
     }

src/AsyncEventSource.h

@@ -272,6 +272,10 @@ class AsyncEventSource : public AsyncWebHandler {
   // close all connected clients
   void close();
 
+  // Cleanup internal resources.
+  // Has to be called periodically in the loop
+  void cleanup();
+
   /**
      * @brief set on-connect callback for the client
      * used to deliver messages to client on first connect