String str;

size_t len{0};

if (message) {

len += strlen(message);

}

if (event) {

len += strlen(event);

}

len += 42; // give it some overhead

if (!str.reserve(len)) {

async_ws_log_e("Failed to allocate");

return asyncsrv::emptyString;

}

if (reconnect) {

str += T_retry_;

str += reconnect;

str += ASYNC_SSE_NEW_LINE_CHAR; // '\n'

}

if (id) {

str += T_id__;

str += id;

str += ASYNC_SSE_NEW_LINE_CHAR; // '\n'

}

if (event != NULL) {

str += T_event_;

str += event;

str += ASYNC_SSE_NEW_LINE_CHAR; // '\n'

}

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 sequence

lineEnd = nextR;

nextLine = nextN + 1;

} else {

// some abnormal \n \r mixed sequence

lineEnd = std::min(nextR, nextN);

nextLine = lineEnd + 1;

}

} else if (nextN != NULL) { // Unix/Mac OS X LF

lineEnd = nextN;

nextLine = nextN + 1;

} else { // some ancient garbage

lineEnd = nextR;

nextLine = nextR + 1;

}

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;

// If the whole message is now acked...

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 - _data->length();

_acked = _data->length();

return extra;

}

// Return that no extra bytes left.

_acked += len;

return 0;

if (!client) {

return 0;

}

if (_sent >= _data->length() || !client->canSend()) {

return 0;

}

size_t len = std::min(_data->length() - _sent, client->space());

/*

size_t written = client->add(_data->c_str() + _sent, len, ASYNC_WRITE_FLAG_COPY); // ASYNC_WRITE_FLAG_MORE

_sent += written;

return written;

: _client(client), _server(server), _lastId(lastId) {

_client->setRxTimeout(0);

_client->onError(NULL, NULL);

_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) {

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);

_client->setNoDelay(true);

// Protect message queue access (size checks and modifications) which is not thread-safe.

asyncsrv::lock_guard_type lock(_lockmq);

if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) {

async_ws_log_w("Event message queue overflow: discard message");

return false;

}

if (_client) {

_messageQueue.emplace_back(message, len);

} else {

_messageQueue.clear();

return false;

}

/*

if (_client && _client->canSend() && _messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2) {

_runQueue();

}

return true;

// Protect message queue access (size checks and modifications) which is not thread-safe.

asyncsrv::lock_guard_type lock(_lockmq);

if (_messageQueue.size() >= SSE_MAX_QUEUED_MESSAGES) {

async_ws_log_w("Event message queue overflow: discard message");

return false;

}

if (_client) {

_messageQueue.emplace_back(std::move(msg));

} else {

_messageQueue.clear();

return false;

}

/*

if (_client && _client->canSend() && _messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2) {

_runQueue();

}

return true;

// Protect message queue access (size checks and modifications) which is not thread-safe.

asyncsrv::lock_guard_type lock(_lockmq);

// 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();

}

// Protect message queue access (size checks and modifications) which is not thread-safe.

asyncsrv::lock_guard_type lock(_lockmq);

if (_messageQueue.size()) {

_runQueue();

}

if (_client) {

_client->close();

}

if (!_client) {

return;

}

_client = nullptr;

_server->_handleDisconnect(this);

if (_client) {

_client->close();

}

if (!connected()) {

return false;

}

return _queueMessage(std::make_shared<String>(generateEventMessage(message, event, id, reconnect)));

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;

_inflight += bytes_written;

if (bytes_written == 0 || _inflight > _max_inflight) {

// Serial.print("_");

break;

}

}

}

// flush socket

if (_client && total_bytes_written) {

_client->send();

}

if (value >= SSE_MIN_INFLIGH && value <= SSE_MAX_INFLIGH) {

_max_inflight = value;

}

AsyncAuthorizationMiddleware *m = new AsyncAuthorizationMiddleware(401, cb);

m->_freeOnRemoval = true;

addMiddleware(m);

if (!client) {

return;

}

if (_connectcb) {

_connectcb(client);

}

asyncsrv::lock_guard_type lock(_client_queue_lock);

_clients.emplace_back(client);

_adjust_inflight_window();

if (_disconnectcb) {

_disconnectcb(client);

}

asyncsrv::lock_guard_type lock(_client_queue_lock);

for (auto i = _clients.begin(); i != _clients.end(); ++i) {

if (i->get() == client) {

_clients.erase(i);

break;

}

}

_adjust_inflight_window();

// While the whole loop is not done, the linked list is locked and so the

// iterator should remain valid even when AsyncEventSource::_handleDisconnect()

// is called very early

asyncsrv::lock_guard_type lock(_client_queue_lock);

for (const auto &c : _clients) {

if (c->connected()) {

/**

c->close();

}

}

size_t aql = 0;

uint32_t nConnectedClients = 0;

asyncsrv::lock_guard_type lock(_client_queue_lock);

for (const auto &c : _clients) {

if (c->connected()) {

aql += c->packetsWaiting();

++nConnectedClients;

}

}

return nConnectedClients == 0 ? 0 : ((aql) + (nConnectedClients / 2)) / (nConnectedClients); // round up

AsyncEvent_SharedData_t shared_msg = std::make_shared<String>(generateEventMessage(message, event, id, reconnect));

asyncsrv::lock_guard_type lock(_client_queue_lock);

size_t hits = 0;

size_t miss = 0;

for (const auto &c : _clients) {

if (c->connected()) {

if (c->write(shared_msg)) {

++hits;

} else {

++miss;

}

}

}

return hits == 0 ? DISCARDED : (miss == 0 ? ENQUEUED : PARTIALLY_ENQUEUED);

asyncsrv::lock_guard_type lock(_client_queue_lock);

size_t n_clients{0};

for (const auto &i : _clients) {

if (i->connected()) {

++n_clients;

}

}

return n_clients;

const size_t clientCount = count();

if (clientCount) {

size_t inflight = SSE_MAX_INFLIGH / clientCount;

for (const auto &c : _clients) {

if (c->connected()) {

c->set_max_inflight_bytes(inflight);

}

}

// Serial.printf("adjusted inflight to: %u\n", inflight);

}

_code = 200;

_contentType = T_text_event_stream;

_sendContentLength = false;

addHeader(T_Cache_Control, T_no_cache);

addHeader(T_Connection, T_keep_alive);

String out;

_assembleHead(out, request->version());

uint32_t lastId = 0;

if (request->hasHeader(T_Last_Event_ID)) {

lastId = strtoul(request->getHeader(T_Last_Event_ID)->value().c_str(), nullptr, 10);

}

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);