AsyncEventSource.cpp

// SPDX-License-Identifier: LGPL-3.0-or-later
// Copyright 2016-2026 Hristo Gochkov, Mathieu Carbou, Emil Muratov, Will Miles

#include "AsyncEventSource.h"
#include "AsyncWebServerLogging.h"

#include <algorithm>
#include <memory>
#include <utility>

#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 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 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;
}

// Message

size_t AsyncEventSourceMessage::ack(size_t len, __attribute__((unused)) uint32_t time) {
  // 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;
}

size_t AsyncEventSourceMessage::write(AsyncClient *client) {
  if (!client) {
    return 0;
  }

  if (_sent >= _data->length() || !client->canSend()) {
    return 0;
  }

  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) {
  size_t sent = write(client);
  return sent && client->send() ? sent : 0;
}

// Client

AsyncEventSourceClient::AsyncEventSourceClient(AsyncWebServerRequest *request, AsyncEventSource *server) : _client(request->clientRelease()), _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;
      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);
  // delete AsyncWebServerRequest object (and bound response) since we have the ownership on client connection now
  delete request;
}

AsyncEventSourceClient::~AsyncEventSourceClient() {
#ifdef ESP32
  // Protect message queue access (size checks and modifications) which is not thread-safe.
  std::lock_guard<std::recursive_mutex> lock(_lockmq);
#endif
  _messageQueue.clear();
  close();
}

bool AsyncEventSourceClient::_queueMessage(const char *message, size_t len) {
#ifdef ESP32
  // Protect message queue access (size checks and modifications) which is not thread-safe.
  std::lock_guard<std::recursive_mutex> lock(_lockmq);
#endif

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

  /*
    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 (_client && _client->canSend() && _messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2) {
    _runQueue();
  }

  return true;
}

bool AsyncEventSourceClient::_queueMessage(AsyncEvent_SharedData_t &&msg) {
#ifdef ESP32
  // Protect message queue access (size checks and modifications) which is not thread-safe.
  std::lock_guard<std::recursive_mutex> lock(_lockmq);
#endif

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

  /*
    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 (_client && _client->canSend() && _messageQueue.size() < SSE_MAX_QUEUED_MESSAGES >> 2) {
    _runQueue();
  }
  return true;
}

void AsyncEventSourceClient::_onAck(size_t len __attribute__((unused)), uint32_t time __attribute__((unused))) {
#ifdef ESP32
  // Protect message queue access (size checks and modifications) which is not thread-safe.
  std::lock_guard<std::recursive_mutex> lock(_lockmq);
#endif

  // 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
  // Protect message queue access (size checks and modifications) which is not thread-safe.
  std::lock_guard<std::recursive_mutex> lock(_lockmq);
#endif
  if (_messageQueue.size()) {
    _runQueue();
  }
}

void AsyncEventSourceClient::_onTimeout(uint32_t time __attribute__((unused))) {
  if (_client) {
    _client->close();
  }
}

void AsyncEventSourceClient::_onDisconnect() {
  if (!_client) {
    return;
  }
  _client = nullptr;
  _server->_handleDisconnect(this);
}

void AsyncEventSourceClient::close() {
  if (_client) {
    _client->close();
  }
}

bool AsyncEventSourceClient::send(const char *message, const char *event, uint32_t id, uint32_t reconnect) {
  if (!connected()) {
    return false;
  }
  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;
      _inflight += bytes_written;
      if (bytes_written == 0 || _inflight > _max_inflight) {
        // Serial.print("_");
        break;
      }
    }
  }

  // flush socket
  if (_client && total_bytes_written) {
    _client->send();
  }
}

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) {
  AsyncAuthorizationMiddleware *m = new AsyncAuthorizationMiddleware(401, cb);
  m->_freeOnRemoval = true;
  addMiddleware(m);
}

void AsyncEventSource::_addClient(AsyncEventSourceClient *client) {
  if (!client) {
    return;
  }
#ifdef ESP32
  std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
#endif
  _clients.emplace_back(client);
  if (_connectcb) {
    _connectcb(client);
  }

  _adjust_inflight_window();
}

void AsyncEventSource::_handleDisconnect(AsyncEventSourceClient *client) {
#ifdef ESP32
  std::lock_guard<std::recursive_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);
      break;
    }
  }
  _adjust_inflight_window();
}

void AsyncEventSource::close() {
  // 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
#ifdef ESP32
  std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
#endif
  for (const auto &c : _clients) {
    if (c->connected()) {
      /**
       * @brief: Fix self-deadlock by using recursive_mutex instead.
       * Due to c->close() shall call the callback function _onDisconnect()
       * The calling flow _onDisconnect() --> _handleDisconnect() --> deadlock
      */
      c->close();
    }
  }
}

// pmb fix
size_t AsyncEventSource::avgPacketsWaiting() const {
  size_t aql = 0;
  uint32_t nConnectedClients = 0;
#ifdef ESP32
  std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
#endif
  for (const auto &c : _clients) {
    if (c->connected()) {
      aql += c->packetsWaiting();
      ++nConnectedClients;
    }
  }
  return nConnectedClients == 0 ? 0 : ((aql) + (nConnectedClients / 2)) / (nConnectedClients);  // round up
}

AsyncEventSource::SendStatus AsyncEventSource::send(const char *message, const char *event, uint32_t id, uint32_t reconnect) {
  AsyncEvent_SharedData_t shared_msg = std::make_shared<String>(generateEventMessage(message, event, id, reconnect));
#ifdef ESP32
  std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
#endif
  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);
}

size_t AsyncEventSource::count() const {
#ifdef ESP32
  std::lock_guard<std::recursive_mutex> lock(_client_queue_lock);
#endif
  size_t n_clients{0};
  for (const auto &i : _clients) {
    if (i->connected()) {
      ++n_clients;
    }
  }

  return n_clients;
}

bool AsyncEventSource::canHandle(AsyncWebServerRequest *request) const {
  return request->isSSE() && request->url().equals(_url);
}

void AsyncEventSource::handleRequest(AsyncWebServerRequest *request) {
  request->send(new AsyncEventSourceResponse(this));
}

// list iteration protected by caller's lock
void AsyncEventSource::_adjust_inflight_window() {
  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);
  }
}

/*  Response  */

AsyncEventSourceResponse::AsyncEventSourceResponse(AsyncEventSource *server) : _server(server) {
  _code = 200;
  _contentType = T_text_event_stream;
  _sendContentLength = false;
  addHeader(T_Cache_Control, T_no_cache);
  addHeader(T_Connection, T_keep_alive);
}

void AsyncEventSourceResponse::_respond(AsyncWebServerRequest *request) {
  String out;
  _assembleHead(out, request->version());
  // unbind client's onAck callback from AsyncWebServerRequest's, we will destroy it on next callback and steal the client,
  // can't do it now 'cause now we are in AsyncWebServerRequest::_onAck 's stack actually
  // here we are loosing time on one RTT delay, but with current design we can't get rid of Req/Resp objects other way
  _request = request;
  request->client()->onAck(
    [](void *r, AsyncClient *c, size_t len, uint32_t time) {
      if (len) {
        static_cast<AsyncEventSourceResponse *>(r)->_switchClient();
      }
    },
    this
  );
  request->client()->write(out.c_str(), _headLength);
  _state = RESPONSE_WAIT_ACK;
}

void AsyncEventSourceResponse::_switchClient() {
  // AsyncEventSourceClient c-tor will take the ownership of AsyncTCP's client connection
  new AsyncEventSourceClient(_request, _server);
  // AsyncEventSourceClient c-tor would also delete _request and *this
};