diff --git a/components/RaftWebServer/RaftWebConnDefs.h b/components/RaftWebServer/RaftWebConnDefs.h index 418b6e7..860fe9b 100644 --- a/components/RaftWebServer/RaftWebConnDefs.h +++ b/components/RaftWebServer/RaftWebConnDefs.h @@ -10,6 +10,7 @@ #include #include +#include "SpiramAwareAllocator.h" // Callback function for any endpoint enum RaftWebConnSendRetVal @@ -44,3 +45,7 @@ typedef std::function RaftWebConnReadyToSendFn; // Function to send on a connection typedef std::function RaftWebConnSendFn; +// Function to send a buffer on a connection, which may take the buffer's contents (leaving it empty) rather than +// copy what it can't send at once +typedef std::function RaftWebConnSendTakeFn; + diff --git a/components/RaftWebServer/RaftWebConnManager.cpp b/components/RaftWebServer/RaftWebConnManager.cpp index f8aca03..94d7b01 100644 --- a/components/RaftWebServer/RaftWebConnManager.cpp +++ b/components/RaftWebServer/RaftWebConnManager.cpp @@ -522,6 +522,34 @@ bool RaftWebConnManager::sendBufOnChannel(const uint8_t* pBuf, uint32_t bufLen, return sendOk; } +///////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// Send a buffer on a channel, which may be taken rather than copied +///////////////////////////////////////////////////////////////////////////////////////////////////////////////// + +bool RaftWebConnManager::sendBufOnChannel(SpiramAwareUint8Vector& buf, uint32_t channelID) +{ + // Main task only (the send path is not protected by locks) + RAFT_CHECK_MAIN_TASK(MODULE_PREFIX, "sendBufOnChannel"); + + // The buffer can only be taken by the responder if it is the only one on the channel + RaftWebResponder* pChannelResponder = nullptr; + uint32_t numOnChannel = 0; + for (uint32_t i = 0; i < _webConnections.size(); i++) + { + if (!_webConnections[i].isActive()) + continue; + RaftWebResponder *pResponder = _webConnections[i].getResponder(); + uint32_t usedChannelID = 0; + if (!pResponder || !pResponder->getChannelID(usedChannelID) || (usedChannelID != channelID)) + continue; + pChannelResponder = pResponder; + numOnChannel++; + } + if (numOnChannel != 1) + return sendBufOnChannel(buf.data(), buf.size(), channelID); + return pChannelResponder->encodeAndSendData(buf); +} + ///////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Send to all server-side events ///////////////////////////////////////////////////////////////////////////////////////////////////////////////// diff --git a/components/RaftWebServer/RaftWebConnManager.h b/components/RaftWebServer/RaftWebConnManager.h index e202e63..0ed52a2 100644 --- a/components/RaftWebServer/RaftWebConnManager.h +++ b/components/RaftWebServer/RaftWebConnManager.h @@ -76,6 +76,9 @@ class RaftWebConnManager // Send a buffer on a channel (main task only) bool sendBufOnChannel(const uint8_t* pBuf, uint32_t bufLen, uint32_t channelID); + // Send a buffer on a channel, which may be taken (leaving it empty) rather than copied + bool sendBufOnChannel(SpiramAwareUint8Vector& buf, uint32_t channelID); + // Send to all server-side events (main task only) void serverSideEventsSendMsg(const char* eventContent, const char* eventGroup); diff --git a/components/RaftWebServer/RaftWebConnection.cpp b/components/RaftWebServer/RaftWebConnection.cpp index 33a3b4d..ecbbe80 100644 --- a/components/RaftWebServer/RaftWebConnection.cpp +++ b/components/RaftWebServer/RaftWebConnection.cpp @@ -209,12 +209,12 @@ void RaftWebConnection::clear() _debugDataRxCount = 0; _maxSendBufferBytes = 0; #ifdef DEBUG_RESPONDER_CLEAR - if (_socketTxQueuedBuffer.size() > 0) + if (txQueuedLen() > 0) { - LOG_I(MODULE_PREFIX, "clear() clearing _socketTxQueuedBuffer with %d bytes", _socketTxQueuedBuffer.size()); + LOG_I(MODULE_PREFIX, "clear() clearing _socketTxQueuedBuffer with %d bytes", txQueuedLen()); } #endif - _socketTxQueuedBuffer.clear(); + txQueueClear(); _header.clear(); } @@ -657,6 +657,8 @@ bool RaftWebConnection::serviceConnHeader(const uint8_t* pRxData, uint32_t dataL std::bind(&RaftWebConnection::canSendOnConn, this), std::bind(&RaftWebConnection::rawSendOnConn, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), _pClientConn->getClientId()); + params.setWebConnRawSendTake( + std::bind(&RaftWebConnection::rawSendOnConnTake, this, std::placeholders::_1, std::placeholders::_2)); _pResponder = _pConnManager->getNewResponder(_header, params, statusCode); #ifdef DEBUG_RESPONDER_CREATE_DELETE if (_pResponder) @@ -811,7 +813,7 @@ bool RaftWebConnection::responderHandleData(const uint8_t* pRxData, uint32_t dat // Return indication of more to come - keep processing if not inactive connStatus = _pResponder->getConnStatus(); - if ((connStatus == CONN_INACTIVE) && (_socketTxQueuedBuffer.size() != 0)) + if ((connStatus == CONN_INACTIVE) && (txQueuedLen() != 0)) { return true; } @@ -1113,7 +1115,7 @@ RaftWebConnSendRetVal RaftWebConnection::canSendOnConn() #endif // Don't accept any more data while the buffer is not empty - if (_socketTxQueuedBuffer.size() != 0) + if (txQueuedLen() != 0) { return WEB_CONN_SEND_EAGAIN; } @@ -1135,7 +1137,7 @@ RaftWebConnSendRetVal RaftWebConnection::canSendOnConn() { LOG_I(MODULE_PREFIX, "canSendOnConn connId %d totalUs %d clientCanSendUs %d bufSize %d result %d", _pClientConn->getClientId(), totalUs, clientCanSendUs, - (int)_socketTxQueuedBuffer.size(), result); + (int)txQueuedLen(), result); } #endif @@ -1147,6 +1149,19 @@ RaftWebConnSendRetVal RaftWebConnection::canSendOnConn() ///////////////////////////////////////////////////////////////////////////////////////////////////////////////// RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs) +{ + return rawSendOnConnImpl(pBuf, bufLen, maxRetryMs, nullptr); +} + +// As rawSendOnConn, but if what can't be sent at once is to be queued and the queue is empty, the queue takes the +// buffer (leaving it empty) rather than copying it +RaftWebConnSendRetVal RaftWebConnection::rawSendOnConnTake(SpiramAwareUint8Vector& buf, uint32_t maxRetryMs) +{ + return rawSendOnConnImpl(buf.data(), buf.size(), maxRetryMs, &buf); +} + +RaftWebConnSendRetVal RaftWebConnection::rawSendOnConnImpl(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs, + SpiramAwareUint8Vector* pTakeBuf) { #if defined(ESP_PLATFORM) && defined(DEBUG_RAW_SEND_ON_CONN_TIMING) static uint64_t totalCycles = 0; @@ -1223,7 +1238,7 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint #endif // Check if data can be sent immediately - if (_socketTxQueuedBuffer.size() == 0) + if (txQueuedLen() == 0) { // Queue is currently empty so try to send #if defined(ESP_PLATFORM) && defined(DEBUG_RAW_SEND_ON_CONN_TIMING) @@ -1266,7 +1281,7 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint retFinal = WEB_CONN_SEND_FAIL; break; } - uint32_t curSize = _socketTxQueuedBuffer.size(); + uint32_t curSize = txQueuedLen(); if (curSize + bytesToAddToQueue > _maxSendBufferBytes) { #ifdef DEBUG_WEB_CONNECTION_DATA_PACKETS @@ -1281,6 +1296,23 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint #if defined(ESP_PLATFORM) && defined(DEBUG_RAW_SEND_ON_CONN_TIMING) uint32_t queueStartCycles = esp_cpu_get_cycle_count(); #endif + // If the queue is empty and the buffer may be taken, the queue takes it rather than copying what is left of it + if (pTakeBuf && (curSize == 0)) + { + _socketTxQueuedBuffer.swap(*pTakeBuf); + _socketTxQueuedPos = bytesWritten; + pTakeBuf->clear(); + retFinal = WEB_CONN_SEND_OK; + break; + } + + // Remove what has been sent from the queue before adding to it + if (_socketTxQueuedPos > 0) + { + _socketTxQueuedBuffer.erase(_socketTxQueuedBuffer.begin(), _socketTxQueuedBuffer.begin() + _socketTxQueuedPos); + _socketTxQueuedPos = 0; + } + // Check the larger buffer can be allocated - an allocation failure can't be detected afterwards (exceptions // are disabled) and would crash. This happens when several connections queue data at once while the link // drains slowly (e.g. weak WiFi) and the heap is fragmented. The buffer is grown to the exact size needed @@ -1309,7 +1341,7 @@ RaftWebConnSendRetVal RaftWebConnection::rawSendOnConn(const uint8_t* pBuf, uint #ifdef DEBUG_WEB_CONNECTION_DATA_PACKETS LOG_I(MODULE_PREFIX, "rawSendOnConn connId %d added %d bytes to send buffer newLen %d", - _pClientConn->getClientId(), bytesToAddToQueue, _socketTxQueuedBuffer.size()); + _pClientConn->getClientId(), bytesToAddToQueue, txQueuedLen()); #endif // The data has been queued for sending (WEB_CONN_SEND_OK) @@ -1544,7 +1576,7 @@ bool RaftWebConnection::handleResponseChunk() #endif // Check if data waiting to be sent - if (_socketTxQueuedBuffer.size() == 0) + if (txQueuedLen() == 0) { // Get next chunk of response uint8_t* pRespBuffer = nullptr; @@ -1622,33 +1654,35 @@ bool RaftWebConnection::handleResponseChunk() bool RaftWebConnection::handleTxQueuedData() { // Check if data in the queue to send - if (_socketTxQueuedBuffer.size() > 0) + if (txQueuedLen() > 0) { #ifdef DEBUG_WEB_CONN_OPEN_CLOSE LOG_I(MODULE_PREFIX, "handleTxQueuedData connId %d HAS DATA: %d bytes queued", - _pClientConn ? _pClientConn->getClientId() : -1, _socketTxQueuedBuffer.size()); + _pClientConn ? _pClientConn->getClientId() : -1, txQueuedLen()); #endif // Try to send uint32_t bytesWritten = 0; - uint32_t bytesToSend = _socketTxQueuedBuffer.size(); - RaftWebConnSendRetVal retVal = _pClientConn->sendDataBuffer(_socketTxQueuedBuffer.data(), bytesToSend, + uint32_t bytesToSend = txQueuedLen(); + RaftWebConnSendRetVal retVal = _pClientConn->sendDataBuffer(_socketTxQueuedBuffer.data() + _socketTxQueuedPos, bytesToSend, MAX_CONTENT_SEND_RETRY_MS, bytesWritten); if (retVal == WEB_CONN_SEND_EAGAIN) return true; #ifdef DEBUG_WEB_CONNECTION_DATA_PACKETS LOG_I(MODULE_PREFIX, "handleTxQueuedData connId %d result %s bytesWritten %d remaining %d", _pClientConn->getClientId(), RaftWebConnDefs::getSendRetValStr(retVal), bytesWritten, - _socketTxQueuedBuffer.size()-bytesWritten); + txQueuedLen()-bytesWritten); #endif if (retVal == WEB_CONN_SEND_FAIL) { // Clear the send buffer - _socketTxQueuedBuffer.clear(); + txQueueClear(); return false; } - // Sent ok so clear items from the queue that were sent - _socketTxQueuedBuffer.erase(_socketTxQueuedBuffer.begin(), _socketTxQueuedBuffer.begin() + bytesWritten); + // Sent ok so move past what was sent (emptying the queue when all is sent) + _socketTxQueuedPos += bytesWritten; + if (_socketTxQueuedPos >= _socketTxQueuedBuffer.size()) + txQueueClear(); } return true; } \ No newline at end of file diff --git a/components/RaftWebServer/RaftWebConnection.h b/components/RaftWebServer/RaftWebConnection.h index 6288573..31d0615 100644 --- a/components/RaftWebServer/RaftWebConnection.h +++ b/components/RaftWebServer/RaftWebConnection.h @@ -108,8 +108,19 @@ class RaftWebConnection // Max send buffer size uint32_t _maxSendBufferBytes; - // Queued data to send + // Queued data to send, from _socketTxQueuedPos (data before it has been sent - it is only removed when more is + // added, so sending from the queue doesn't move what is left each time) std::vector> _socketTxQueuedBuffer; + uint32_t _socketTxQueuedPos = 0; + uint32_t txQueuedLen() const + { + return _socketTxQueuedBuffer.size() - _socketTxQueuedPos; + } + void txQueueClear() + { + _socketTxQueuedBuffer.clear(); + _socketTxQueuedPos = 0; + } // Heap to leave free (for the network stack) when growing the send buffer static const uint32_t SEND_BUFFER_HEAP_MARGIN_BYTES = 8192; @@ -147,7 +158,10 @@ class RaftWebConnection void setHTTPResponseStatus(RaftHttpStatusCode reponseCode); // Raw send on connection - used by websockets, etc - RaftWebConnSendRetVal rawSendOnConn(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs); + RaftWebConnSendRetVal rawSendOnConn(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs); + RaftWebConnSendRetVal rawSendOnConnTake(SpiramAwareUint8Vector& buf, uint32_t maxRetryMs); + RaftWebConnSendRetVal rawSendOnConnImpl(const uint8_t* pBuf, uint32_t bufLen, uint32_t maxRetryMs, + SpiramAwareUint8Vector* pTakeBuf); // Header handling bool getStandardHeaders(String& headerStr); diff --git a/components/RaftWebServer/RaftWebRequestParams.h b/components/RaftWebServer/RaftWebRequestParams.h index 7c82f5d..d75b7b5 100644 --- a/components/RaftWebServer/RaftWebRequestParams.h +++ b/components/RaftWebServer/RaftWebRequestParams.h @@ -31,10 +31,21 @@ class RaftWebRequestParams return _webConnReadyToSend; } + // Send that may take the buffer (see RaftWebConnSendTakeFn) - not set for every connection + void setWebConnRawSendTake(RaftWebConnSendTakeFn webConnRawSendTake) + { + _webConnRawSendTake = webConnRawSendTake; + } + RaftWebConnSendTakeFn getWebConnRawSendTake() const + { + return _webConnRawSendTake; + } + // Connection ID (for debugging) uint32_t connId = 0; private: RaftWebConnSendFn _webConnRawSend; + RaftWebConnSendTakeFn _webConnRawSendTake = nullptr; RaftWebConnReadyToSendFn _webConnReadyToSend; }; diff --git a/components/RaftWebServer/RaftWebResponder.h b/components/RaftWebServer/RaftWebResponder.h index 6c4e010..6abcbd3 100644 --- a/components/RaftWebServer/RaftWebResponder.h +++ b/components/RaftWebServer/RaftWebResponder.h @@ -114,6 +114,13 @@ class RaftWebResponder return false; } + // Encode and send data whose buffer may be taken (leaving it empty) rather than copied - by default it is sent + // as above + virtual bool encodeAndSendData(SpiramAwareUint8Vector& buf) + { + return encodeAndSendData(buf.data(), buf.size()); + } + // Send event content and group virtual void sendEvent(const char* eventContent, const char* eventGroup) { diff --git a/components/RaftWebServer/RaftWebResponderWS.cpp b/components/RaftWebServer/RaftWebResponderWS.cpp index f490e5b..cca3b41 100644 --- a/components/RaftWebServer/RaftWebResponderWS.cpp +++ b/components/RaftWebServer/RaftWebResponderWS.cpp @@ -67,6 +67,7 @@ RaftWebResponderWS::RaftWebResponderWS(RaftWebHandlerWS* pWebHandler, const Raft _webSocketLink.setup(std::bind(&RaftWebResponderWS::onWebSocketEvent, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), params.getWebConnRawSend(), pingIntervalMs, true, disconnIfNoPongMs, isBinary); + _webSocketLink.setRawConnSendTakeFn(params.getWebConnRawSendTake()); _webSocketLink.setMaxRxMsgLen(_packetMaxBytes); } @@ -410,6 +411,16 @@ bool RaftWebResponderWS::isReadyToSend() ///////////////////////////////////////////////////////////////////////////////////////////////////////////////// bool RaftWebResponderWS::encodeAndSendData(const uint8_t* pBuf, uint32_t bufLen) +{ + return encodeAndSendDataImpl(pBuf, bufLen, nullptr); +} + +bool RaftWebResponderWS::encodeAndSendData(SpiramAwareUint8Vector& buf) +{ + return encodeAndSendDataImpl(buf.data(), buf.size(), &buf); +} + +bool RaftWebResponderWS::encodeAndSendDataImpl(const uint8_t* pBuf, uint32_t bufLen, SpiramAwareUint8Vector* pTakeBuf) { #ifdef DEBUG_WS_SEND_APP_DATA LOG_I(MODULE_PREFIX, "encodeAndSendData connId %d len=%d status=%d", @@ -467,7 +478,9 @@ bool RaftWebResponderWS::encodeAndSendData(const uint8_t* pBuf, uint32_t bufLen) return putRslt; #else // Send - RaftWebConnSendRetVal retVal = _webSocketLink.sendMsg(_webSocketLink.msgOpCodeDefault(), pBuf, bufLen); + RaftWebConnSendRetVal retVal = pTakeBuf ? + _webSocketLink.sendMsg(_webSocketLink.msgOpCodeDefault(), *pTakeBuf) : + _webSocketLink.sendMsg(_webSocketLink.msgOpCodeDefault(), pBuf, bufLen); #ifdef DEBUG_WS_SEND_APP_DATA LOG_W(MODULE_PREFIX, "encodeAndSendData connId %d sent len %d retc %d", diff --git a/components/RaftWebServer/RaftWebResponderWS.h b/components/RaftWebServer/RaftWebResponderWS.h index cd010b8..e1fe1eb 100644 --- a/components/RaftWebServer/RaftWebResponderWS.h +++ b/components/RaftWebServer/RaftWebResponderWS.h @@ -70,6 +70,7 @@ class RaftWebResponderWS : public RaftWebResponder // Send a frame of data virtual bool encodeAndSendData(const uint8_t* pBuf, uint32_t bufLen) override final; + virtual bool encodeAndSendData(SpiramAwareUint8Vector& buf) override final; // Get responder type virtual const char* getResponderType() override final @@ -91,6 +92,9 @@ class RaftWebResponderWS : public RaftWebResponder } private: + // Encode and send data (pTakeBuf, if not null, is the buffer holding it, which may be taken) + bool encodeAndSendDataImpl(const uint8_t* pBuf, uint32_t bufLen, SpiramAwareUint8Vector* pTakeBuf); + // Handler RaftWebHandlerWS* _pWebHandler; diff --git a/components/RaftWebServer/RaftWebServer.h b/components/RaftWebServer/RaftWebServer.h index fee50aa..cf23034 100644 --- a/components/RaftWebServer/RaftWebServer.h +++ b/components/RaftWebServer/RaftWebServer.h @@ -50,6 +50,12 @@ class RaftWebServer return _connManager.sendBufOnChannel(pBuf, bufLen, channelID); } + // Send a buffer on a channel, which may be taken (leaving it empty) rather than copied (main task only) + bool sendBufferOnChannel(SpiramAwareUint8Vector& buf, uint32_t channelID) + { + return _connManager.sendBufOnChannel(buf, channelID); + } + // Send to all server-side events (main task only) void serverSideEventsSendMsg(const char* eventContent, const char* eventGroup); diff --git a/components/RaftWebServer/RaftWebSocketLink.cpp b/components/RaftWebServer/RaftWebSocketLink.cpp index edccab0..03a586a 100644 --- a/components/RaftWebServer/RaftWebSocketLink.cpp +++ b/components/RaftWebServer/RaftWebSocketLink.cpp @@ -367,6 +367,17 @@ uint32_t RaftWebSocketLink::getTxData(uint8_t*& pBuf, uint32_t bufMaxLen) ///////////////////////////////////////////////////////////////////////////////////////////////////////////////// RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, const uint8_t *pBuf, uint32_t bufLen) +{ + return sendMsgImpl(opCode, pBuf, bufLen, nullptr); +} + +RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, SpiramAwareUint8Vector& buf) +{ + return sendMsgImpl(opCode, buf.data(), buf.size(), &buf); +} + +RaftWebConnSendRetVal RaftWebSocketLink::sendMsgImpl(RaftWebSocketOpCodes opCode, const uint8_t *pBuf, uint32_t bufLen, + SpiramAwareUint8Vector* pTakeBuf) { #ifdef DEBUG_WEBSOCKET_SEND // Log all sends to debug reconnection issue @@ -449,32 +460,55 @@ RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, co return WEB_CONN_SEND_FRAME_ERROR; } - // Frame: header then data, appended to reserved space (it isn't zero-filled first, so large messages are - // written once rather than twice) - std::vector> frameBuffer; - frameBuffer.reserve(frameLen); - frameBuffer.insert(frameBuffer.end(), header, header + pos); - frameBuffer.insert(frameBuffer.end(), pBuf, pBuf + bufLen); + // Unmasked data (sent by a server) is sent as it is, after the header - without copying it into a frame first. + // If the connection can't take all of it, what is left is queued in order by the connection, and if it can't be + // queued the send fails and the connection is closed, so a frame is never sent in part +#ifdef DEBUG_WEBSOCKET_LINK_SEND + uint64_t timeNowUs = micros(); +#endif + RaftWebConnSendRetVal sendRetc = RaftWebConnSendRetVal::WEB_CONN_SEND_FAIL; + if (!_maskSentData) + { +#ifdef DEBUG_WEBSOCKET_TIME_SEND_MSG + framePrepareUs += (micros() - startUs); + startUs = micros(); +#endif + if (_rawConnSendFn) + { + sendRetc = _rawConnSendFn(header, pos, MAX_WS_SEND_RETRY_MS); + if ((sendRetc == RaftWebConnSendRetVal::WEB_CONN_SEND_OK) && (bufLen > 0)) + { + // The data's buffer is taken by the connection (rather than what it can't send at once copied) if + // it may be + if (pTakeBuf && _rawConnSendTakeFn) + sendRetc = _rawConnSendTakeFn(*pTakeBuf, MAX_WS_SEND_RETRY_MS); + else + sendRetc = _rawConnSendFn(pBuf, bufLen, MAX_WS_SEND_RETRY_MS); - // Mask the data - if (_maskSentData) + // The header has gone, so the frame must be completed or the connection closed + if (sendRetc != RaftWebConnSendRetVal::WEB_CONN_SEND_OK) + sendRetc = RaftWebConnSendRetVal::WEB_CONN_SEND_FAIL; + } + } + } + else { + // Masked data (sent by a client): header then data, appended to reserved space (it isn't zero-filled first, + // so large messages are written once rather than twice) + std::vector> frameBuffer; + frameBuffer.reserve(frameLen); + frameBuffer.insert(frameBuffer.end(), header, header + pos); + frameBuffer.insert(frameBuffer.end(), pBuf, pBuf + bufLen); for (uint32_t i = 0; i < bufLen; i++) frameBuffer[pos + i] ^= maskBytes[i % WSHeaderInfo::WEB_SOCKET_MASK_KEY_BYTES]; - } #ifdef DEBUG_WEBSOCKET_TIME_SEND_MSG - framePrepareUs += (micros() - startUs); - startUs = micros(); + framePrepareUs += (micros() - startUs); + startUs = micros(); #endif - - // Send -#ifdef DEBUG_WEBSOCKET_LINK_SEND - uint64_t timeNowUs = micros(); -#endif - RaftWebConnSendRetVal sendRetc = RaftWebConnSendRetVal::WEB_CONN_SEND_FAIL; - if (_rawConnSendFn) - sendRetc = _rawConnSendFn(frameBuffer.data(), frameBuffer.size(), MAX_WS_SEND_RETRY_MS); + if (_rawConnSendFn) + sendRetc = _rawConnSendFn(frameBuffer.data(), frameBuffer.size(), MAX_WS_SEND_RETRY_MS); + } #ifdef DEBUG_WEBSOCKET_TIME_SEND_MSG rawConnSendUs += (micros() - startUs); @@ -498,7 +532,7 @@ RaftWebConnSendRetVal RaftWebSocketLink::sendMsg(RaftWebSocketOpCodes opCode, co uint64_t timeTakenUs = micros() - timeNowUs; LOG_I(MODULE_PREFIX, "sendMsg result %s send %d bytes took %dus", RaftWebConnDefs::getSendRetValStr(sendRetc), - frameBuffer.size(), (int)timeTakenUs); + frameLen, (int)timeTakenUs); #endif return sendRetc; diff --git a/components/RaftWebServer/RaftWebSocketLink.h b/components/RaftWebServer/RaftWebSocketLink.h index 158cf49..47cd757 100644 --- a/components/RaftWebServer/RaftWebSocketLink.h +++ b/components/RaftWebServer/RaftWebSocketLink.h @@ -51,6 +51,16 @@ class RaftWebSocketLink // Send message RaftWebConnSendRetVal sendMsg(RaftWebSocketOpCodes opCode, const uint8_t* pBuf, uint32_t bufLen); + // Send a message whose buffer may be taken (leaving it empty) rather than copied, if the connection can take it + // (see setRawConnSendTakeFn) + RaftWebConnSendRetVal sendMsg(RaftWebSocketOpCodes opCode, SpiramAwareUint8Vector& buf); + + // Raw send on the connection which may take the buffer (optional) + void setRawConnSendTakeFn(RaftWebConnSendTakeFn rawConnSendTakeFn) + { + _rawConnSendTakeFn = rawConnSendTakeFn; + } + // Check active bool isActive() { @@ -100,8 +110,13 @@ class RaftWebSocketLink // Received data not yet processed SpiramAwareUint8Vector _rxDataToProcess; - // Raw send on the connection + // Raw send on the connection, and one which may take the buffer RaftWebConnSendFn _rawConnSendFn = nullptr; + RaftWebConnSendTakeFn _rawConnSendTakeFn = nullptr; + + // Send a message (pTakeBuf, if not null, is the buffer holding it, which may be taken) + RaftWebConnSendRetVal sendMsgImpl(RaftWebSocketOpCodes opCode, const uint8_t* pBuf, uint32_t bufLen, + SpiramAwareUint8Vector* pTakeBuf); // Data to be sent String _wsUpgradeResponse; diff --git a/components/WebServer/WebServer.cpp b/components/WebServer/WebServer.cpp index 893c5f5..76b9c27 100644 --- a/components/WebServer/WebServer.cpp +++ b/components/WebServer/WebServer.cpp @@ -381,6 +381,10 @@ void WebServer::webSocketSetup() interfaceName.c_str(), wsName.c_str(), [this](CommsChannelMsg& msg) { + // A message just encoded for sending (e.g. a publish) can have its buffer taken by the + // connection, rather than copied, for what can't be sent at once + if (msg.isBufTakeable()) + return _raftWebServer.sendBufferOnChannel(msg.getCmdVector(), msg.getChannelID()); return _raftWebServer.sendBufferOnChannel(msg.getBuf(), msg.getBufLen(), msg.getChannelID()); },