From d9be76c4a90daa25eccd21e04b45e1a06833929b Mon Sep 17 00:00:00 2001 From: Jasper Parker Date: Sun, 20 Sep 2026 00:33:57 +0800 Subject: [PATCH] Add stable IDs for targeted SSE client closure --- docs/eventsource.md | 23 ++++++++ .../ServerSentEvents/ServerSentEvents.ino | 6 +-- src/AsyncEventSource.cpp | 52 +++++++++++++++++-- src/AsyncEventSource.h | 26 +++++++++- 4 files changed, 98 insertions(+), 9 deletions(-) diff --git a/docs/eventsource.md b/docs/eventsource.md index 3d5dd86a5..1d6095f9b 100644 --- a/docs/eventsource.md +++ b/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 diff --git a/examples/arduino/ServerSentEvents/ServerSentEvents.ino b/examples/arduino/ServerSentEvents/ServerSentEvents.ino index bc6718c8c..adb851944 100644 --- a/examples/arduino/ServerSentEvents/ServerSentEvents.ino +++ b/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); diff --git a/src/AsyncEventSource.cpp b/src/AsyncEventSource.cpp index 39d94e3fa..82461213c 100644 --- a/src/AsyncEventSource.cpp +++ b/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 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); } diff --git a/src/AsyncEventSource.h b/src/AsyncEventSource.h index e6f94dff6..cd9c28cc3 100644 --- a/src/AsyncEventSource.h +++ b/src/AsyncEventSource.h @@ -50,7 +50,9 @@ class AsyncEventSource; class AsyncEventSourceResponse; class AsyncEventSourceClient; +using AsyncEventSourceClientId = uint32_t; using ArEventHandlerFunction = std::function; +using ArEventConnectHandlerFunction = std::function; using ArAuthorizeConnectHandler = ArAuthorizeFunction; // shared message object container using AsyncEvent_SharedData_t = std::shared_ptr; @@ -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;