handle receiving and distributing on NMEA messages

lib/nmea2kto0183/N2kDataToNMEA0183.cpp

@@ -31,10 +31,11 @@
 
 
 
-N2kDataToNMEA0183::N2kDataToNMEA0183(GwLog * logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183) : tNMEA2000::tMsgHandler(0,NMEA2000){
+N2kDataToNMEA0183::N2kDataToNMEA0183(GwLog * logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183, int id) 
+: tNMEA2000::tMsgHandler(0,NMEA2000){
     SendNMEA0183MessageCallback=0;
     pNMEA0183=NMEA0183;
-    
+    sourceId=id;   
   }
 
 
@@ -46,10 +47,11 @@ void N2kDataToNMEA0183::loop() {
 //*****************************************************************************
 void N2kDataToNMEA0183::SendMessage(const tNMEA0183Msg &NMEA0183Msg) {
   if ( pNMEA0183 != 0 ) pNMEA0183->SendMessage(NMEA0183Msg);
-  if ( SendNMEA0183MessageCallback != 0 ) SendNMEA0183MessageCallback(NMEA0183Msg);
+  if ( SendNMEA0183MessageCallback != 0 ) SendNMEA0183MessageCallback(NMEA0183Msg, sourceId);
 }
 
-N2kDataToNMEA0183* N2kDataToNMEA0183::create(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183){
+N2kDataToNMEA0183* N2kDataToNMEA0183::create(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, 
-  return new N2kToNMEA0183Functions(logger,boatData,NMEA2000,NMEA0183);
+    tNMEA0183 *NMEA0183, int sourceId){
+  return new N2kToNMEA0183Functions(logger,boatData,NMEA2000,NMEA0183, sourceId);
 }
 //*****************************************************************************

lib/nmea2kto0183/N2kDataToNMEA0183.h

@@ -33,22 +33,19 @@ OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 class N2kDataToNMEA0183 : public tNMEA2000::tMsgHandler
 {
 public:
-  using tSendNMEA0183MessageCallback = void (*)(const tNMEA0183Msg &NMEA0183Msg);
+  using tSendNMEA0183MessageCallback = void (*)(const tNMEA0183Msg &NMEA0183Msg, int id);
 
 protected:
   GwLog *logger;
   GwBoatData *boatData;
-
   tNMEA0183 *pNMEA0183;
+  int sourceId;
   tSendNMEA0183MessageCallback SendNMEA0183MessageCallback;
-
-
   void SendMessage(const tNMEA0183Msg &NMEA0183Msg);
-
+  N2kDataToNMEA0183(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183, int sourceId);
-  N2kDataToNMEA0183(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183);
 
 public:
-  static N2kDataToNMEA0183* create(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183);
+  static N2kDataToNMEA0183* create(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183, int sourceId);
   virtual void HandleMsg(const tN2kMsg &N2kMsg) = 0;
   void SetSendNMEA0183MessageCallback(tSendNMEA0183MessageCallback _SendNMEA0183MessageCallback)
   {

lib/nmea2kto0183/N2kToNMEA0183Functions.h

@@ -849,7 +849,8 @@ private:
     }
 
 public:
-    N2kToNMEA0183Functions(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183) : N2kDataToNMEA0183(logger, boatData, NMEA2000, NMEA0183)
+    N2kToNMEA0183Functions(GwLog *logger, GwBoatData *boatData, tNMEA2000 *NMEA2000, tNMEA0183 *NMEA0183, int sourceId) 
+    : N2kDataToNMEA0183(logger, boatData, NMEA2000, NMEA0183,sourceId)
     {
         LastPosSend = 0;
         lastLoopTime = 0;

lib/serial/GwSerial.cpp

@@ -13,17 +13,24 @@ class SerialWriter : public GwBufferWriter{
 
 
 };
-GwSerial::GwSerial(GwLog *logger, uart_port_t num)
+GwSerial::GwSerial(GwLog *logger, uart_port_t num, int id,bool allowRead)
 {
+    this->id=id;
     this->logger = logger;
     this->num = num;
-    this->buffer = new GwBuffer(logger,1600);
+    this->buffer = new GwBuffer(logger,GwBuffer::TX_BUFFER_SIZE);
     this->writer = new SerialWriter(num);
+    this->allowRead=allowRead;
+    if (allowRead){
+        this->readBuffer=new GwBuffer(logger, GwBuffer::RX_BUFFER_SIZE);
+    }
+
 }
 GwSerial::~GwSerial()
 {
     delete buffer;
     delete writer;
+    if (readBuffer) delete readBuffer;
 }
 int GwSerial::setup(int baud, int rxpin, int txpin)
 {
@@ -54,13 +61,32 @@ int GwSerial::setup(int baud, int rxpin, int txpin)
 bool GwSerial::isInitialized() { return initialized; }
 size_t GwSerial::enqueue(const uint8_t *data, size_t len)
 {
+    if (! isInitialized()) return 0;
     return buffer->addData(data, len);
 }
 GwBuffer::WriteStatus GwSerial::write(){
+    if (! isInitialized()) return GwBuffer::ERROR;
     return buffer->fetchData(writer,false);
 }
-const char *GwSerial::read(){
+void GwSerial::sendToClients(const char *buf,int sourceId){
+    if ( sourceId == id) return;
+    size_t len=strlen(buf);
+    size_t enqueued=enqueue((const uint8_t*)buf,len);
+    if (enqueued != len){
+        LOG_DEBUG(GwLog::DEBUG,"GwSerial overflow on channel %d",id);
+        overflows++;
+    }
+}
+void GwSerial::loop(bool handleRead){
+    write();
+    if (! handleRead) return;
     char buffer[10];
-    uart_read_bytes(num,(uint8_t *)(&buffer),10,0);
+    int rt=uart_read_bytes(num,(uint8_t *)(&buffer),10,0);
-    return NULL;   
+    if (allowRead & rt > 0){
+        readBuffer->addData((uint8_t *)(&buffer),rt,true);
+    }
+}
+bool GwSerial::readMessages(GwBufferWriter *writer){
+    if (! allowRead) return false;
+    return readBuffer->fetchMessage(writer,'\n',true) == GwBuffer::OK;
 }

lib/serial/GwSerial.h

@@ -7,19 +7,24 @@ class SerialWriter;
 class GwSerial{
     private:
         GwBuffer *buffer;
-        GwBuffer *readBuffer;
+        GwBuffer *readBuffer=NULL;
         GwLog *logger; 
         SerialWriter *writer;
         uart_port_t num;
         bool initialized=false;
+        bool allowRead=true;
+        GwBuffer::WriteStatus write();
+        int id=-1;
+        int overflows=0;
+        size_t enqueue(const uint8_t *data, size_t len);
     public:
         static const int bufferSize=200;
-        GwSerial(GwLog *logger,uart_port_t num);
+        GwSerial(GwLog *logger,uart_port_t num,int id,bool allowRead=true);
         ~GwSerial();
         int setup(int baud,int rxpin,int txpin);
         bool isInitialized();
-        size_t enqueue(const uint8_t *data, size_t len);
+        void sendToClients(const char *buf,int sourceId);
-        GwBuffer::WriteStatus write();
+        void loop(bool handleRead=true);
-        const char *read();
+        bool readMessages(GwBufferWriter *writer);
 };
 #endif

lib/socketserver/GwSocketServer.cpp

@@ -175,7 +175,7 @@ void GwSocketServer::begin(){
     MDNS.addService("_nmea-0183","_tcp",config->getInt(config->serverPort));    
 
 }
-void GwSocketServer::loop()
+void GwSocketServer::loop(bool handleRead)
 {
     WiFiClient client = server->available(); // listen for incoming clients
 
@@ -228,7 +228,7 @@ void GwSocketServer::loop()
         }
         else
         {
-            client->read();
+            if (handleRead) client->read();
         }
     }
 }
@@ -245,12 +245,6 @@ bool GwSocketServer::readMessages(GwBufferWriter *writer){
 }
 void GwSocketServer::sendToClients(const char *buf,int source){
     int len=strlen(buf);
-    char buffer[len+2];
-    memcpy(buffer,buf,len);
-    buffer[len]=0x0d;
-    len++;
-    buffer[len]=0x0a;
-    len++;
     int sourceIndex=source-minId;
     for (int i = 0; i < maxClients; i++)
     {
@@ -258,7 +252,7 @@ void GwSocketServer::sendToClients(const char *buf,int source){
         gwClientPtr client = clients[i];
         if (! client->hasClient()) continue;
         if ( client->client->connected() ) {
-        bool rt=client->enqueue((uint8_t*)buffer,len);
+        bool rt=client->enqueue((uint8_t*)buf,len);
         if (!rt){
             LOG_DEBUG(GwLog::DEBUG,"overflow in send to %s",client->remoteIp.c_str());    
         }

lib/socketserver/GwSocketServer.h

@@ -22,7 +22,7 @@ class GwSocketServer{
         GwSocketServer(const GwConfigHandler *config,GwLog *logger,int minId);
         ~GwSocketServer();
         void begin();
-        void loop();
+        void loop(bool handleRead=true);
         void sendToClients(const char *buf,int sourceId);
         int numClients();
         bool readMessages(GwBufferWriter *writer);

src/main.cpp

@@ -65,7 +65,7 @@ Preferences preferences;             // Nonvolatile storage on ESP32 - To store
 bool SendNMEA0183Conversion = true; // Do we send NMEA2000 -> NMEA0183 conversion
 bool SendSeaSmart = false; // Do we send NMEA2000 messages in SeaSmart format
 
-N2kDataToNMEA0183 *nmea0183Converter=N2kDataToNMEA0183::create(&logger, &boatData,&NMEA2000, 0);
+N2kDataToNMEA0183 *nmea0183Converter=N2kDataToNMEA0183::create(&logger, &boatData,&NMEA2000, 0, N2K_CHANNEL_ID);
 
 // Set the information for other bus devices, which messages we support
 const unsigned long TransmitMessages[] PROGMEM = {127489L, // Engine dynamic
@@ -73,7 +73,7 @@ const unsigned long TransmitMessages[] PROGMEM = {127489L, // Engine dynamic
 };
 // Forward declarations
 void HandleNMEA2000Msg(const tN2kMsg &N2kMsg);
-void SendNMEA0183Message(const tNMEA0183Msg &NMEA0183Msg);
+void SendNMEA0183Message(const tNMEA0183Msg &NMEA0183Msg,int id);
 
 AsyncWebServer webserver(80);
 
@@ -165,12 +165,12 @@ GwConfigInterface *sendTCP=NULL;
 GwConfigInterface *sendSeasmart=NULL;
 GwConfigInterface *systemName=NULL;
 
-GwSerial usbSerial(&logger, UART_NUM_0);
+GwSerial usbSerial(&logger, UART_NUM_0, USB_CHANNEL_ID);
 class GwSerialLog : public GwLogWriter{
   public:
     virtual ~GwSerialLog(){}
     virtual void write(const char *data){
-      usbSerial.enqueue((const uint8_t*)data,strlen(data)); //ignore any errors
+      usbSerial.sendToClients(data,-1); //ignore any errors
     }
 
 };
@@ -385,30 +385,40 @@ void HandleNMEA2000Msg(const tN2kMsg &N2kMsg) {
   socketServer.sendToClients(buf,N2K_CHANNEL_ID);
 }
 
-
+void sendBufferToChannels(const char * buffer, int sourceId){
-//*****************************************************************************
-void SendNMEA0183Message(const tNMEA0183Msg &NMEA0183Msg) {
-  if ( ! sendTCP->asBoolean() && ! sendUsb->asBoolean() ) return;
-
-  char buf[MAX_NMEA0183_MESSAGE_SIZE];
-  if ( !NMEA0183Msg.GetMessage(buf, MAX_NMEA0183_MESSAGE_SIZE) ) return;
   if (sendTCP->asBoolean()){
-    socketServer.sendToClients(buf,N2K_CHANNEL_ID);
+    socketServer.sendToClients(buffer,sourceId);
   }
   if (sendUsb->asBoolean()){
-    int len=strlen(buf);
+    usbSerial.sendToClients(buffer,sourceId);
-    if (len >= (MAX_NMEA0183_MESSAGE_SIZE -2)) return;
-    buf[len]=0x0d;
-    len++;
-    buf[len]=0x0a;
-    len++;
-    buf[len]=0;
-    usbSerial.enqueue((const uint8_t*)buf,len);
   }
 }
 
+//*****************************************************************************
+void SendNMEA0183Message(const tNMEA0183Msg &NMEA0183Msg, int sourceId) {
+  if ( ! sendTCP->asBoolean() && ! sendUsb->asBoolean() ) return;
+
+  char buf[MAX_NMEA0183_MESSAGE_SIZE+3];
+  if ( !NMEA0183Msg.GetMessage(buf, MAX_NMEA0183_MESSAGE_SIZE) ) return;
+  size_t len=strlen(buf);
+  buf[len]=0x0d;
+  buf[len+1]=0x0a;
+  buf[len+2]=0;
+  sendBufferToChannels(buf,sourceId);
+}
+
+void handleReceivedNmeaMessage(const char *buf, int sourceId){
+  //TODO - for now only send out again
+  //add the conversion to N2K here
+  sendBufferToChannels(buf,sourceId);
+}
+
+void handleSendAndRead(bool handleRead){
+  socketServer.loop(handleRead);  
+  usbSerial.loop(handleRead);
+}
 class NMEAMessageReceiver : public GwBufferWriter{
-  uint8_t buffer[GwBuffer::RX_BUFFER_SIZE+1];
+  uint8_t buffer[GwBuffer::RX_BUFFER_SIZE+4];
   uint8_t *writePointer=buffer;
   public:
     virtual int write(const uint8_t *buffer,size_t len){
@@ -421,23 +431,36 @@ class NMEAMessageReceiver : public GwBufferWriter{
     }
     virtual void done(){
       if (writePointer == buffer) return;
-      logger.logDebug(GwLog::DEBUG,"NMEA-IN[%d]: %s",id,(const char *)buffer);
       uint8_t *p;
-      for (p=writePointer-1;p>=buffer;p--){
+      for (p=writePointer-1;p>=buffer && *p <= 0x20;p--){
-        if (*p <= 0x20) *p=0;
+        *p=0;
+      }
+      if (p > buffer){
+        p++;
+        *p=0x0d;
+        p++;
+        *p=0x0a;
+        p++;
+        *p=0;
       }
       for (p=buffer; *p != 0 && p < writePointer && *p <= 0x20;p++){}
-      logger.logDebug(GwLog::DEBUG,"NMEA[%d]: %s",id,(const char *)p);
+      //very simple NMEA check
+      if (*p != '!' && *p != '$'){
+        logger.logDebug(GwLog::DEBUG,"unknown line [%d] - ignore: %s",id,(const char *)p);  
+      }
+      else{
+        logger.logDebug(GwLog::DEBUG,"NMEA[%d]: %s",id,(const char *)p);
+        handleReceivedNmeaMessage((const char *)p,id);
+        //trigger sending to empty buffers
+        handleSendAndRead(false);
+      }
       writePointer=buffer;
     }
 };
 void loop() {
   gwWifi.loop();
 
-  socketServer.loop();  
+  handleSendAndRead(true);
-  if (usbSerial.write() == GwBuffer::ERROR){
-    //logger.logDebug(GwLog::DEBUG,"overflow in USB serial");
-  }
   NMEA2000.ParseMessages();
 
   int SourceAddress = NMEA2000.GetN2kSource();
@@ -461,6 +484,6 @@ void loop() {
   socketServer.readMessages(&receiver);
   //read channels
 
-  usbSerial.read();
+  usbSerial.readMessages(&receiver);
 
 }