AsyncWSocket keepalive pongs

 - keepalive pongs implementation for Client class
 - message size/q cap for WSocketServer class

Author: vortigont

src/AsyncWSocket.cpp

@@ -152,17 +152,18 @@ WSocketClient::WSocketClient(uint32_t id, AsyncWebServerRequest *request, WSocke
   _max_msgsize(msgsize),
   _max_qcap(qcapsize)
 {
+  _lastPong = millis();
   // disable connection timeout
   _client->setRxTimeout(0);
   _client->setNoDelay(true);
   // set AsyncTCP callbacks
   _client->onAck( [](void *r, AsyncClient *c, size_t len, uint32_t rtt) { (void)c; reinterpret_cast<WSocketClient*>(r)->_clientSend(len); }, this );
   //_client->onAck( [](void *r, AsyncClient *c, size_t len, uint32_t rtt) { (void)c; reinterpret_cast<WSocketClient*>(r)->_onAck(len, rtt); }, this );
-  _client->onDisconnect( [](void *r, AsyncClient *c) { reinterpret_cast<WSocketClient*>(r)->_onDisconnect(c); }, this );
-  _client->onTimeout( [](void *r, AsyncClient *c, uint32_t time) { (void)c; reinterpret_cast<WSocketClient*>(r)->_onTimeout(time); }, this );
-  _client->onData( [](void *r, AsyncClient *c, void *buf, size_t len) { (void)c; reinterpret_cast<WSocketClient*>(r)->_onData(buf, len); }, this );
-  _client->onPoll( [](void *r, AsyncClient *c) { (void)c; reinterpret_cast<WSocketClient*>(r)->_clientSend(); }, this );
-  _client->onError( [](void *r, AsyncClient *c, int8_t error) { (void)c; log_e("err:%d", error); }, this );
+  _client->onDisconnect(  [](void *r, AsyncClient *c) { reinterpret_cast<WSocketClient*>(r)->_onDisconnect(c); }, this );
+  _client->onTimeout(     [](void *r, AsyncClient *c, uint32_t time) { (void)c; reinterpret_cast<WSocketClient*>(r)->_onTimeout(time); }, this );
+  _client->onData(        [](void *r, AsyncClient *c, void *buf, size_t len) { (void)c; reinterpret_cast<WSocketClient*>(r)->_onData(buf, len); }, this );
+  _client->onPoll(        [](void *r, AsyncClient *c) { (void)c; reinterpret_cast<WSocketClient*>(r)->_keepalive(); reinterpret_cast<WSocketClient*>(r)->_clientSend(); }, this );
+  _client->onError(       [](void *r, AsyncClient *c, int8_t error) { (void)c; log_e("err:%d", error); }, this );
   delete request;
 }
 
@@ -652,6 +653,13 @@ void WSocketClient::_sendEvent(event_t e){
     _cb(this, e);
 }
 
+void WSocketClient::_keepalive(){
+  if (millis() - _lastPong > _keepAlivePeriod){
+    enqueueMessage(std::make_shared< WSMessageContainer<std::string> >(WSFrameType_t::pong, true, "WSocketClient Pong" ));
+    _lastPong = millis();
+  }
+}
+
 
 // ***** WSocketServer implementation *****
 
@@ -666,10 +674,11 @@ bool WSocketServer::newClient(AsyncWebServerRequest *request){
       if (eventHandler)
         eventHandler(c, e);
       else
-        c->dequeueMessage(); }  // silently discard incoming messages when there is no callback set
-    );
+        c->dequeueMessage(); },   // silently discard incoming messages when there is no callback set
+      msgsize, qcap);
   }
   _clients.back().setOverflowPolicy(_overflow_policy);
+  _clients.back().setKeepALive(_keepAlivePeriod);
   if (eventHandler) eventHandler(&_clients.back(), WSocketClient::event_t::connect);
   return true;
 }
@@ -742,6 +751,13 @@ WSocketServer::msgall_err_t WSocketServer::pingAll(const char *data, size_t len)
   return cnt == _clients.size() ? msgall_err_t::ok : msgall_err_t::partial;
 }
 
+WSocketClient::err_t WSocketServer::message(uint32_t id, WSMessagePtr m){
+if (WSocketClient *c = _getClient(id))
+  return c->enqueueMessage(std::move(m));
+else
+  return WSocketClient::err_t::disconnected;
+}
+
 WSocketServer::msgall_err_t WSocketServer::messageAll(WSMessagePtr m){
   size_t cnt{0};
   for (auto &c : _clients) {
@@ -781,12 +797,13 @@ bool WSocketServerWorker::newClient(AsyncWebServerRequest *request){
     #ifdef ESP32
     std::lock_guard<std::mutex> lock (clientslock);
     #endif
-    _clients.emplace_back(getNextId(), request, [this](WSocketClient *c, WSocketClient::event_t e){ if (_task_hndlr) xTaskNotifyGive(_task_hndlr); });
+    _clients.emplace_back(getNextId(), request, [this](WSocketClient *c, WSocketClient::event_t e){ if (_task_hndlr) xTaskNotifyGive(_task_hndlr); }, msgsize, qcap);
   }
 
   // create events group where we'll pick events
   _clients.back().createEventGroupHandle();
   _clients.back().setOverflowPolicy(getOverflowPolicy());
+  _clients.back().setKeepALive(_keepAlivePeriod);
   xEventGroupSetBits(_clients.back().getEventGroupHandle(), enum2uint32(WSocketClient::event_t::connect));
   if (_task_hndlr)
     xTaskNotifyGive(_task_hndlr);

src/AsyncWSocket.h

@@ -211,7 +211,7 @@ class WSMessageClose : public WSMessageContainer<std::string> {
   WSMessageClose (uint16_t status, Args&&... args) : WSMessageContainer<std::string>(WSFrameType_t::close, true, std::forward<Args>(args)...), _status_code(status) {
     // convert code to message body
     uint16_t buff = htons (status);
-    container.append((char*)(&buff), 2);
+    container.insert(0, (char*)(&buff), 2);
   };
 
   uint16_t getStatusCode() const override { return _status_code; }
@@ -379,6 +379,18 @@ class WSocketClient {
   // check if client can enqueue and send new messages
   err_t canSend() const;
 
+  /**
+   * @brief Set the WebSOcket ping Keep A Live
+   * if set, client will send pong packet it's peer periodically to keep the connection alive
+   * ping does not require a reply from peer
+   * 
+   * @param seconds 
+   */
+  void setKeepALive(size_t seconds){ _keepAlivePeriod = seconds * 1000; };
+
+  // get keepalive value
+  size_t getKeepALive() const { return _keepAlivePeriod / 1000; };
+
   /**
    * @brief access Event Group Handle for the client
    * 
@@ -422,6 +434,9 @@ class WSocketClient {
   // in-flight data credits
   size_t _in_flight_credit{WS_IN_FLIGHT_CREDITS};
 
+  // keepalive
+  unsigned long _keepAlivePeriod{0}, _lastPong;
+
   /**
    * @brief go through out Q and send message data ()
    * @note this method will grab a mutex lock on outQ internally
@@ -451,6 +466,8 @@ class WSocketClient {
    */
   void _sendEvent(event_t e);
 
+  void _keepalive();
+
   // AsyncTCP callbacks
   void _onTimeout(uint32_t time);
   void _onDisconnect(AsyncClient *c);
@@ -459,8 +476,8 @@ class WSocketClient {
 
 /**
  * @brief WebServer Handler implementation that plays the role of a WebSocket server
- * it inherits behavior of original WebSocket server and uses AsyncTCP callback
- * to handle incoming messages
+ * it inherits behavior of original WebSocket server and uses AsyncTCP's thread to
+ * run callbacks on incoming messages
  * 
  */
 class WSocketServer : public AsyncWebHandler {
@@ -472,9 +489,34 @@ class WSocketServer : public AsyncWebHandler {
     none            // no clients or all outbound queues are full, message discarded
   };
 
-  explicit WSocketServer(const char* url, WSocketClient::event_cb_t handler = {}) : _url(url), eventHandler(handler) {}
+  explicit WSocketServer(const char* url, WSocketClient::event_cb_t handler = {}, size_t msgsize = 8 * 1024, size_t qcap = 4) : _url(url), eventHandler(handler) {}
   ~WSocketServer() = default;
 
+  /**
+   * @copydoc WSClient::setOverflowPolicy(overflow_t policy)
+   */
+  void setOverflowPolicy(WSocketClient::overflow_t policy){ _overflow_policy = policy; }
+  WSocketClient::overflow_t getOverflowPolicy() const { return _overflow_policy; }
+
+  /**
+   * @brief Set Message Size limit for the incoming messages
+   * if peer tries to send us message larger then defined limit size,
+   * the message will be discarded and peer's connection would be closed with respective error code
+   * @note only new connections would be affected with changed value
+   * 
+   * @param size 
+   */
+  void setMaxMessageSize(size_t size){ msgsize = size; }
+  size_t getMaxMessageSize(size_t size) const { return msgsize; }
+
+  /**
+   * @brief Set in/out Message Queue Size
+   * 
+   * @param size 
+   */
+  void setMessageQueueSize(size_t size){ qcap = size; }
+  size_t getMessageQueueSize(size_t size){ return qcap; }
+
   /**
    * @brief check if client with specified id can accept new message for sending
    * 
@@ -498,12 +540,6 @@ class WSocketServer : public AsyncWebHandler {
     return _getClient(id) != nullptr;
   }
 
-  /**
-   * @copydoc WSClient::setOverflowPolicy(overflow_t policy)
-   */
-  void setOverflowPolicy(WSocketClient::overflow_t policy){ _overflow_policy = policy; }
-  WSocketClient::overflow_t getOverflowPolicy() const { return _overflow_policy; }
-
   /**
    * @brief disconnect client
    * 
@@ -549,21 +585,30 @@ class WSocketServer : public AsyncWebHandler {
   msgall_err_t pingAll(const char *data = NULL, size_t len = 0);
 
   /**
-   * @brief send message to specific client
+   * @brief Set the WebSocket client Keep A Live
+   * if set, server will pong it's peers periodically to keep connections alive
+   * @note it does not check for replies and it's validity, it only sends messages to
+   * help keep TCP connection alive through firewalls/routers
+   * 
+   * @param seconds 
+   */
+  void setKeepALive(size_t seconds){ _keepAlivePeriod = seconds; };
+
+  // get keepalive value
+  size_t getKeepALive() const { return _keepAlivePeriod; };
+
+
+  /**
+   * @brief send generic message to specific client
    * 
    * @param id 
    * @param m 
    * @return WSocketClient::err_t 
    */
-  WSocketClient::err_t message(uint32_t id, WSMessagePtr m){
-    if (WSocketClient *c = _getClient(id))
-      return c->enqueueMessage(std::move(m));
-    else
-      return WSocketClient::err_t::disconnected;
-  }
+  WSocketClient::err_t message(uint32_t id, WSMessagePtr m);
 
   /**
-   * @brief send message to all available clients
+   * @brief send genric message to all available clients
    * 
    * @param m 
    * @return msgall_err_t 
@@ -629,6 +674,11 @@ class WSocketServer : public AsyncWebHandler {
   #endif
   // WSocketClient events handler
   WSocketClient::event_cb_t eventHandler;
+  unsigned long _keepAlivePeriod{0};
+  // max message size
+  size_t msgsize;
+  // client's queue capacity
+  size_t qcap;
 
   // return next available client's ID
   uint32_t getNextId() {