dani/CommandStation-EX
1
0
Fork 0
mirror of https://github.com/DCC-EX/CommandStation-EX.git synced 2025-02-19 23:46:02 +01:00
Code Releases Activity

Updated code for inbound wifi data buffering

Browse Source
This commit is contained in:
Mark Muzzin 2022-05-02 23:18:05 -04:00
parent 0ab3fe07c5
commit 4be8d58f76
2 changed files with 374 additions and 270 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

505
WifiInboundHandler.cpp
View File

@ -1,25 +1,27 @@
/*
* © 2021 Fred Decker
* © 2021 Fred Decker
* © 2020-2021 Chris Harlow
* © 2020, Chris Harlow. All rights reserved.
* © 2020, Harald Barth.
*
* This file is part of Asbelos DCC API
*
* This is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* It is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with CommandStation. If not, see <https://www.gnu.org/licenses/>.
*/
© 2022 Mark Muzzin
© 2021 Fred Decker
© 2021 Fred Decker
© 2020-2021 Chris Harlow
© 2020, Chris Harlow. All rights reserved.
© 2020, Harald Barth.
This file is part of Asbelos DCC API
This is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
It is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with CommandStation. If not, see <https://www.gnu.org/licenses/>.
*/
#ifndef ARDUINO_AVR_UNO_WIFI_REV2
#include <Arduino.h>
#include "WifiInboundHandler.h"
@ -30,7 +32,7 @@
WifiInboundHandler * WifiInboundHandler::singleton;
void WifiInboundHandler::setup(Stream * ESStream) {
singleton=new WifiInboundHandler(ESStream);
singleton = new WifiInboundHandler(ESStream);
}
void WifiInboundHandler::loop() {
@ -39,221 +41,320 @@ void WifiInboundHandler::loop() {
WifiInboundHandler::WifiInboundHandler(Stream * ESStream) {
wifiStream=ESStream;
clientPendingCIPSEND=-1;
inboundRing=new RingStream(INBOUND_RING);
outboundRing=new RingStream(OUTBOUND_RING);
pendingCipsend=false;
}
wifiStream = ESStream;
clientPendingCIPSEND = -1;
inboundRing = new RingStream(INBOUND_RING);
outboundRing = new RingStream(OUTBOUND_RING);
cipSendStatus = CIP_SEND_NONE;
}
// Handle any inbound transmission
// +IPD,x,lll:data is stored in streamer[x]
// Other input returns
void WifiInboundHandler::loop1() {
// First handle all inbound traffic events because they will block the sending
if (loop2()!=INBOUND_IDLE) return;
// Other input returns
void WifiInboundHandler::loop1()
{
// First handle all inbound traffic events because they will block the sending
if (loop2() != INBOUND_IDLE) return;
WiThrottle::loop(outboundRing);
// if nothing is already CIPSEND pending, we can CIPSEND one reply
if (clientPendingCIPSEND<0) {
clientPendingCIPSEND=outboundRing->read();
if (clientPendingCIPSEND>=0) {
currentReplySize=outboundRing->count();
pendingCipsend=true;
}
}
WiThrottle::loop(outboundRing);
if (pendingCipsend) {
if (Diag::WIFI) DIAG( F("WiFi: [[CIPSEND=%d,%d]]"), clientPendingCIPSEND, currentReplySize);
StringFormatter::send(wifiStream, F("AT+CIPSEND=%d,%d\r\n"), clientPendingCIPSEND, currentReplySize);
pendingCipsend=false;
return;
// If no sends are pending we can check if there is any more data to send
if (cipSendStatus == CIP_SEND_NONE) {
// If nothing is already CIPSEND pending, we can CIPSEND one reply
if (clientPendingCIPSEND < 0) {
//Read the next character in the outbound ring. If its -1, no data in the ring
clientPendingCIPSEND = outboundRing->read();
//If there is data in the ring
if (clientPendingCIPSEND >= 0 ) {
if (Diag::WIFI) DIAG( F("cipSendStatus = CIP_SEND_PENDING"));
//Get the length of the reply to send
currentReplySize = outboundRing->count();
//Set the pending CIPSEND flag
cipSendStatus = CIP_SEND_PENDING;
}
// if something waiting to execute, we can call it
int clientId=inboundRing->read();
if (clientId>=0) {
int count=inboundRing->count();
if (Diag::WIFI) DIAG(F("Wifi EXEC: %d %d:"),clientId,count);
byte cmd[count+1];
for (int i=0;i<count;i++) cmd[i]=inboundRing->read();
cmd[count]=0;
if (Diag::WIFI) DIAG(F("%e"),cmd);
outboundRing->mark(clientId); // remember start of outbound data
CommandDistributor::parse(clientId,cmd,outboundRing);
// The commit call will either write the lenbgth bytes
// OR rollback to the mark because the reply is empty or commend generated more than fits the buffer
if (!outboundRing->commit()) {
DIAG(F("OUTBOUND FULL processing cmd:%s"),cmd);
}
return;
}
}
}
}
if (cipSendStatus == CIP_SEND_PENDING) {
if (Diag::WIFI) DIAG( F("WiFi: [[CIPSEND=%d,%d]]"), clientPendingCIPSEND, currentReplySize);
StringFormatter::send(wifiStream, F("AT+CIPSEND=%d,%d\r\n"), clientPendingCIPSEND, currentReplySize);
if (Diag::WIFI) DIAG( F("pendingCipsend = CIP_SEND_SENT"));
cipSendStatus = CIP_SEND_SENT;
return;
}
// if something waiting to execute, we can call it
int clientId = inboundRing->read();
if (clientId >= 0) {
int count = inboundRing->count();
if (Diag::WIFI) DIAG(F("Wifi EXEC: %d %d:"), clientId, count);
byte cmd[count + 1];
for (int i = 0; i < count; i++) cmd[i] = inboundRing->read();
cmd[count] = 0;
if (Diag::WIFI) DIAG(F("%e"), cmd);
// This is a Finite State Automation (FSA) handling the inbound bytes from an ES AT command processor
outboundRing->mark(clientId); // remember start of outbound data
CommandDistributor::parse(clientId, cmd, outboundRing);
// The commit call will either write the length bytes
// OR rollback to the mark because the reply is empty or commend generated more than fits the buffer
if (outboundRing->commit() == false) {
DIAG(F("OUTBOUND FULL processing cmd:%s"), cmd);
outboundRing->flush();
inboundRing->flush();
}
return;
}
}
// Fixed length atoi
int WifiInboundHandler::antoi(char* str, int len) {
int res = 0;
for (int i = 0; i != len; ++i)
res = res * 10 + str[i] - '0';
return res;
}
// This loop processes the inbound bytes from an ES AT command processor
WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() {
while (wifiStream->available()) {
int ch = wifiStream->read();
// echo the char to the diagnostic stream in escaped format
if (Diag::WIFI) {
// DIAG(F(" %d/"), loopState);
StringFormatter::printEscape(ch); // DIAG in disguise
INBOUND_STATE retVal = INBOUND_IDLE;
char *stringStartPos = NULL;
char *stringEndPos = NULL;
char ch = 0;
int i;
bool exitLoop = false;
//Main message input loop
//This loop fills the small receive buffer one message at a time
//The execption is the +IPD message. Only the data up to the ':' character (end of length)
//is put in this buffer. The actual +IPD data is written directly into the inbound buffer
while (!exitLoop)
{
//We exit this loop either if there is no data available, or we have a complete message to process
while (wifiStream->available() && messageToProcess == RECV_MSG_NONE)
{
//Read next character
ch = wifiStream->read();
//Add to receive buffer if there is space
if (recBufferPos < INBOUND_CMD_BUFFER) {
recBuffer[recBufferPos++] = ch;
}
else {
if (Diag::WIFI) DIAG(F("WARN: Message too long for INBOUND BUFFER, purging"));
memset(recBuffer, 0, INBOUND_CMD_BUFFER);
recBufferPos = 0;
}
if (Diag::WIFI) StringFormatter::printEscape(ch);
//Process data ending in \r\n
if ((recBuffer[recBufferPos - 2] == '\r') && (recBuffer[recBufferPos - 1] == '\n')) {
messageToProcess = RECV_MSG_CRLF;
}
//Ready to send data, will always be the first char in the message
else if (recBuffer[0] == '>') {
messageToProcess = RECV_MSG_SND_DATA;
}
//IPD message
else if ((stringStartPos = strstr(recBuffer, "+IPD,")) != NULL && (stringEndPos = strstr(recBuffer, ":")) != NULL) {
//Advance the start string to the beginning of the length
stringStartPos += strlen("+IPD,");
//Get the end of the client ID
stringEndPos = strstr(stringStartPos, ",");
//Get the client Id
runningClientId = antoi(stringStartPos, stringEndPos - stringStartPos);
//Get the data length
stringStartPos = stringEndPos + 1;
stringEndPos = strstr(stringStartPos, ":");
dataLength = antoi(stringStartPos, stringEndPos - stringStartPos);
if(dataLength > 0) {
//Set the remaining data length
dataRemaining = dataLength;
messageToProcess = RECV_MSG_IPD_MSG;
}
}
}
switch (loopState) {
case ANYTHING: // looking for +IPD, > , busy , n,CONNECTED, n,CLOSED, ERROR, SEND OK
if (ch == '+') {
loopState = IPD;
break;
//Process messages
switch (messageToProcess) {
case RECV_MSG_NONE:
//Check if the buffer position is not zero
if (recBufferPos != 0) {
//No message to process, but buffer has some data already
retVal = INBOUND_BUSY;
}
if (ch=='>') {
if (Diag::WIFI) DIAG(F("[XMIT %d]"),currentReplySize);
for (int i=0;i<currentReplySize;i++) {
int cout=outboundRing->read();
wifiStream->write(cout);
if (Diag::WIFI) StringFormatter::printEscape(cout); // DIAG in disguise
}
clientPendingCIPSEND=-1;
pendingCipsend=false;
loopState=SKIPTOEND;
break;
else {
//No message to process, no data in buffer
retVal = INBOUND_IDLE;
}
if (ch=='R') { // Received ... bytes
loopState=SKIPTOEND;
break;
}
if (ch=='S') { // SEND OK probably
loopState=SKIPTOEND;
break;
}
if (ch=='b') { // This is a busy indicator... probabaly must restart a CIPSEND
pendingCipsend=(clientPendingCIPSEND>=0);
loopState=SKIPTOEND;
break;
}
if (ch>='0' && ch<='9') {
runningClientId=ch-'0';
loopState=GOT_CLIENT_ID;
break;
exitLoop = true;
break;
case RECV_MSG_CRLF:
//Check for a SEND FAIL message
if (strstr(recBuffer, "SEND FAIL") != NULL) {
if (Diag::WIFI) DIAG(F("[SEND FAIL - detected"));
if (cipSendStatus == CIP_SEND_DATA_SENT) {
if (Diag::WIFI) DIAG( F("pendingCipsend = CIP_SEND_NONE"));
cipSendStatus = CIP_SEND_NONE;
}
}
if (ch=='E' || ch=='l') { // ERROR or "link is not valid"
if (clientPendingCIPSEND>=0) {
//Check for a SEND OK message
else if (strstr(recBuffer, "SEND OK") != NULL) {
if (Diag::WIFI) DIAG(F("[SEND OK - detected"));
if (cipSendStatus == CIP_SEND_DATA_SENT) {
if (Diag::WIFI) DIAG( F("pendingCipsend = CIP_SEND_NONE"));
cipSendStatus = CIP_SEND_NONE;
}
}
//Check for an error
else if (strstr(recBuffer, "ERROR") != NULL) {
if (clientPendingCIPSEND >= 0) {
if (Diag::WIFI) DIAG(F("[ERROR detected - purging CIPSEND"));
// A CIPSEND was errored... just toss it away
purgeCurrentCIPSEND();
purgeCurrentCIPSEND();
}
loopState=SKIPTOEND;
break;
}
break;
case IPD: // Looking for I in +IPD
loopState = (ch == 'I') ? IPD1 : SKIPTOEND;
break;
case IPD1: // Looking for P in +IPD
loopState = (ch == 'P') ? IPD2 : SKIPTOEND;
break;
case IPD2: // Looking for D in +IPD
loopState = (ch == 'D') ? IPD3 : SKIPTOEND;
break;
case IPD3: // Looking for , After +IPD
loopState = (ch == ',') ? IPD4_CLIENT : SKIPTOEND;
break;
case IPD4_CLIENT: // reading connection id
if (ch >= '0' || ch <='9'){
runningClientId=ch-'0';
loopState=IPD5;
//Check for an tcp connection closed
else if (strstr(recBuffer, "link is not valid") != NULL) {
if (clientPendingCIPSEND >= 0) {
if (Diag::WIFI) DIAG(F("['link is not valid' detected - purging CIPSEND]"));
purgeCurrentCIPSEND();
}
}
else loopState=SKIPTOEND;
break;
case IPD5: // Looking for , After +IPD,client
loopState = (ch == ',') ? IPD6_LENGTH : SKIPTOEND;
dataLength=0; // ready to start collecting the length
case RECV_MSG_SND_DATA:
if (Diag::WIFI) DIAG(F("[XMIT %d]"), currentReplySize);
for (i = 0; i < currentReplySize; i++)
{
//Read data from the outboundRing and write to to Wifi
int cout = outboundRing->read();
wifiStream->write(cout);
if (Diag::WIFI) StringFormatter::printEscape(cout); // DIAG in disguise
}
//Set CIPSEND flag to -1
clientPendingCIPSEND = -1;
if (Diag::WIFI) DIAG( F("pendingCipsend = CIP_SEND_DATA_SENT"));
cipSendStatus = CIP_SEND_DATA_SENT;
break;
case IPD6_LENGTH: // reading for length
if (ch == ':') {
if (dataLength==0) {
loopState=ANYTHING;
break;
}
if (Diag::WIFI) DIAG(F("Wifi inbound data(%d:%d):"),runningClientId,dataLength);
if (inboundRing->freeSpace()<=(dataLength+1)) {
// This input would overflow the inbound ring, ignore it
loopState=IPD_IGNORE_DATA;
if (Diag::WIFI) DIAG(F("Wifi OVERFLOW IGNORING:"));
break;
}
case RECV_MSG_IPD_MSG:
//Mark the location with the client ID
if (dataRemaining == dataLength) {
if (Diag::WIFI) DIAG(F("Wifi inbound data(%d:%d):"), runningClientId, dataLength);
inboundRing->mark(runningClientId);
loopState=IPD_DATA;
break;
}
dataLength = dataLength * 10 + (ch - '0');
break;
case IPD_DATA: // reading data
inboundRing->write(ch);
dataLength--;
if (dataLength == 0) {
inboundRing->commit();
loopState = ANYTHING;
while (wifiStream->available()) {
//Read next character (data byte)
ch = wifiStream->read();
if (Diag::WIFI) DIAG(F("[+IPD %d]"), dataRemaining);
if (Diag::WIFI) StringFormatter::printEscape(ch);
dataRemaining--;
//Check if we would overflow the inbound ring
if (inboundRing->freeSpace() <= (dataLength + 1))
{
DIAG(F("Wifi OVERFLOW IGNORING:"));
//If we have an overflow, flush the remaining bytes
do{
ch = wifiStream->read();
}while(dataRemaining--);
outboundRing->flush();
inboundRing->flush();
messageToProcess = RECV_MSG_IPD_DONE;
break;
}
else
{
// Write data to inbound ring
inboundRing->write(ch);
// Commit inbound ring data
if (dataRemaining == 0) {
if (Diag::WIFI) DIAG(F("COMMIT"));
if(inboundRing->commit() == false) {
outboundRing->flush();
inboundRing->flush();
}
messageToProcess = RECV_MSG_IPD_DONE;
break;
}
}
}
break;
case IPD_IGNORE_DATA: // ignoring data that would not fit in inbound ring
dataLength--;
if (dataLength == 0) loopState = ANYTHING;
break;
case GOT_CLIENT_ID: // got x before CLOSE or CONNECTED
loopState=(ch==',') ? GOT_CLIENT_ID2: SKIPTOEND;
break;
case GOT_CLIENT_ID2: // got "x,"
if (ch=='C') {
// got "x C" before CLOSE or CONNECTED, or CONNECT FAILED
if (runningClientId==clientPendingCIPSEND) purgeCurrentCIPSEND();
else CommandDistributor::forget(runningClientId);
//Catch-all for messages we don't care about
default:
//Check if the buffer position is not zero
if (recBufferPos != 0)
{
//In process of receiving, return with busy so we can wait for the remaining part of the message
retVal = INBOUND_BUSY;
}
else {
// Not receiving a message so this message was not important
messageToProcess = RECV_MSG_GENERIC;
}
loopState=SKIPTOEND;
break;
case SKIPTOEND: // skipping for /n
if (ch=='\n') loopState=ANYTHING;
break;
} // switch
} // available
return (loopState==ANYTHING) ? INBOUND_IDLE: INBOUND_BUSY;
}
//If we processed a message, clear and get ready for the next message
if (messageToProcess != RECV_MSG_NONE && messageToProcess != RECV_MSG_IPD_MSG) {
//Reset buffer to start processing next message
memset(recBuffer, 0, INBOUND_CMD_BUFFER);
//Reset receive buffer position
recBufferPos = 0;
//Reset message to process
messageToProcess = RECV_MSG_NONE;
}
}
return retVal;
}
void WifiInboundHandler::purgeCurrentCIPSEND() {
// A CIPSEND was sent but errored... or the client closed just toss it away
CommandDistributor::forget(clientPendingCIPSEND);
DIAG(F("Wifi: DROPPING CIPSEND=%d,%d"),clientPendingCIPSEND,currentReplySize);
for (int i=0;i<currentReplySize;i++) outboundRing->read();
pendingCipsend=false;
clientPendingCIPSEND=-1;
// A CIPSEND was sent but errored... or the client closed just toss it away
//CommandDistributor::forget(clientPendingCIPSEND);
DIAG(F("Wifi: DROPPING CIPSEND=%d,%d"), clientPendingCIPSEND, currentReplySize);
for (int i = 0; i < currentReplySize; i++) {
outboundRing->read();
}
//If CIP SEND was sent, flush out by writing 0s */
if (cipSendStatus == CIP_SEND_SENT) {
for (int i = 0; i < currentReplySize; i++) {
int c = 0;
wifiStream->write(c);
}
}
cipSendStatus = CIP_SEND_NONE;
clientPendingCIPSEND = -1;
}
#endif

139
WifiInboundHandler.h
View File

@ -1,24 +1,25 @@
/*
* © 2021 Harald Barth
* © 2021 Fred Decker
* (c) 2021 Fred Decker. All rights reserved.
* (c) 2020 Chris Harlow. All rights reserved.
*
* This file is part of CommandStation-EX
*
* This is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* It is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with CommandStation. If not, see <https://www.gnu.org/licenses/>.
*/
© 2022 Mark Muzzin
© 2021 Harald Barth
© 2021 Fred Decker
(c) 2021 Fred Decker. All rights reserved.
(c) 2020 Chris Harlow. All rights reserved.
This file is part of CommandStation-EX
This is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
It is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with CommandStation. If not, see <https://www.gnu.org/licenses/>.
*/
#ifndef WifiInboundHandler_h
#define WifiInboundHandler_h
@ -27,57 +28,59 @@
#include "DIAG.h"
class WifiInboundHandler {
public:
static void setup(Stream * ESStream);
static void loop();
private:
public:
static void setup(Stream * ESStream);
static void loop();
static WifiInboundHandler * singleton;
enum INBOUND_STATE : byte {
INBOUND_BUSY, // keep calling in loop()
INBOUND_IDLE // Nothing happening, outbound may xcall CIPSEND
};
private:
enum LOOP_STATE : byte {
ANYTHING, // ready for +IPD, n CLOSED, n CONNECTED, busy etc...
SKIPTOEND, // skip to newline
// +IPD,client,length:data
IPD, // got +
IPD1, // got +I
IPD2, // got +IP
IPD3, // got +IPD
IPD4_CLIENT, // got +IPD, reading cient id
IPD5, // got +IPD,c
IPD6_LENGTH, // got +IPD,c, reading length
IPD_DATA, // got +IPD,c,ll,: collecting data
IPD_IGNORE_DATA, // got +IPD,c,ll,: ignoring the data that won't fit inblound Ring
static WifiInboundHandler * singleton;
GOT_CLIENT_ID, // clientid prefix to CONNECTED / CLOSED
GOT_CLIENT_ID2 // clientid prefix to CONNECTED / CLOSED
};
enum INBOUND_STATE : byte {
INBOUND_BUSY, // Keep calling in loop()
INBOUND_IDLE // Nothing happening, outbound may xcall CIPSEND
};
WifiInboundHandler(Stream * ESStream);
void loop1();
INBOUND_STATE loop2();
void purgeCurrentCIPSEND();
Stream * wifiStream;
static const int INBOUND_RING = 512;
static const int OUTBOUND_RING = 2048;
RingStream * inboundRing;
RingStream * outboundRing;
LOOP_STATE loopState=ANYTHING;
int runningClientId; // latest client inbound processing data or CLOSE
int dataLength; // dataLength of +IPD
int clientPendingCIPSEND=-1;
int currentReplySize;
bool pendingCipsend;
enum RECV_MSG_STATE : byte {
RECV_MSG_NONE, // No message to process
RECV_MSG_CRLF, // Message ending in CRLF
RECV_MSG_SND_DATA, // Send data message
RECV_MSG_IPD_MSG, // +IPD Message in progress
RECV_MSG_IPD_DONE, // +IPD Message done
RECV_MSG_GENERIC // Catchall for incoming messages not used
};
enum CIP_SEND_STATUS : byte {
CIP_SEND_NONE,
CIP_SEND_PENDING,
CIP_SEND_SENT,
CIP_SEND_DATA_SENT
};
WifiInboundHandler(Stream * ESStream);
void loop1();
INBOUND_STATE loop2();
void purgeCurrentCIPSEND();
int antoi(char* str, int len);
Stream * wifiStream;
static const int INBOUND_CMD_BUFFER = 32;
static const int INBOUND_RING = 512;
static const int OUTBOUND_RING = 2048;
RingStream * inboundRing;
RingStream * outboundRing;
RECV_MSG_STATE messageToProcess = RECV_MSG_NONE; // Track the current message to process
int runningClientId; // latest client inbound processing data or CLOSE
int dataLength; // dataLength of +IPD
int dataRemaining = 0; // remaining +IPD data to receive
int clientPendingCIPSEND = -1;
int currentReplySize;
CIP_SEND_STATUS cipSendStatus;
char recBuffer[INBOUND_CMD_BUFFER] = {'\0'};
int recBufferPos = 0;
int recBufferWatermark = 0;
};
#endif