thooge
/
esp32-nmea2000-obp60
mirror of https://github.com/thooge/esp32-nmea2000-obp60.git
1
0
Fork 0
Code Releases Activity

intermediate: restructure buffer handling

This commit is contained in:
wellenvogel 2021-12-01 22:55:37 +01:00
parent d8950c4eb4
commit 9dcb98bb51
7 changed files with 119 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
lib/queue/GwBuffer.cpp
View File

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

22
lib/queue/GwBuffer.h
View File

@ -4,6 +4,7 @@
#include <stdint.h>
#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);
};

47
lib/serial/GwSerial.cpp
View File

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

4
lib/serial/GwSerial.h
View File

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

6
lib/socketserver/GwSocketServer.cpp
View File

@ -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++){

2
lib/socketserver/GwSocketServer.h
View File

@ -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

20
src/main.cpp
View File

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