xPablo.cz/ESPAsyncWebServer
2
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Aktualizace na verzi 3.4.5

Browse Source
This commit is contained in:
Pavel Brychta
2025-01-03 16:45:58 +01:00
parent 7c828c70d8
commit 697a3adf08
17 changed files with 893 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

326
src/AsyncEventSource.cpp
View File

@ -23,116 +23,99 @@
#endif
#include "AsyncEventSource.h"
#define ASYNC_SSE_NEW_LINE_CHAR (char)0xa
using namespace asyncsrv;
static String generateEventMessage(const char* message, const char* event, uint32_t id, uint32_t reconnect) {
String ev;
String str;
size_t len{0};
if (message)
len += strlen(message);
if (event)
len += strlen(event);
len += 42; // give it some overhead
str.reserve(len);
if (reconnect) {
ev += T_retry_;
ev += reconnect;
ev += T_rn;
str += T_retry_;
str += reconnect;
str += ASYNC_SSE_NEW_LINE_CHAR; // '\n'
}
if (id) {
ev += T_id__;
ev += id;
ev += T_rn;
str += T_id__;
str += id;
str += ASYNC_SSE_NEW_LINE_CHAR; // '\n'
}
if (event != NULL) {
ev += T_event_;
ev += event;
ev += T_rn;
str += T_event_;
str += event;
str += ASYNC_SSE_NEW_LINE_CHAR; // '\n'
}
if (message != NULL) {
size_t messageLen = strlen(message);
char* lineStart = (char*)message;
char* lineEnd;
do {
char* nextN = strchr(lineStart, '\n');
char* nextR = strchr(lineStart, '\r');
if (nextN == NULL && nextR == NULL) {
size_t llen = ((char*)message + messageLen) - lineStart;
char* ldata = (char*)malloc(llen + 1);
if (ldata != NULL) {
memcpy(ldata, lineStart, llen);
ldata[llen] = 0;
ev += T_data_;
ev += ldata;
ev += T_rnrn;
free(ldata);
}
lineStart = (char*)message + messageLen;
if (!message)
return str;
size_t messageLen = strlen(message);
char* lineStart = (char*)message;
char* lineEnd;
do {
char* nextN = strchr(lineStart, '\n');
char* nextR = strchr(lineStart, '\r');
if (nextN == NULL && nextR == NULL) {
// a message is a single-line string
str += T_data_;
str += message;
str += T_nn;
return str;
}
// a message is a multi-line string
char* nextLine = NULL;
if (nextN != NULL && nextR != NULL) { // windows line-ending \r\n
if (nextR + 1 == nextN) {
// normal \r\n sequense
lineEnd = nextR;
nextLine = nextN + 1;
} else {
char* nextLine = NULL;
if (nextN != NULL && nextR != NULL) {
if (nextR < nextN) {
lineEnd = nextR;
if (nextN == (nextR + 1))
nextLine = nextN + 1;
else
nextLine = nextR + 1;
} else {
lineEnd = nextN;
if (nextR == (nextN + 1))
nextLine = nextR + 1;
else
nextLine = nextN + 1;
}
} else if (nextN != NULL) {
lineEnd = nextN;
nextLine = nextN + 1;
} else {
lineEnd = nextR;
nextLine = nextR + 1;
}
size_t llen = lineEnd - lineStart;
char* ldata = (char*)malloc(llen + 1);
if (ldata != NULL) {
memcpy(ldata, lineStart, llen);
ldata[llen] = 0;
ev += T_data_;
ev += ldata;
ev += T_rn;
free(ldata);
}
lineStart = nextLine;
if (lineStart == ((char*)message + messageLen))
ev += T_rn;
// some abnormal \n \r mixed sequence
lineEnd = std::min(nextR, nextN);
nextLine = lineEnd + 1;
}
} while (lineStart < ((char*)message + messageLen));
}
} else if (nextN != NULL) { // Unix/Mac OS X LF
lineEnd = nextN;
nextLine = nextN + 1;
} else { // some ancient garbage
lineEnd = nextR;
nextLine = nextR + 1;
}
return ev;
str += T_data_;
str.concat(lineStart, lineEnd - lineStart);
str += ASYNC_SSE_NEW_LINE_CHAR; // \n
lineStart = nextLine;
} while (lineStart < ((char*)message + messageLen));
// append another \n to terminate message
str += ASYNC_SSE_NEW_LINE_CHAR; // '\n'
return str;
}
// Message
AsyncEventSourceMessage::AsyncEventSourceMessage(const char* data, size_t len)
: _data(nullptr), _len(len), _sent(0), _acked(0) {
_data = (uint8_t*)malloc(_len + 1);
if (_data == nullptr) {
_len = 0;
} else {
memcpy(_data, data, len);
_data[_len] = 0;
}
}
AsyncEventSourceMessage::~AsyncEventSourceMessage() {
if (_data != NULL)
free(_data);
}
size_t AsyncEventSourceMessage::ack(size_t len, __attribute__((unused)) uint32_t time) {
// If the whole message is now acked...
if (_acked + len > _len) {
if (_acked + len > _data->length()) {
// Return the number of extra bytes acked (they will be carried on to the next message)
const size_t extra = _acked + len - _len;
_acked = _len;
const size_t extra = _acked + len - _data->length();
_acked = _data->length();
return extra;
}
// Return that no extra bytes left.
@ -144,13 +127,25 @@ size_t AsyncEventSourceMessage::write(AsyncClient* client) {
if (!client)
return 0;
if (_sent >= _len || !client->canSend()) {
if (_sent >= _data->length() || !client->canSend()) {
return 0;
}
size_t len = min(_len - _sent, client->space());
size_t sent = client->add((const char*)_data + _sent, len);
_sent += sent;
return sent;
size_t len = std::min(_data->length() - _sent, client->space());
/*
add() would call lwip's tcp_write() under the AsyncTCP hood with apiflags argument.
By default apiflags=ASYNC_WRITE_FLAG_COPY
we could have used apiflags with this flag unset to pass data by reference and avoid copy to socket buffer,
but looks like it does not work for Arduino's lwip in ESP32/IDF
it is enforced in https://github.com/espressif/esp-lwip/blob/0606eed9d8b98a797514fdf6eabb4daf1c8c8cd9/src/core/tcp_out.c#L422C5-L422C30
if LWIP_NETIF_TX_SINGLE_PBUF is set, and it is set indeed in IDF
https://github.com/espressif/esp-idf/blob/a0f798cfc4bbd624aab52b2c194d219e242d80c1/components/lwip/port/include/lwipopts.h#L744
So let's just keep it enforced ASYNC_WRITE_FLAG_COPY and keep in mind that there is no zero-copy
*/
size_t written = client->add(_data->c_str() + _sent, len, ASYNC_WRITE_FLAG_COPY); // ASYNC_WRITE_FLAG_MORE
_sent += written;
return written;
}
size_t AsyncEventSourceMessage::send(AsyncClient* client) {
@ -160,20 +155,19 @@ size_t AsyncEventSourceMessage::send(AsyncClient* client) {
// Client
AsyncEventSourceClient::AsyncEventSourceClient(AsyncWebServerRequest* request, AsyncEventSource* server) {
_client = request->client();
_server = server;
_lastId = 0;
AsyncEventSourceClient::AsyncEventSourceClient(AsyncWebServerRequest* request, AsyncEventSource* server)
: _client(request->client()), _server(server) {
if (request->hasHeader(T_Last_Event_ID))
_lastId = atoi(request->getHeader(T_Last_Event_ID)->value().c_str());
_client->setRxTimeout(0);
_client->onError(NULL, NULL);
_client->onAck([](void* r, AsyncClient* c, size_t len, uint32_t time) { (void)c; ((AsyncEventSourceClient*)(r))->_onAck(len, time); }, this);
_client->onPoll([](void* r, AsyncClient* c) { (void)c; ((AsyncEventSourceClient*)(r))->_onPoll(); }, this);
_client->onAck([](void* r, AsyncClient* c, size_t len, uint32_t time) { (void)c; static_cast<AsyncEventSourceClient*>(r)->_onAck(len, time); }, this);
_client->onPoll([](void* r, AsyncClient* c) { (void)c; static_cast<AsyncEventSourceClient*>(r)->_onPoll(); }, this);
_client->onData(NULL, NULL);
_client->onTimeout([this](void* r, AsyncClient* c __attribute__((unused)), uint32_t time) { ((AsyncEventSourceClient*)(r))->_onTimeout(time); }, this);
_client->onDisconnect([this](void* r, AsyncClient* c) { ((AsyncEventSourceClient*)(r))->_onDisconnect(); delete c; }, this);
_client->onTimeout([this](void* r, AsyncClient* c __attribute__((unused)), uint32_t time) { static_cast<AsyncEventSourceClient*>(r)->_onTimeout(time); }, this);
_client->onDisconnect([this](void* r, AsyncClient* c) { static_cast<AsyncEventSourceClient*>(r)->_onDisconnect(); delete c; }, this);
_server->_addClient(this);
delete request;
@ -190,29 +184,61 @@ AsyncEventSourceClient::~AsyncEventSourceClient() {
}
bool AsyncEventSourceClient::_queueMessage(const char* message, size_t len) {
if (!_client)
if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) {
#ifdef ESP8266
ets_printf(String(F("ERROR: Too many messages queued\n")).c_str());
#elif defined(ESP32)
log_e("Event message queue overflow: discard message");
#endif
return false;
}
#ifdef ESP32
// length() is not thread-safe, thus acquiring the lock before this call..
std::lock_guard<std::mutex> lock(_lockmq);
#endif
_messageQueue.emplace_back(message, len);
/*
throttle queue run
if Q is filled for >25% then network/CPU is congested, since there is no zero-copy mode for socket buff
forcing Q run will only eat more heap ram and blow the buffer, let's just keep data in our own queue
the queue will be processed at least on each onAck()/onPoll() call from AsyncTCP
*/
if (_messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2 && _client->canSend()) {
_runQueue();
}
return true;
}
bool AsyncEventSourceClient::_queueMessage(AsyncEvent_SharedData_t&& msg) {
if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) {
#ifdef ESP8266
ets_printf(String(F("ERROR: Too many messages queued\n")).c_str());
#elif defined(ESP32)
log_e("Too many messages queued: deleting message");
log_e("Event message queue overflow: discard message");
#endif
return false;
}
_messageQueue.emplace_back(message, len);
// runqueue trigger when new messages added
if (_client->canSend()) {
#ifdef ESP32
// length() is not thread-safe, thus acquiring the lock before this call..
std::lock_guard<std::mutex> lock(_lockmq);
#endif
_messageQueue.emplace_back(std::move(msg));
/*
throttle queue run
if Q is filled for >25% then network/CPU is congested, since there is no zero-copy mode for socket buff
forcing Q run will only eat more heap ram and blow the buffer, let's just keep data in our own queue
the queue will be processed at least on each onAck()/onPoll() call from AsyncTCP
*/
if (_messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2 && _client->canSend()) {
_runQueue();
}
return true;
}
@ -221,15 +247,33 @@ void AsyncEventSourceClient::_onAck(size_t len __attribute__((unused)), uint32_t
// Same here, acquiring the lock early
std::lock_guard<std::mutex> lock(_lockmq);
#endif
_runQueue();
// adjust in-flight len
if (len < _inflight)
_inflight -= len;
else
_inflight = 0;
// acknowledge as much messages's data as we got confirmed len from a AsyncTCP
while (len && _messageQueue.size()) {
len = _messageQueue.front().ack(len);
if (_messageQueue.front().finished()) {
// now we could release full ack'ed messages, we were keeping it unless send confirmed from AsyncTCP
_messageQueue.pop_front();
}
}
// try to send another batch of data
if (_messageQueue.size())
_runQueue();
}
void AsyncEventSourceClient::_onPoll() {
#ifdef ESP32
// Same here, acquiring the lock early
std::lock_guard<std::mutex> lock(_lockmq);
#endif
if (_messageQueue.size()) {
#ifdef ESP32
// Same here, acquiring the lock early
std::lock_guard<std::mutex> lock(_lockmq);
#endif
_runQueue();
}
}
@ -251,50 +295,42 @@ void AsyncEventSourceClient::close() {
_client->close();
}
bool AsyncEventSourceClient::write(const char* message, size_t len) {
return connected() && _queueMessage(message, len);
}
bool AsyncEventSourceClient::send(const char* message, const char* event, uint32_t id, uint32_t reconnect) {
if (!connected())
return false;
String ev = generateEventMessage(message, event, id, reconnect);
return _queueMessage(ev.c_str(), ev.length());
}
size_t AsyncEventSourceClient::packetsWaiting() const {
#ifdef ESP32
std::lock_guard<std::mutex> lock(_lockmq);
#endif
return _messageQueue.size();
return _queueMessage(std::make_shared<String>(generateEventMessage(message, event, id, reconnect)));
}
void AsyncEventSourceClient::_runQueue() {
if (!_client)
return;
// there is no need to lock the mutex here, 'cause all the calls to this method must be already lock'ed
size_t total_bytes_written = 0;
for (auto i = _messageQueue.begin(); i != _messageQueue.end(); ++i) {
if (!i->sent()) {
const size_t bytes_written = i->write(_client);
total_bytes_written += bytes_written;
if (bytes_written == 0)
_inflight += bytes_written;
if (bytes_written == 0 || _inflight > _max_inflight) {
// Serial.print("_");
break;
}
}
}
if (total_bytes_written > 0)
// flush socket
if (total_bytes_written)
_client->send();
size_t len = total_bytes_written;
while (len && _messageQueue.size()) {
len = _messageQueue.front().ack(len);
if (_messageQueue.front().finished()) {
_messageQueue.pop_front();
}
}
}
void AsyncEventSourceClient::set_max_inflight_bytes(size_t value) {
if (value >= SSE_MIN_INFLIGH && value <= SSE_MAX_INFLIGH)
_max_inflight = value;
}
/* AsyncEventSource */
void AsyncEventSource::authorizeConnect(ArAuthorizeConnectHandler cb) {
AuthorizationMiddleware* m = new AuthorizationMiddleware(401, cb);
m->_freeOnRemoval = true;
@ -310,18 +346,21 @@ void AsyncEventSource::_addClient(AsyncEventSourceClient* client) {
_clients.emplace_back(client);
if (_connectcb)
_connectcb(client);
_adjust_inflight_window();
}
void AsyncEventSource::_handleDisconnect(AsyncEventSourceClient* client) {
if (_disconnectcb)
_disconnectcb(client);
#ifdef ESP32
std::lock_guard<std::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);
}
_adjust_inflight_window();
}
void AsyncEventSource::close() {
@ -358,14 +397,14 @@ size_t AsyncEventSource::avgPacketsWaiting() const {
AsyncEventSource::SendStatus AsyncEventSource::send(
const char* message, const char* event, uint32_t id, uint32_t reconnect) {
String ev = generateEventMessage(message, event, id, reconnect);
AsyncEvent_SharedData_t shared_msg = std::make_shared<String>(generateEventMessage(message, event, id, reconnect));
#ifdef ESP32
std::lock_guard<std::mutex> lock(_client_queue_lock);
#endif
size_t hits = 0;
size_t miss = 0;
for (const auto& c : _clients) {
if (c->write(ev.c_str(), ev.length()))
if (c->write(shared_msg))
++hits;
else
++miss;
@ -393,7 +432,16 @@ void AsyncEventSource::handleRequest(AsyncWebServerRequest* request) {
request->send(new AsyncEventSourceResponse(this));
}
// Response
void AsyncEventSource::_adjust_inflight_window() {
if (_clients.size()) {
size_t inflight = SSE_MAX_INFLIGH / _clients.size();
for (const auto& c : _clients)
c->set_max_inflight_bytes(inflight);
// Serial.printf("adjusted inflight to: %u\n", inflight);
}
}
/* Response */
AsyncEventSourceResponse::AsyncEventSourceResponse(AsyncEventSource* server) {
_server = server;

174
src/AsyncEventSource.h
View File

@ -21,22 +21,29 @@
#define ASYNCEVENTSOURCE_H_
#include <Arduino.h>
#ifdef ESP32
#include <AsyncTCP.h>
#include <mutex>
#ifndef SSE_MAX_QUEUED_MESSAGES
#define SSE_MAX_QUEUED_MESSAGES 32
#endif
#define SSE_MIN_INFLIGH 2 * 1460 // allow 2 MSS packets
#define SSE_MAX_INFLIGH 16 * 1024 // but no more than 16k, no need to blow it, since same data is kept in local Q
#elif defined(ESP8266)
#include <ESPAsyncTCP.h>
#ifndef SSE_MAX_QUEUED_MESSAGES
#define SSE_MAX_QUEUED_MESSAGES 8
#endif
#define SSE_MIN_INFLIGH 2 * 1460 // allow 2 MSS packets
#define SSE_MAX_INFLIGH 8 * 1024 // but no more than 8k, no need to blow it, since same data is kept in local Q
#elif defined(TARGET_RP2040)
#include <AsyncTCP_RP2040W.h>
#ifndef SSE_MAX_QUEUED_MESSAGES
#define SSE_MAX_QUEUED_MESSAGES 32
#endif
#define SSE_MIN_INFLIGH 2 * 1460 // allow 2 MSS packets
#define SSE_MAX_INFLIGH 16 * 1024 // but no more than 16k, no need to blow it, since same data is kept in local Q
#endif
#include <ESPAsyncWebServer.h>
@ -53,58 +60,155 @@ class AsyncEventSourceResponse;
class AsyncEventSourceClient;
using ArEventHandlerFunction = std::function<void(AsyncEventSourceClient* client)>;
using ArAuthorizeConnectHandler = ArAuthorizeFunction;
// shared message object container
using AsyncEvent_SharedData_t = std::shared_ptr<String>;
/**
* @brief Async Event Message container with shared message content data
*
*/
class AsyncEventSourceMessage {
private:
uint8_t* _data;
size_t _len;
size_t _sent;
// size_t _ack;
size_t _acked;
const AsyncEvent_SharedData_t _data;
size_t _sent{0}; // num of bytes already sent
size_t _acked{0}; // num of bytes acked
public:
AsyncEventSourceMessage(const char* data, size_t len);
~AsyncEventSourceMessage();
AsyncEventSourceMessage(AsyncEvent_SharedData_t data) : _data(data) {};
#ifdef ESP32
AsyncEventSourceMessage(const char* data, size_t len) : _data(std::make_shared<String>(data, len)) {};
#else
// esp8266's String does not have constructor with data/length arguments. Use a concat method here
AsyncEventSourceMessage(const char* data, size_t len) { _data->concat(data, len); };
#endif
/**
* @brief acknowledge sending len bytes of data
* @note if num of bytes to ack is larger then the unacknowledged message length the number of carried over bytes are returned
*
* @param len bytes to acknowlegde
* @param time
* @return size_t number of extra bytes carried over
*/
size_t ack(size_t len, uint32_t time = 0);
/**
* @brief write message data to client's buffer
* @note this method does NOT call client's send
*
* @param client
* @return size_t number of bytes written
*/
size_t write(AsyncClient* client);
/**
* @brief writes message data to client's buffer and calls client's send method
*
* @param client
* @return size_t returns num of bytes the clien was able to send()
*/
size_t send(AsyncClient* client);
bool finished() { return _acked == _len; }
bool sent() { return _sent == _len; }
// returns true if full message's length were acked
bool finished() { return _acked == _data->length(); }
/**
* @brief returns true if all data has been sent already
*
*/
bool sent() { return _sent == _data->length(); }
};
/**
* @brief class holds a sse messages queue for a particular client's connection
*
*/
class AsyncEventSourceClient {
private:
AsyncClient* _client;
AsyncEventSource* _server;
uint32_t _lastId;
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
std::list<AsyncEventSourceMessage> _messageQueue;
#ifdef ESP32
mutable std::mutex _lockmq;
#endif
bool _queueMessage(const char* message, size_t len);
bool _queueMessage(AsyncEvent_SharedData_t&& msg);
void _runQueue();
public:
AsyncEventSourceClient(AsyncWebServerRequest* request, AsyncEventSource* server);
~AsyncEventSourceClient();
AsyncClient* client() { return _client; }
void close();
bool write(const char* message, size_t len);
/**
* @brief Send an SSE message to client
* it will craft an SSE message and place it to client's message queue
*
* @param message body string, could be single or multi-line string sepprated by \n, \r, \r\n
* @param event body string, a sinle line string
* @param id sequence id
* @param reconnect client's reconnect timeout
* @return true if message was placed in a queue
* @return false if queue is full
*/
bool send(const char* message, const char* event = NULL, uint32_t id = 0, uint32_t reconnect = 0);
bool send(const String& message, const String& event, uint32_t id = 0, uint32_t reconnect = 0) { return send(message.c_str(), event.c_str(), id, reconnect); }
bool send(const String& message, const char* event, uint32_t id = 0, uint32_t reconnect = 0) { return send(message.c_str(), event, id, reconnect); }
bool send(const char* message, const char* event = NULL, uint32_t id = 0, uint32_t reconnect = 0);
/**
* @brief place supplied preformatted SSE message to the message queue
* @note message must a properly formatted SSE string according to https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events
*
* @param message data
* @return true on success
* @return false on queue overflow or no client connected
*/
bool write(AsyncEvent_SharedData_t message) { return connected() && _queueMessage(std::move(message)); };
[[deprecated("Use _write(AsyncEvent_SharedData_t message) instead to share same data with multiple SSE clients")]]
bool write(const char* message, size_t len) { return connected() && _queueMessage(message, len); };
// close client's connection
void close();
// getters
AsyncClient* client() { return _client; }
bool connected() const { return _client && _client->connected(); }
uint32_t lastId() const { return _lastId; }
size_t packetsWaiting() const;
size_t packetsWaiting() const { return _messageQueue.size(); };
// system callbacks (do not call)
/**
* @brief Sets max amount of bytes that could be written to client's socket while awaiting delivery acknowledge
* used to throttle message delivery length to tradeoff memory consumption
* @note actual amount of data written could possible be a bit larger but no more than available socket buff space
*
* @param value
*/
void set_max_inflight_bytes(size_t value);
/**
* @brief Get current max inflight bytes value
*
* @return size_t
*/
size_t get_max_inflight_bytes() const { return _max_inflight; }
// system callbacks (do not call if from user code!)
void _onAck(size_t len, uint32_t time);
void _onPoll();
void _onTimeout(uint32_t time);
void _onDisconnect();
};
/**
* @brief a class that maintains all connected HTTP clients subscribed to SSE delivery
* dispatches supplied messages to the client's queues
*
*/
class AsyncEventSource : public AsyncWebHandler {
private:
String _url;
@ -117,6 +221,9 @@ class AsyncEventSource : public AsyncWebHandler {
ArEventHandlerFunction _connectcb = nullptr;
ArEventHandlerFunction _disconnectcb = nullptr;
// this method manipulates in-fligh data size for connected client depending on number of active connections
void _adjust_inflight_window();
public:
typedef enum {
DISCARDED = 0,
@ -124,23 +231,47 @@ class AsyncEventSource : public AsyncWebHandler {
PARTIALLY_ENQUEUED = 2,
} SendStatus;
AsyncEventSource(const char* url) : _url(url) {};
AsyncEventSource(const String& url) : _url(url) {};
~AsyncEventSource() { close(); };
const char* url() const { return _url.c_str(); }
// close all connected clients
void close();
/**
* @brief set on-connect callback for the client
* used to deliver messages to client on first connect
*
* @param cb
*/
void onConnect(ArEventHandlerFunction cb) { _connectcb = cb; }
/**
* @brief Send an SSE message to client
* it will craft an SSE message and place it to all connected client's message queues
*
* @param message body string, could be single or multi-line string sepprated by \n, \r, \r\n
* @param event body string, a sinle line string
* @param id sequence id
* @param reconnect client's reconnect timeout
* @return SendStatus if message was placed in any/all/part of the client's queues
*/
SendStatus send(const char* message, const char* event = NULL, uint32_t id = 0, uint32_t reconnect = 0);
SendStatus send(const String& message, const String& event, uint32_t id = 0, uint32_t reconnect = 0) { return send(message.c_str(), event.c_str(), id, reconnect); }
SendStatus send(const String& message, const char* event, uint32_t id = 0, uint32_t reconnect = 0) { return send(message.c_str(), event, id, reconnect); }
// The client pointer sent to the callback is only for reference purposes. DO NOT CALL ANY METHOD ON IT !
void onDisconnect(ArEventHandlerFunction cb) { _disconnectcb = cb; }
void authorizeConnect(ArAuthorizeConnectHandler cb);
SendStatus send(const String& message, const String& event, uint32_t id = 0, uint32_t reconnect = 0) { return send(message.c_str(), event.c_str(), id, reconnect); }
SendStatus send(const String& message, const char* event, uint32_t id = 0, uint32_t reconnect = 0) { return send(message.c_str(), event, id, reconnect); }
SendStatus send(const char* message, const char* event = NULL, uint32_t id = 0, uint32_t reconnect = 0);
// number of clients connected
// returns number of connected clients
size_t count() const;
// returns average number of messages pending in all client's queues
size_t avgPacketsWaiting() const;
// system callbacks (do not call)
// system callbacks (do not call from user code!)
void _addClient(AsyncEventSourceClient* client);
void _handleDisconnect(AsyncEventSourceClient* client);
bool canHandle(AsyncWebServerRequest* request) const override final;
@ -149,7 +280,6 @@ class AsyncEventSource : public AsyncWebHandler {
class AsyncEventSourceResponse : public AsyncWebServerResponse {
private:
String _content;
AsyncEventSource* _server;
public:

19
src/AsyncWebSocket.cpp
View File

@ -287,7 +287,6 @@ AsyncWebSocketClient::AsyncWebSocketClient(AsyncWebServerRequest* request, Async
_client->onTimeout([](void* r, AsyncClient* c, uint32_t time) { (void)c; ((AsyncWebSocketClient*)(r))->_onTimeout(time); }, this);
_client->onData([](void* r, AsyncClient* c, void* buf, size_t len) { (void)c; ((AsyncWebSocketClient*)(r))->_onData(buf, len); }, this);
_client->onPoll([](void* r, AsyncClient* c) { (void)c; ((AsyncWebSocketClient*)(r))->_onPoll(); }, this);
_server->_handleEvent(this, WS_EVT_CONNECT, request, NULL, 0);
delete request;
memset(&_pinfo, 0, sizeof(_pinfo));
}
@ -451,6 +450,8 @@ void AsyncWebSocketClient::close(uint16_t code, const char* message) {
if (_status != WS_CONNECTED)
return;
_status = WS_DISCONNECTING;
if (code) {
uint8_t packetLen = 2;
if (message != NULL) {
@ -496,30 +497,37 @@ void AsyncWebSocketClient::_onDisconnect() {
}
void AsyncWebSocketClient::_onData(void* pbuf, size_t plen) {
// Serial.println("onData");
_lastMessageTime = millis();
uint8_t* data = (uint8_t*)pbuf;
while (plen > 0) {
if (!_pstate) {
const uint8_t* fdata = data;
_pinfo.index = 0;
_pinfo.final = (fdata[0] & 0x80) != 0;
_pinfo.opcode = fdata[0] & 0x0F;
_pinfo.masked = (fdata[1] & 0x80) != 0;
_pinfo.len = fdata[1] & 0x7F;
// log_d("WS[%" PRIu32 "]: _onData: %" PRIu32, _clientId, plen);
// log_d("WS[%" PRIu32 "]: _status = %" PRIu32, _clientId, _status);
// log_d("WS[%" PRIu32 "]: _pinfo: index: %" PRIu64 ", final: %" PRIu8 ", opcode: %" PRIu8 ", masked: %" PRIu8 ", len: %" PRIu64, _clientId, _pinfo.index, _pinfo.final, _pinfo.opcode, _pinfo.masked, _pinfo.len);
data += 2;
plen -= 2;
if (_pinfo.len == 126) {
if (_pinfo.len == 126 && plen >= 2) {
_pinfo.len = fdata[3] | (uint16_t)(fdata[2]) << 8;
data += 2;
plen -= 2;
} else if (_pinfo.len == 127) {
} else if (_pinfo.len == 127 && plen >= 8) {
_pinfo.len = fdata[9] | (uint16_t)(fdata[8]) << 8 | (uint32_t)(fdata[7]) << 16 | (uint32_t)(fdata[6]) << 24 | (uint64_t)(fdata[5]) << 32 | (uint64_t)(fdata[4]) << 40 | (uint64_t)(fdata[3]) << 48 | (uint64_t)(fdata[2]) << 56;
data += 8;
plen -= 8;
}
if (_pinfo.masked) {
if (_pinfo.masked && plen >= 4) { // if ws.close() is called, Safari sends a close frame with plen 2 and masked bit set. We must not decrement plen which is already 0.
memcpy(_pinfo.mask, data, 4);
data += 4;
plen -= 4;
@ -772,6 +780,7 @@ void AsyncWebSocket::_handleEvent(AsyncWebSocketClient* client, AwsEventType typ
AsyncWebSocketClient* AsyncWebSocket::_newClient(AsyncWebServerRequest* request) {
_clients.emplace_back(request, this);
_handleEvent(&_clients.back(), WS_EVT_CONNECT, request, NULL, 0);
return &_clients.back();
}

6
src/ESPAsyncWebServer.h
View File

@ -48,10 +48,10 @@
#include "literals.h"
#define ASYNCWEBSERVER_VERSION "3.3.23"
#define ASYNCWEBSERVER_VERSION "3.4.5"
#define ASYNCWEBSERVER_VERSION_MAJOR 3
#define ASYNCWEBSERVER_VERSION_MINOR 3
#define ASYNCWEBSERVER_VERSION_REVISION 23
#define ASYNCWEBSERVER_VERSION_MINOR 4
#define ASYNCWEBSERVER_VERSION_REVISION 5
#define ASYNCWEBSERVER_FORK_mathieucarbou
#ifdef ASYNCWEBSERVER_REGEX

57
src/WebRequest.cpp
View File

@ -28,14 +28,14 @@
using namespace asyncsrv;
enum { PARSE_REQ_START,
PARSE_REQ_HEADERS,
PARSE_REQ_BODY,
PARSE_REQ_END,
PARSE_REQ_FAIL };
enum { PARSE_REQ_START = 0,
PARSE_REQ_HEADERS = 1,
PARSE_REQ_BODY = 2,
PARSE_REQ_END = 3,
PARSE_REQ_FAIL = 4 };
AsyncWebServerRequest::AsyncWebServerRequest(AsyncWebServer* s, AsyncClient* c)
: _client(c), _server(s), _handler(NULL), _response(NULL), _temp(), _parseState(0), _version(0), _method(HTTP_ANY), _url(), _host(), _contentType(), _boundary(), _authorization(), _reqconntype(RCT_HTTP), _authMethod(AsyncAuthType::AUTH_NONE), _isMultipart(false), _isPlainPost(false), _expectingContinue(false), _contentLength(0), _parsedLength(0), _multiParseState(0), _boundaryPosition(0), _itemStartIndex(0), _itemSize(0), _itemName(), _itemFilename(), _itemType(), _itemValue(), _itemBuffer(0), _itemBufferIndex(0), _itemIsFile(false), _tempObject(NULL) {
: _client(c), _server(s), _handler(NULL), _response(NULL), _temp(), _parseState(PARSE_REQ_START), _version(0), _method(HTTP_ANY), _url(), _host(), _contentType(), _boundary(), _authorization(), _reqconntype(RCT_HTTP), _authMethod(AsyncAuthType::AUTH_NONE), _isMultipart(false), _isPlainPost(false), _expectingContinue(false), _contentLength(0), _parsedLength(0), _multiParseState(0), _boundaryPosition(0), _itemStartIndex(0), _itemSize(0), _itemName(), _itemFilename(), _itemType(), _itemValue(), _itemBuffer(0), _itemBufferIndex(0), _itemIsFile(false), _tempObject(NULL) {
c->onError([](void* r, AsyncClient* c, int8_t error) { (void)c; AsyncWebServerRequest *req = (AsyncWebServerRequest*)r; req->_onError(error); }, this);
c->onAck([](void* r, AsyncClient* c, size_t len, uint32_t time) { (void)c; AsyncWebServerRequest *req = (AsyncWebServerRequest*)r; req->_onAck(len, time); }, this);
c->onDisconnect([](void* r, AsyncClient* c) { AsyncWebServerRequest *req = (AsyncWebServerRequest*)r; req->_onDisconnect(); delete c; }, this);
@ -67,6 +67,18 @@ AsyncWebServerRequest::~AsyncWebServerRequest() {
}
void AsyncWebServerRequest::_onData(void* buf, size_t len) {
// SSL/TLS handshake detection
#ifndef ASYNC_TCP_SSL_ENABLED
if (_parseState == PARSE_REQ_START && len && ((uint8_t*)buf)[0] == 0x16) { // 0x16 indicates a Handshake message (SSL/TLS).
#ifdef ESP32
log_d("SSL/TLS handshake detected: resetting connection");
#endif
_parseState = PARSE_REQ_FAIL;
_client->abort();
return;
}
#endif
size_t i = 0;
while (true) {
@ -74,6 +86,12 @@ void AsyncWebServerRequest::_onData(void* buf, size_t len) {
// Find new line in buf
char* str = (char*)buf;
for (i = 0; i < len; i++) {
// Check for null characters in header
if (!str[i]) {
_parseState = PARSE_REQ_FAIL;
_client->abort();
return;
}
if (str[i] == '\n') {
break;
}
@ -142,6 +160,8 @@ void AsyncWebServerRequest::_onData(void* buf, size_t len) {
if (!_sent) {
if (!_response)
send(501, T_text_plain, "Handler did not handle the request");
else if (!_response->_sourceValid())
send(500, T_text_plain, "Invalid data in handler");
_client->setRxTimeout(0);
_response->_respond(this);
_sent = true;
@ -246,6 +266,8 @@ bool AsyncWebServerRequest::_parseReqHead() {
_method = HTTP_HEAD;
} else if (m == T_OPTIONS) {
_method = HTTP_OPTIONS;
} else {
return false;
}
String g;
@ -257,6 +279,9 @@ bool AsyncWebServerRequest::_parseReqHead() {
_url = urlDecode(u);
_addGetParams(g);
if (!_url.length())
return false;
if (!_temp.startsWith(T_HTTP_1_0))
_version = 1;
@ -564,10 +589,14 @@ void AsyncWebServerRequest::_parseLine() {
if (_parseState == PARSE_REQ_START) {
if (!_temp.length()) {
_parseState = PARSE_REQ_FAIL;
_client->close();
_client->abort();
} else {
_parseReqHead();
_parseState = PARSE_REQ_HEADERS;
if (_parseReqHead()) {
_parseState = PARSE_REQ_HEADERS;
} else {
_parseState = PARSE_REQ_FAIL;
_client->abort();
}
}
return;
}
@ -589,6 +618,8 @@ void AsyncWebServerRequest::_parseLine() {
if (!_sent) {
if (!_response)
send(501, T_text_plain, "Handler did not handle the request");
else if (!_response->_sourceValid())
send(500, T_text_plain, "Invalid data in handler");
_client->setRxTimeout(0);
_response->_respond(this);
_sent = true;
@ -765,14 +796,6 @@ void AsyncWebServerRequest::send(AsyncWebServerResponse* response) {
if (_response)
delete _response;
_response = response;
if (_response == NULL) {
_client->close(true);
_onDisconnect();
_sent = true;
return;
}
if (!_response->_sourceValid())
send(500);
}
void AsyncWebServerRequest::redirect(const char* url, int code) {

4
src/WebResponseImpl.h
View File

@ -47,6 +47,10 @@ class AsyncBasicResponse : public AsyncWebServerResponse {
class AsyncAbstractResponse : public AsyncWebServerResponse {
private:
// amount of responce data in-flight, i.e. sent, but not acked yet
size_t _in_flight{0};
// in-flight queue credits
size_t _in_flight_credit{2};
String _head;
// Data is inserted into cache at begin().
// This is inefficient with vector, but if we use some other container,

31
src/WebResponses.cpp
View File

@ -352,7 +352,21 @@ size_t AsyncAbstractResponse::_ack(AsyncWebServerRequest* request, size_t len, u
request->client()->close();
return 0;
}
// return a credit for each chunk of acked data (polls does not give any credits)
if (len)
++_in_flight_credit;
// for chunked responses ignore acks if there are no _in_flight_credits left
if (_chunked && !_in_flight_credit) {
#ifdef ESP32
log_d("(chunk) out of in-flight credits");
#endif
return 0;
}
_ackedLength += len;
_in_flight -= (_in_flight > len) ? len : _in_flight;
// get the size of available sock space
size_t space = request->client()->space();
size_t headLen = _head.length();
@ -364,16 +378,31 @@ size_t AsyncAbstractResponse::_ack(AsyncWebServerRequest* request, size_t len, u
String out = _head.substring(0, space);
_head = _head.substring(space);
_writtenLength += request->client()->write(out.c_str(), out.length());
_in_flight += out.length();
--_in_flight_credit; // take a credit
return out.length();
}
}
if (_state == RESPONSE_CONTENT) {
// for response data we need to control the queue and in-flight fragmentation. Sending small chunks could give low latency,
// but flood asynctcp's queue and fragment socket buffer space for large responses.
// Let's ignore polled acks and acks in case when we have more in-flight data then the available socket buff space.
// That way we could balance on having half the buffer in-flight while another half is filling up, while minimizing events in asynctcp q
if (_in_flight > space) {
// log_d("defer user call %u/%u", _in_flight, space);
// take the credit back since we are ignoring this ack and rely on other inflight data
if (len)
--_in_flight_credit;
return 0;
}
size_t outLen;
if (_chunked) {
if (space <= 8) {
return 0;
}
outLen = space;
} else if (!_sendContentLength) {
outLen = space;
@ -422,6 +451,8 @@ size_t AsyncAbstractResponse::_ack(AsyncWebServerRequest* request, size_t len, u
if (outLen) {
_writtenLength += request->client()->write((const char*)buf, outLen);
_in_flight += outLen;
--_in_flight_credit; // take a credit
}
if (_chunked) {

3
src/WebServer.cpp
View File

@ -56,8 +56,7 @@ AsyncWebServer::AsyncWebServer(uint16_t port)
c->setRxTimeout(3);
AsyncWebServerRequest* r = new AsyncWebServerRequest((AsyncWebServer*)s, c);
if (r == NULL) {
c->close(true);
c->free();
c->abort();
delete c;
}
},

1
src/literals.h
View File

@ -65,6 +65,7 @@ namespace asyncsrv {
static constexpr const char* T_response = "response";
static constexpr const char* T_retry_ = "retry: ";
static constexpr const char* T_retry_after = "retry-after";
static constexpr const char* T_nn = "\n\n";
static constexpr const char* T_rn = "\r\n";
static constexpr const char* T_rnrn = "\r\n\r\n";
static constexpr const char* T_Transfer_Encoding = "transfer-encoding";