Add stable IDs for targeted SSE client closure

docs/eventsource.md

@@ -36,6 +36,29 @@ void loop(){
 }
 ```
 
+### Identify and close one connection
+
+Each connected client has a nonzero ID that remains stable for that connection.
+Use the request-aware connect callback when application state such as an
+authenticated session must be associated with the connection. Do not retain the
+client pointer after a callback; retain `client->id()` instead.
+
+```cpp
+AsyncEventSourceClientId clientId = 0;
+
+events.onConnectWithRequest([](AsyncWebServerRequest *request, AsyncEventSourceClient *client) {
+  clientId = client->id();
+  // Read authentication/session state from request here.
+});
+
+// Call from the application's normal service context, not an AsyncTCP callback.
+events.closeClient(clientId);
+```
+
+`closeClient()` returns `false` when the ID is zero or no longer connected. IDs
+are scoped to an `AsyncEventSource` instance and may eventually be reused after
+the 32-bit counter wraps.
+
 **IMPORTANT**: Use `AsyncAuthenticationMiddleware` instead of the deprecated `setAuthentication()` method for authentication.
 
 ```cpp

examples/arduino/ServerSentEvents/ServerSentEvents.ino

@@ -70,13 +70,13 @@ void setup() {
     request->send(200, "text/html", (uint8_t *)htmlContent, htmlContentLength);
   });
 
-  events.onConnect([](AsyncEventSourceClient *client) {
-    Serial.printf("SSE Client connected!");
+  events.onConnectWithRequest([](AsyncWebServerRequest *request, AsyncEventSourceClient *client) {
+    Serial.printf("SSE Client %" PRIu32 " connected to %s!", client->id(), request->url().c_str());
     client->send("hello!", NULL, millis(), 1000);
   });
 
   events.onDisconnect([](AsyncEventSourceClient *client) {
-    Serial.printf("SSE Client disconnected!");
+    Serial.printf("SSE Client %" PRIu32 " disconnected!", client->id());
   });
 
   server.addHandler(&events);

src/AsyncEventSource.cpp

@@ -147,7 +147,9 @@ size_t AsyncEventSourceMessage::send(AsyncClient *client) {
 
 // Client
 
-AsyncEventSourceClient::AsyncEventSourceClient(AsyncClient *client, AsyncEventSource *server, uint32_t lastId)
+AsyncEventSourceClient::AsyncEventSourceClient(
+  AsyncClient *client, AsyncEventSource *server, uint32_t lastId, AsyncWebServerRequest *request
+)
   : _client(client), _server(server), _lastId(lastId) {
 
   _client->setRxTimeout(0);
@@ -181,7 +183,7 @@ AsyncEventSourceClient::AsyncEventSourceClient(AsyncClient *client, AsyncEventSo
     this
   );
 
-  _server->_addClient(this);
+  _server->_addClient(this, request);
   _client->setNoDelay(true);
 }
 
@@ -349,14 +351,25 @@ void AsyncEventSource::authorizeConnect(ArAuthorizeConnectHandler cb) {
   addMiddleware(m);
 }
 
-void AsyncEventSource::_addClient(AsyncEventSourceClient *client) {
+void AsyncEventSource::_addClient(AsyncEventSourceClient *client, AsyncWebServerRequest *request) {
   if (!client) {
     return;
   }
 
+  {
+    asyncsrv::lock_guard_type lock(_client_queue_lock);
+    client->_id = _nextClientId++;
+    if (_nextClientId == 0) {
+      _nextClientId = 1;
+    }
+  }
+
   if (_connectcb) {
     _connectcb(client);
   }
+  if (_requestConnectcb) {
+    _requestConnectcb(request, client);
+  }
 
   asyncsrv::lock_guard_type lock(_client_queue_lock);
   _clients.emplace_back(client);
@@ -395,6 +408,37 @@ void AsyncEventSource::close() {
   }
 }
 
+bool AsyncEventSource::closeClient(AsyncEventSourceClientId id) {
+  if (id == 0) {
+    return false;
+  }
+
+  std::unique_ptr<AsyncEventSourceClient> client;
+  {
+    asyncsrv::lock_guard_type lock(_client_queue_lock);
+    for (auto i = _clients.begin(); i != _clients.end(); ++i) {
+      if ((*i)->id() == id) {
+        client = std::move(*i);
+        _clients.erase(i);
+        _adjust_inflight_window();
+        break;
+      }
+    }
+  }
+
+  if (!client) {
+    return false;
+  }
+
+  client->close();
+  if (client->connected()) {
+    asyncsrv::lock_guard_type lock(_client_queue_lock);
+    _clients.emplace_back(std::move(client));
+    _adjust_inflight_window();
+  }
+  return true;
+}
+
 // pmb fix
 size_t AsyncEventSource::avgPacketsWaiting() const {
   size_t aql = 0;
@@ -480,5 +524,5 @@ void AsyncEventSourceResponse::_respond(AsyncWebServerRequest *request) {
   request->client()->write(out.c_str(), _headLength);
   // Add a new AsyncEventSourceClient to the server's list of clients
   // This adopts the ownership of the AsyncTCP's client pointer from `request` parameter
-  new AsyncEventSourceClient(request->clientRelease(), _server, lastId);
+  new AsyncEventSourceClient(request->clientRelease(), _server, lastId, request);
 }

src/AsyncEventSource.h

@@ -50,7 +50,9 @@
 class AsyncEventSource;
 class AsyncEventSourceResponse;
 class AsyncEventSourceClient;
+using AsyncEventSourceClientId = uint32_t;
 using ArEventHandlerFunction = std::function<void(AsyncEventSourceClient *client)>;
+using ArEventConnectHandlerFunction = std::function<void(AsyncWebServerRequest *request, AsyncEventSourceClient *client)>;
 using ArAuthorizeConnectHandler = ArAuthorizeFunction;
 // shared message object container
 using AsyncEvent_SharedData_t = std::shared_ptr<String>;
@@ -130,8 +132,10 @@ class AsyncEventSourceMessage {
  */
 class AsyncEventSourceClient {
 private:
+  friend class AsyncEventSource;
   AsyncClient *_client;
   AsyncEventSource *_server;
+  AsyncEventSourceClientId _id{0};
   uint32_t _lastId{0};
   size_t _inflight{0};                    // num of unacknowledged bytes that has been written to socket buffer
   size_t _max_inflight{SSE_MAX_INFLIGH};  // max num of unacknowledged bytes that could be written to socket buffer
@@ -150,7 +154,9 @@ class AsyncEventSourceClient {
    * @param server
    * @param lastId
    */
-  AsyncEventSourceClient(AsyncClient *client, AsyncEventSource *server, uint32_t lastId = 0);
+  AsyncEventSourceClient(
+    AsyncClient *client, AsyncEventSource *server, uint32_t lastId = 0, AsyncWebServerRequest *request = nullptr
+  );
   ~AsyncEventSourceClient();
 
   /**
@@ -203,6 +209,9 @@ class AsyncEventSourceClient {
   uint32_t lastId() const {
     return _lastId;
   }
+  AsyncEventSourceClientId id() const {
+    return _id;
+  }
   size_t packetsWaiting() const {
     asyncsrv::lock_guard_type lock(_lockmq);
     return _messageQueue.size();
@@ -246,7 +255,9 @@ class AsyncEventSource : public AsyncWebHandler {
   // since simultaneous access from different tasks is possible
   mutable asyncsrv::mutex_type _client_queue_lock;
   ArEventHandlerFunction _connectcb = nullptr;
+  ArEventConnectHandlerFunction _requestConnectcb = nullptr;
   ArEventHandlerFunction _disconnectcb = nullptr;
+  AsyncEventSourceClientId _nextClientId{1};
 
   // this method manipulates in-fligh data size for connected client depending on number of active connections
   void _adjust_inflight_window();
@@ -270,6 +281,9 @@ class AsyncEventSource : public AsyncWebHandler {
   // close all connected clients
   void close();
 
+  // close one connected client by its stable event-source ID
+  bool closeClient(AsyncEventSourceClientId id);
+
   /**
      * @brief set on-connect callback for the client
      * used to deliver messages to client on first connect
@@ -280,6 +294,14 @@ class AsyncEventSource : public AsyncWebHandler {
     _connectcb = cb;
   }
 
+  /**
+   * @brief set an on-connect callback that also receives the originating request
+   * @note the request pointer is valid only for the duration of the callback
+   */
+  void onConnectWithRequest(ArEventConnectHandlerFunction cb) {
+    _requestConnectcb = cb;
+  }
+
   /**
      * @brief Send an SSE message to client
      * it will craft an SSE message and place it to all connected client's message queues
@@ -311,7 +333,7 @@ class AsyncEventSource : public AsyncWebHandler {
   size_t avgPacketsWaiting() const;
 
   // system callbacks (do not call from user code!)
-  void _addClient(AsyncEventSourceClient *client);
+  void _addClient(AsyncEventSourceClient *client, AsyncWebServerRequest *request = nullptr);
   void _handleDisconnect(AsyncEventSourceClient *client);
   bool canHandle(AsyncWebServerRequest *request) const final;
   void handleRequest(AsyncWebServerRequest *request) final;