Bugfix: AsyncEventSource writes multiple events per tcp send; Improve…

…ment: don't hold onto event items until ack, immediately remove them from queue

Ref: esphome#41

src/AsyncEventSource.cpp

@@ -123,8 +123,7 @@ AsyncEventSourceMessage::~AsyncEventSourceMessage() {
         free(_data);
 }
 
-size_t AsyncEventSourceMessage::ack(size_t len, uint32_t time) {
-  (void)time;
+size_t AsyncEventSourceMessage::ack(size_t len) {
   // If the whole message is now acked...
   if(_acked + len > _len){
      // Return the number of extra bytes acked (they will be carried on to the next message)
@@ -137,19 +136,24 @@ size_t AsyncEventSourceMessage::ack(size_t len, uint32_t time) {
   return 0;
 }
 
-size_t AsyncEventSourceMessage::send(AsyncClient *client) {
+size_t AsyncEventSourceMessage::write_buffer(AsyncClient *client) {
   if (!client->canSend())
     return 0;
   const size_t len = _len - _sent;
   if(client->space() < len){
     return 0;
   }
   size_t sent = client->add((const char *)_data, len);
-  client->send();
   _sent += sent;
   return sent;
 }
 
+size_t AsyncEventSourceMessage::send(AsyncClient *client) {
+  size_t sent = write_buffer(client);
+  client->send();
+  return sent;
+}
+
 // Client
 
 AsyncEventSourceClient::AsyncEventSourceClient(AsyncWebServerRequest *request, AsyncEventSource *server)
@@ -171,6 +175,8 @@ AsyncEventSourceClient::AsyncEventSourceClient(AsyncWebServerRequest *request, A
 
   _server->_addClient(this);
   delete request;
+
+  _client->setNoDelay(true);
 }
 
 AsyncEventSourceClient::~AsyncEventSourceClient(){
@@ -185,8 +191,9 @@ void AsyncEventSourceClient::_queueMessage(AsyncEventSourceMessage *dataMessage)
     delete dataMessage;
     return;
   }
+
   if(_messageQueue.length() >= SSE_MAX_QUEUED_MESSAGES){
-      ets_printf("ERROR: Too many messages queued\n");
+      ets_printf("AsyncEventSourceClient: ERROR: Queue is full, communications too slow, dropping event");
       delete dataMessage;
   } else {
       _messageQueue.add(dataMessage);
@@ -196,12 +203,6 @@ void AsyncEventSourceClient::_queueMessage(AsyncEventSourceMessage *dataMessage)
 }
 
 void AsyncEventSourceClient::_onAck(size_t len, uint32_t time){
-  while(len && !_messageQueue.isEmpty()){
-    len = _messageQueue.front()->ack(len, time);
-    if(_messageQueue.front()->finished())
-      _messageQueue.remove(_messageQueue.front());
-  }
-
   _runQueue();
 }
 
@@ -247,14 +248,33 @@ void AsyncEventSourceClient::_runQueue(){
   this->_messageQueue_processing = true;
 #endif // ESP32
 
-  while(!_messageQueue.isEmpty() && _messageQueue.front()->finished()){
-    _messageQueue.remove(_messageQueue.front());
-  }
-
+  size_t total_bytes_written = 0;
   for(auto i = _messageQueue.begin(); i != _messageQueue.end(); ++i)
   {
-    if(!(*i)->sent())
-      (*i)->send(_client);
+    if(!(*i)->sent()) {
+      size_t bytes_written = (*i)->write_buffer(_client);
+      total_bytes_written += bytes_written;
+      if(bytes_written == 0)
+        break;
+      // todo: there is a further optimization to write a partial event to squeeze the last few bytes into the outgoing tcp send buffer, in 
+      //           fact all of this code is already set up to do so, it's only write_buffer that needs to be updated to allow it instead of 
+      //           returning zero when the full event won't fit into what's left of the buffer
+      // todo: windows is taking 40-50ms to send an ack back while it waits for more data which won't come since this code must wait for ack first 
+      //           due to system resource limitations - if the dashboard javascript just sends a single byte back per event received (which this
+      //           code would of course throw away as meaningless) then windows (or whatever other host runs the webbrower) will piggyback an ack
+      //           onto that outgoing packet for us, reducing roundtrip ack latency and potentially as much as trippling throughput again 
+      //           (measured: ESP-01: 20ms to send another packet after ack received, windows: 40-50ms to ack after receiving a packet)
+    }
+  }
+  if(total_bytes_written > 0)
+    _client->send();
+
+  size_t len = total_bytes_written;
+  while(len && !_messageQueue.isEmpty()){
+    len = _messageQueue.front()->ack(len);
+    if(_messageQueue.front()->finished()){
+      _messageQueue.remove(_messageQueue.front());
+    }
   }
 
 #if defined(ESP32)

src/AsyncEventSource.h

@@ -68,7 +68,8 @@ class AsyncEventSourceMessage {
   public:
     AsyncEventSourceMessage(const char * data, size_t len);
     ~AsyncEventSourceMessage();
-    size_t ack(size_t len, uint32_t time __attribute__((unused)));
+    size_t ack(size_t len);
+    size_t write_buffer(AsyncClient *client);
     size_t send(AsyncClient *client);
     bool finished(){ return _acked == _len; }
     bool sent() { return _sent == _len; }