From 9dcb98bb519d6ce5a9032f0735e93d3d2a016bb2 Mon Sep 17 00:00:00 2001 From: wellenvogel Date: Wed, 1 Dec 2021 22:55:37 +0100 Subject: [PATCH] intermediate: restructure buffer handling --- lib/queue/GwBuffer.cpp | 72 ++++++++++++++++++++++++++++- lib/queue/GwBuffer.h | 22 ++++++--- lib/serial/GwSerial.cpp | 47 ++++++++----------- lib/serial/GwSerial.h | 4 +- lib/socketserver/GwSocketServer.cpp | 6 +-- lib/socketserver/GwSocketServer.h | 2 +- src/main.cpp | 20 ++++---- 7 files changed, 119 insertions(+), 54 deletions(-) diff --git a/lib/queue/GwBuffer.cpp b/lib/queue/GwBuffer.cpp index 0f9d676..bfbc91b 100644 --- a/lib/queue/GwBuffer.cpp +++ b/lib/queue/GwBuffer.cpp @@ -18,8 +18,9 @@ GwBuffer::GwBuffer(GwLog *logger,size_t bufferSize) GwBuffer::~GwBuffer(){ delete buffer; } -void GwBuffer::reset() +void GwBuffer::reset(String reason) { + LOG_DEBUG(GwLog::LOG,"reseting buffer %p, reason %s",this,reason.c_str()); writePointer = buffer; readPointer = buffer; lp("reset"); @@ -131,6 +132,45 @@ GwBuffer::WriteStatus GwBuffer::fetchData(GwBufferWriter *writer, int maxLen,boo } return (written == len)?OK:AGAIN; } +size_t GwBuffer::fetchData(int maxLen, GwBufferHandleFunction handler, void *param){ + if (usedSpace() < 1) return 0; + size_t len=0; + if (writePointer > readPointer){ + len=writePointer-readPointer; + } + else{ + len=bufferSize-offset(readPointer)-1; + } + if (maxLen >= 0 && maxLen < len) len=maxLen; + size_t handled=handler(readPointer,len,param); + if (handled > len) handled=len; + readPointer+=handled; + if (offset(readPointer) >= bufferSize ) readPointer-=bufferSize; + return handled; +} +size_t GwBuffer::fillData(int maxLen, GwBufferHandleFunction handler, void *param) +{ + if (freeSpace() < 1) + return 0; + size_t len = 0; + if (writePointer > readPointer) + { + len = bufferSize - offset(writePointer) - 1; + } + else + { + len = readPointer - writePointer - 1; + } + if (maxLen >= 0 && maxLen < len) + len = maxLen; + size_t handled = handler(writePointer, len,param); + if (handled > len) + handled = len; + writePointer += handled; + if (offset(writePointer) >= bufferSize) + writePointer -= bufferSize; + return handled; +} int GwBuffer::findChar(char x){ lp("findChar",x); @@ -159,4 +199,34 @@ GwBuffer::WriteStatus GwBuffer::fetchMessage(GwBufferWriter *writer,char delimit return AGAIN; } return fetchData(writer,pos+1,true); +} + +size_t GwMessageFetcher::fetchMessageToBuffer(GwBuffer *gwbuffer,uint8_t *buffer, size_t bufferLen,char delimiter){ + int offset=gwbuffer->findChar(delimiter); + if (offset <0) { + if (! gwbuffer->freeSpace()){ + gwbuffer->reset(F("Message to long for input buffer")); + } + return 0; + } + offset+=1; //we include the delimiter + if (offset >= bufferLen){ + gwbuffer->reset(F("Message to long for message buffer")); + return 0; + } + size_t fetched=0; + while (fetched < offset){ + size_t max=offset-fetched; + size_t rd=gwbuffer->fetchData(max,[](uint8_t *buffer, size_t len, void *p)->size_t{ + memcpy(p,buffer,len); + return len; + },buffer+fetched); + if (rd == 0){ + //some internal error + gwbuffer->reset(F("Internal fetch error")); + return 0; + } + fetched+=rd; + } + return fetched; } \ No newline at end of file diff --git a/lib/queue/GwBuffer.h b/lib/queue/GwBuffer.h index 9a133d2..d8d51f8 100644 --- a/lib/queue/GwBuffer.h +++ b/lib/queue/GwBuffer.h @@ -4,6 +4,7 @@ #include #include "GwLog.h" +class GwBuffer; class GwBufferWriter{ public: int id=0; //can be set be users @@ -12,11 +13,19 @@ class GwBufferWriter{ virtual ~GwBufferWriter(){}; }; +class GwMessageFetcher{ + public: + int id=0; + virtual bool handleBuffer(GwBuffer *buffer)=0; + virtual size_t fetchMessageToBuffer(GwBuffer *gwbuffer,uint8_t *buffer, size_t bufferLen,char delimiter); +}; + /** * an implementation of a * buffer to safely inserte data if it fits * and to write out data if possible */ +typedef size_t (*GwBufferHandleFunction)(uint8_t *buffer, size_t len, void *param); class GwBuffer{ public: static const size_t TX_BUFFER_SIZE=1620; // app. 20 NMEA messages @@ -36,25 +45,26 @@ class GwBuffer{ } GwLog *logger; void lp(const char *fkt,int p=0); - /** - * find the first occurance of x in the buffer, -1 if not found - */ - int findChar(char x); public: GwBuffer(GwLog *logger,size_t bufferSize); ~GwBuffer(); - void reset(); + void reset(String reason=""); size_t freeSpace(); size_t usedSpace(); size_t addData(const uint8_t *data,size_t len,bool addPartial=false); + size_t fillData(int maxLen,GwBufferHandleFunction handler, void *param); int read(); int peek(); + size_t fetchData(int maxLen,GwBufferHandleFunction handler, void *param); /** * write some data to the buffer writer * return an error if the buffer writer returned < 0 */ WriteStatus fetchData(GwBufferWriter *writer, int maxLen=-1,bool errorIf0 = true); - + /** + * find the first occurance of x in the buffer, -1 if not found + */ + int findChar(char x); WriteStatus fetchMessage(GwBufferWriter *writer,char delimiter,bool emptyIfFull=true); }; diff --git a/lib/serial/GwSerial.cpp b/lib/serial/GwSerial.cpp index d9f1a80..474148e 100644 --- a/lib/serial/GwSerial.cpp +++ b/lib/serial/GwSerial.cpp @@ -39,23 +39,7 @@ class GwSerialStream: public Stream{ }; -class SerialWriter : public GwBufferWriter{ - private: - HardwareSerial *serial; - public: - SerialWriter(HardwareSerial *serial){ - this->serial=serial; - } - virtual ~SerialWriter(){} - virtual int write(const uint8_t *buffer,size_t len){ - size_t numWrite=serial->availableForWrite(); - if (numWrite < len) len=numWrite; - if (!len) return 0; - return serial->write(buffer,len); - } - -}; GwSerial::GwSerial(GwLog *logger, int num, int id,bool allowRead) { LOG_DEBUG(GwLog::DEBUG,"creating GwSerial %p port %d",this,(int)num); @@ -68,13 +52,11 @@ GwSerial::GwSerial(GwLog *logger, int num, int id,bool allowRead) this->readBuffer=new GwBuffer(logger, GwBuffer::RX_BUFFER_SIZE); } this->serial=new HardwareSerial(num); - this->writer = new SerialWriter(serial); } GwSerial::~GwSerial() { delete buffer; - delete writer; if (readBuffer) delete readBuffer; delete serial; } @@ -93,7 +75,12 @@ size_t GwSerial::enqueue(const uint8_t *data, size_t len, bool partial) } GwBuffer::WriteStatus GwSerial::write(){ if (! isInitialized()) return GwBuffer::ERROR; - return buffer->fetchData(writer,-1,false); + size_t numWrite=serial->availableForWrite(); + size_t rt=buffer->fetchData(numWrite,[](uint8_t *buffer,size_t len, void *p){ + return ((GwSerial *)p)->serial->write(buffer,len); + },this); + LOG_DEBUG(GwLog::DEBUG+1,"Serial %p write %d",this,rt); + return buffer->usedSpace()?GwBuffer::AGAIN:GwBuffer::OK; } size_t GwSerial::sendToClients(const char *buf,int sourceId,bool partial){ if ( sourceId == id) return 0; @@ -109,25 +96,29 @@ void GwSerial::loop(bool handleRead){ write(); if (! isInitialized()) return; if (! handleRead) return; - char buffer[10]; size_t available=serial->available(); if (! available) return; - if (available > 10) available=10; - int rt=serial->readBytes(buffer,available); - if (allowRead && rt > 0){ - size_t wr=readBuffer->addData((uint8_t *)(&buffer),rt,true); - LOG_DEBUG(GwLog::DEBUG+2,"GwSerial read %d bytes, to buffer %d bytes",rt,wr); + if (allowRead){ + size_t rd=readBuffer->fillData(available,[](uint8_t *buffer, size_t len, void *p)->size_t{ + return ((GwSerial *)p)->serial->readBytes(buffer,len); + },this); + this->logger->logDebug(GwLog::DEBUG+2,"GwSerial read %d bytes",rd); + } + else{ + uint8_t buffer[10]; + if (available > 10) available=10; + serial->readBytes(buffer,available); } } -bool GwSerial::readMessages(GwBufferWriter *writer){ +bool GwSerial::readMessages(GwMessageFetcher *writer){ if (! isInitialized()) return false; if (! allowRead) return false; - return readBuffer->fetchMessage(writer,'\n',true) == GwBuffer::OK; + return writer->handleBuffer(readBuffer); } void GwSerial::flush(){ if (! isInitialized()) return; - while (buffer->fetchData(writer,-1,false) == GwBuffer::AGAIN){ + while (write() == GwBuffer::AGAIN){ vTaskDelay(1); } } diff --git a/lib/serial/GwSerial.h b/lib/serial/GwSerial.h index fe6adfb..1fd23d6 100644 --- a/lib/serial/GwSerial.h +++ b/lib/serial/GwSerial.h @@ -3,14 +3,12 @@ #include "HardwareSerial.h" #include "GwLog.h" #include "GwBuffer.h" -class SerialWriter; class GwSerialStream; class GwSerial{ private: GwBuffer *buffer; GwBuffer *readBuffer=NULL; GwLog *logger; - SerialWriter *writer; int num; bool initialized=false; bool allowRead=true; @@ -27,7 +25,7 @@ class GwSerial{ bool isInitialized(); size_t sendToClients(const char *buf,int sourceId,bool partial=false); void loop(bool handleRead=true); - bool readMessages(GwBufferWriter *writer); + bool readMessages(GwMessageFetcher *writer); void flush(); Stream *getStream(bool partialWrites); friend GwSerialStream; diff --git a/lib/socketserver/GwSocketServer.cpp b/lib/socketserver/GwSocketServer.cpp index 1263afc..c8ca51f 100644 --- a/lib/socketserver/GwSocketServer.cpp +++ b/lib/socketserver/GwSocketServer.cpp @@ -154,9 +154,9 @@ class GwClient{ } return true; } - bool messagesFromBuffer(GwBufferWriter *writer){ + bool messagesFromBuffer(GwMessageFetcher *writer){ if (! allowRead) return false; - return readBuffer->fetchMessage(writer,'\n',true) == GwBuffer::OK; + return writer->handleBuffer(readBuffer); } }; @@ -239,7 +239,7 @@ void GwSocketServer::loop(bool handleRead) } } -bool GwSocketServer::readMessages(GwBufferWriter *writer){ +bool GwSocketServer::readMessages(GwMessageFetcher *writer){ if (! allowReceive || ! clients) return false; bool hasMessages=false; for (int i = 0; i < maxClients; i++){ diff --git a/lib/socketserver/GwSocketServer.h b/lib/socketserver/GwSocketServer.h index 61166c1..1968887 100644 --- a/lib/socketserver/GwSocketServer.h +++ b/lib/socketserver/GwSocketServer.h @@ -25,6 +25,6 @@ class GwSocketServer{ void loop(bool handleRead=true); void sendToClients(const char *buf,int sourceId); int numClients(); - bool readMessages(GwBufferWriter *writer); + bool readMessages(GwMessageFetcher *writer); }; #endif \ No newline at end of file diff --git a/src/main.cpp b/src/main.cpp index 669daf5..c46b88f 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -783,20 +783,15 @@ void handleSendAndRead(bool handleRead){ usbSerial->loop(handleRead); if (serial1) serial1->loop(handleRead); } -class NMEAMessageReceiver : public GwBufferWriter{ - uint8_t buffer[GwBuffer::RX_BUFFER_SIZE+4]; +class NMEAMessageReceiver : public GwMessageFetcher{ + static const int bufferSize=GwBuffer::RX_BUFFER_SIZE+4; + uint8_t buffer[bufferSize]; uint8_t *writePointer=buffer; public: - virtual int write(const uint8_t *buffer,size_t len){ - size_t toWrite=GwBuffer::RX_BUFFER_SIZE-(writePointer-buffer); - if (toWrite > len) toWrite=len; - memcpy(writePointer,buffer,toWrite); - writePointer+=toWrite; - *writePointer=0; - return toWrite; - } - virtual void done(){ - if (writePointer == buffer) return; + virtual bool handleBuffer(GwBuffer *gwbuffer){ + size_t len=fetchMessageToBuffer(gwbuffer,buffer,bufferSize-4,'\n'); + writePointer=buffer+len; + if (writePointer == buffer) return false; uint8_t *p; for (p=writePointer-1;p>=buffer && *p <= 0x20;p--){ *p=0; @@ -821,6 +816,7 @@ class NMEAMessageReceiver : public GwBufferWriter{ handleSendAndRead(false); } writePointer=buffer; + return true; } }; NMEAMessageReceiver receiver;