me-no-dev/ESPAsyncWebServer
1
0
Fork 1
mirror of https://github.com/me-no-dev/ESPAsyncWebServer.git synced 2025-08-03 20:54:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor AsyncEventSourceMessage

Browse Source
This commit is contained in:
Emil Muratov
2024-06-26 21:01:52 +09:00
committed by Mathieu Carbou
parent bb4eb89c8e
commit 48968b5be5
2 changed files with 19 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/AsyncEventSource.cpp
View File

@@ -183,30 +183,25 @@ AsyncEventSourceClient::~AsyncEventSourceClient(){
close(); close();
} }
void AsyncEventSourceClient::_queueMessage(AsyncEventSourceMessage *dataMessage){ void AsyncEventSourceClient::_queueMessage(const char * message, size_t len){
if(dataMessage == NULL)
return;
if(!connected()){
delete dataMessage;
return;
}
#ifdef ESP32 #ifdef ESP32
//length() is not thread-safe, thus acquiring the lock before this call.. //length() is not thread-safe, thus acquiring the lock before this call..
std::lock_guard<std::mutex> lock(_lockmq); std::lock_guard<std::mutex> lock(_lockmq);
#endif #endif
if(_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES){ if(_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES){
#ifdef ESP8266 #ifdef ESP8266
ets_printf(String(F("ERROR: Too many messages queued\n")).c_str()); ets_printf(String(F("ERROR: Too many messages queued\n")).c_str());
#else #else
log_e("Too many messages queued: deleting message"); log_e("Too many messages queued: deleting message");
#endif #endif
delete dataMessage; return;
} else { }
_messageQueue.emplace_back(dataMessage);
// runqueue trigger when new messages added _messageQueue.emplace_back(message, len);
if(_client->canSend()) { // runqueue trigger when new messages added
_runQueue(); if(_client->canSend()) {
} _runQueue();
} }
} }
@@ -216,8 +211,8 @@ void AsyncEventSourceClient::_onAck(size_t len, uint32_t time){
std::lock_guard<std::mutex> lock(_lockmq); std::lock_guard<std::mutex> lock(_lockmq);
#endif #endif
while(len && _messageQueue.size()){ while(len && _messageQueue.size()){
len = _messageQueue.front()->ack(len, time); len = _messageQueue.front().ack(len, time);
if(_messageQueue.front()->finished()) if(_messageQueue.front().finished())
_messageQueue.pop_front(); _messageQueue.pop_front();
} }
_runQueue(); _runQueue();
@@ -248,12 +243,14 @@ void AsyncEventSourceClient::close(){
} }
void AsyncEventSourceClient::write(const char * message, size_t len){ void AsyncEventSourceClient::write(const char * message, size_t len){
_queueMessage(new AsyncEventSourceMessage(message, len)); if(!connected()) return;
_queueMessage(message, len);
} }
void AsyncEventSourceClient::send(const char *message, const char *event, uint32_t id, uint32_t reconnect){ void AsyncEventSourceClient::send(const char *message, const char *event, uint32_t id, uint32_t reconnect){
if(!connected()) return;
String ev = generateEventMessage(message, event, id, reconnect); String ev = generateEventMessage(message, event, id, reconnect);
_queueMessage(new AsyncEventSourceMessage(ev.c_str(), ev.length())); _queueMessage(ev.c_str(), ev.length());
} }
size_t AsyncEventSourceClient::packetsWaiting() const { size_t AsyncEventSourceClient::packetsWaiting() const {
@@ -267,8 +264,8 @@ void AsyncEventSourceClient::_runQueue() {
// Calls to this private method now already protected by _lockmq acquisition // Calls to this private method now already protected by _lockmq acquisition
// so no extra call of _lockmq.lock() here.. // so no extra call of _lockmq.lock() here..
for ( auto &i : _messageQueue){ for ( auto &i : _messageQueue){
if (!i->sent()) if (!i.sent())
i->send(_client); i.send(_client);
} }
} }

5
src/AsyncEventSource.h
View File

@@ -79,12 +79,11 @@ class AsyncEventSourceClient {
AsyncClient *_client; AsyncClient *_client;
AsyncEventSource *_server; AsyncEventSource *_server;
uint32_t _lastId; uint32_t _lastId;
std::list< std::unique_ptr<AsyncEventSourceMessage> > _messageQueue; std::list< AsyncEventSourceMessage > _messageQueue;
// ArFi 2020-08-27 for protecting/serializing _messageQueue
#ifdef ESP32 #ifdef ESP32
mutable std::mutex _lockmq; mutable std::mutex _lockmq;
#endif #endif
void _queueMessage(AsyncEventSourceMessage *dataMessage); void _queueMessage(const char * message, size_t len);
void _runQueue(); void _runQueue();
public: public: