diff --git a/CommandDistributor.cpp b/CommandDistributor.cpp index 4d13f40..de4e8bf 100644 --- a/CommandDistributor.cpp +++ b/CommandDistributor.cpp @@ -22,10 +22,10 @@ DCCEXParser * CommandDistributor::parser=0; -void CommandDistributor::parse(byte clientId,byte * buffer, Print * streamer) { +void CommandDistributor::parse(byte clientId,byte * buffer, RingStream * streamer) { if (buffer[0] == '<') { if (!parser) parser = new DCCEXParser(); parser->parse(streamer, buffer, true); // tell JMRI parser that ACKS are blocking because we can't handle the async } - else WiThrottle::getThrottle(clientId)->parse(*streamer, buffer); + else WiThrottle::getThrottle(clientId)->parse(streamer, buffer); } diff --git a/CommandDistributor.h b/CommandDistributor.h index 2bec81b..155ada0 100644 --- a/CommandDistributor.h +++ b/CommandDistributor.h @@ -19,11 +19,12 @@ #ifndef CommandDistributor_h #define CommandDistributor_h #include "DCCEXParser.h" +#include "RingStream.h" class CommandDistributor { public : - static void parse(byte clientId,byte* buffer, Print * streamer); + static void parse(byte clientId,byte* buffer, RingStream * streamer); private: static DCCEXParser * parser; }; diff --git a/RingStream.cpp b/RingStream.cpp index eee0753..911bb80 100644 --- a/RingStream.cpp +++ b/RingStream.cpp @@ -28,38 +28,72 @@ RingStream::RingStream( const uint16_t len) _pos_read=0; _buffer[0]=0; _overflow=false; + _mark=0; + _count=0; } -size_t RingStream::write(uint8_t byte) { +size_t RingStream::write(uint8_t b) { if (_overflow) return 0; - _buffer[_pos_write] = byte; + _buffer[_pos_write] = b; ++_pos_write; - if (_pos_write>=_len) _pos_write=0; + if (_pos_write==_len) _pos_write=0; if (_pos_write==_pos_read) { _overflow=true; - DIAG(F("\nRingStream(%d) OVERFLOW %d %d \n"),_len, _pos_write, _pos_read); return 0; } + _count++; return 1; } int RingStream::read() { - if (_pos_read==_pos_write) return -1; + if ((_pos_read==_pos_write) && !_overflow) return -1; // empty byte b=_buffer[_pos_read]; _pos_read++; - if (_pos_read>=_len) _pos_read=0; + if (_pos_read==_len) _pos_read=0; _overflow=false; return b; } int RingStream::count() { - int peek=_pos_read; - int counter=0; - while(_buffer[peek]) { - counter++; - peek++; - if (peek >= _len) peek=0; + return (read()<<8) | read(); } - return counter; + +int RingStream::freeSpace() { + // allow space for client flag and length bytes + if (_pos_read>_pos_write) return _pos_read-_pos_write-3; + else return _len - _pos_write + _pos_read-3; +} + + +// mark start of message with client id (0...9) +void RingStream::mark(uint8_t b) { + _mark=_pos_write; + write(b); // client id + write((uint8_t)0); // count MSB placemarker + write((uint8_t)0); // count LSB placemarker + _count=0; +} + +bool RingStream::commit() { + if (_overflow) { + DIAG(F("\nRingStream(%d) commit(%d) OVERFLOW\n"),_len, _count); + // just throw it away + _pos_write=_mark; + _overflow=false; + return false; // commit failed } + if (_count==0) { + // ignore empty response + _pos_write=_mark; + return true; // true=commit ok + } + // Go back to the _mark and inject the count 1 byte later + _mark++; + if (_mark==_len) _mark=0; + _buffer[_mark]=highByte(_count); + _mark++; + if (_mark==_len) _mark=0; + _buffer[_mark]=lowByte(_count); + return true; // commit worked +} diff --git a/RingStream.h b/RingStream.h index 7a129e0..b32e062 100644 --- a/RingStream.h +++ b/RingStream.h @@ -30,12 +30,17 @@ class RingStream : public Print { using Print::write; int read(); int count(); + int freeSpace(); + void mark(uint8_t b); + bool commit(); private: int _len; int _pos_write; int _pos_read; bool _overflow; + int _mark; + int _count; byte * _buffer; }; diff --git a/WiThrottle.cpp b/WiThrottle.cpp index ac503a0..61f7f92 100644 --- a/WiThrottle.cpp +++ b/WiThrottle.cpp @@ -99,7 +99,7 @@ WiThrottle::~WiThrottle() { } } -void WiThrottle::parse(Print & stream, byte * cmdx) { +void WiThrottle::parse(RingStream * stream, byte * cmdx) { byte * cmd=cmdx; @@ -205,7 +205,7 @@ int WiThrottle::getLocoId(byte * cmd) { if (cmd[0]!='L' && cmd[0]!='S') return 0; // should not match any locos return getInt(cmd+1); } -void WiThrottle::multithrottle(Print & stream, byte * cmd){ +void WiThrottle::multithrottle(RingStream * stream, byte * cmd){ char throttleChar=cmd[1]; int locoid=getLocoId(cmd+3); // -1 for * byte * aval=cmd; @@ -256,7 +256,7 @@ void WiThrottle::multithrottle(Print & stream, byte * cmd){ } } -void WiThrottle::locoAction(Print & stream, byte* aval, char throttleChar, int cab){ +void WiThrottle::locoAction(RingStream * stream, byte* aval, char throttleChar, int cab){ // Note cab=-1 for all cabs in the consist called throttleChar. // DIAG(F("\nLoco Action aval=%c%c throttleChar=%c, cab=%d"), aval[0],aval[1],throttleChar, cab); switch (aval[0]) { @@ -334,10 +334,19 @@ int WiThrottle::WiTToDCCSpeed(int WiTSpeed) { return WiTSpeed + 1; //offset others by 1 } -void WiThrottle::loop() { +void WiThrottle::loop(RingStream * stream) { // for each WiThrottle, check the heartbeat for (WiThrottle* wt=firstThrottle; wt!=NULL ; wt=wt->nextThrottle) wt->checkHeartbeat(); + + // TODO... any broadcasts to be done + (void)stream; + /* MUST follow this model in this loop. + * stream->mark(); + * send 1 digit client id, and any data + * stream->commit() + */ + } void WiThrottle::checkHeartbeat() { diff --git a/WiThrottle.h b/WiThrottle.h index c1b5bba..161020d 100644 --- a/WiThrottle.h +++ b/WiThrottle.h @@ -19,6 +19,7 @@ #ifndef WiThrottle_h #define WiThrottle_h +#include "RingStream.h" struct MYLOCO { char throttle; //indicates which throttle letter on client, often '0','1' or '2' @@ -27,8 +28,8 @@ struct MYLOCO { class WiThrottle { public: - static void loop(); - void parse(Print & stream, byte * cmd); + static void loop(RingStream * stream); + void parse(RingStream * stream, byte * cmd); static WiThrottle* getThrottle( int wifiClient); static bool annotateLeftRight; private: @@ -56,9 +57,9 @@ class WiThrottle { bool lastPowerState; // last power state sent to this client int DCCToWiTSpeed(int DCCSpeed); int WiTToDCCSpeed(int WiTSpeed); - void multithrottle(Print & stream, byte * cmd); - void locoAction(Print & stream, byte* aval, char throttleChar, int cab); - void accessory(Print & stream, byte* cmd); + void multithrottle(RingStream * stream, byte * cmd); + void locoAction(RingStream * stream, byte* aval, char throttleChar, int cab); + void accessory(RingStream *, byte* cmd); void checkHeartbeat(); }; #endif diff --git a/WifiInboundHandler.cpp b/WifiInboundHandler.cpp index 8a779ee..2463921 100644 --- a/WifiInboundHandler.cpp +++ b/WifiInboundHandler.cpp @@ -31,20 +31,17 @@ 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) { - int next=outboundRing->read(); - if (next>=0) { + clientPendingCIPSEND=outboundRing->read(); + if (clientPendingCIPSEND>=0) { currentReplySize=outboundRing->count(); - if (currentReplySize==0) { - outboundRing->read(); // drop end marker - } - else { - clientPendingCIPSEND=next-'0'; // convert back to int - pendingCipsend=true; - } + pendingCipsend=true; } - } + } + if (pendingCipsend) { if (Diag::WIFI) DIAG( F("\nWiFi: [[CIPSEND=%d,%d]]"), clientPendingCIPSEND, currentReplySize); @@ -55,19 +52,20 @@ void WifiInboundHandler::loop1() { // if something waiting to execute, we can call it - int next=inboundRing->read(); - if (next>0) { - int clientId=next-'0'; //convert char to int + int clientId=inboundRing->read(); + if (clientId>=0) { int count=inboundRing->count(); - if (Diag::WIFI) DIAG(F("\nExec waiting %d %d:"),clientId,count); + if (Diag::WIFI) DIAG(F("\nWifi EXEC: %d %d:"),clientId,count); byte cmd[count+1]; for (int i=0;iread(); cmd[count]=0; if (Diag::WIFI) DIAG(F("%e\n"),cmd); - outboundRing->print(clientId); + outboundRing->mark(clientId); // remember start of outbound data CommandDistributor::parse(clientId,cmd,outboundRing); - outboundRing->write((byte)0); + // 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 + outboundRing->commit(); return; } } @@ -87,7 +85,7 @@ WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() { } switch (loopState) { - case ANYTHING: // looking for +IPD, > , busy , n,CONNECTED, n,CLOSED + case ANYTHING: // looking for +IPD, > , busy , n,CONNECTED, n,CLOSED, ERROR, SEND OK if (ch == '+') { loopState = IPD; @@ -95,9 +93,12 @@ WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() { } if (ch=='>') { - if (Diag::WIFI) DIAG(F("[[XMIT %d]]"),currentReplySize); - for (int i=0;iwrite(outboundRing->read()); - outboundRing->read(); // drop the end marker + if (Diag::WIFI) DIAG(F("[XMIT %d]"),currentReplySize); + for (int i=0;iread(); + wifiStream->write(cout); + if (Diag::WIFI) StringFormatter::printEscape(cout); // DIAG in disguise + } clientPendingCIPSEND=-1; pendingCipsend=false; loopState=SKIPTOEND; @@ -108,6 +109,11 @@ WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() { 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); @@ -120,6 +126,15 @@ WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() { loopState=GOT_CLIENT_ID; break; } + + if (ch=='E' || ch=='l') { // ERROR or "link is not valid" + if (clientPendingCIPSEND>=0) { + // A CIPSEND was errored... just toss it away + purgeCurrentCIPSEND(); + } + loopState=SKIPTOEND; + break; + } break; @@ -159,7 +174,13 @@ WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() { break; } if (Diag::WIFI) DIAG(F("\nWifi inbound data(%d:%d):"),runningClientId,dataLength); - inboundRing->print(runningClientId); // prefix inbound with client id + if (inboundRing->freeSpace()<=(dataLength+1)) { + // This input would overflow the inbound ring, ignore it + loopState=IPD_IGNORE_DATA; + if (Diag::WIFI) DIAG(F("\nWifi OVERFLOW IGNORING:")); + break; + } + inboundRing->mark(runningClientId); loopState=IPD_DATA; break; } @@ -167,32 +188,30 @@ WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() { break; case IPD_DATA: // reading data - inboundRing->write(ch); + inboundRing->write(ch); dataLength--; if (dataLength == 0) { - inboundRing->write((byte)0); + inboundRing->commit(); loopState = ANYTHING; } 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," before CLOSE or CONNECTED - loopState=(ch=='C') ? GOT_CLIENT_ID3: SKIPTOEND; + case GOT_CLIENT_ID2: // got "x," + if (ch=='C') { + // got "x C" before CLOSE or CONNECTED, or CONNECT FAILED + if (runningClientId==clientPendingCIPSEND) purgeCurrentCIPSEND(); + } + loopState=SKIPTOEND; break; - - case GOT_CLIENT_ID3: // got "x C" before CLOSE or CONNECTED (which is ignored) - if(ch=='L') { - // CLOSE - if (runningClientId==clientPendingCIPSEND) { - // clear the outbound for this client - for (int i=0;i<=currentReplySize;i++) outboundRing->read(); - } - } - loopState=SKIPTOEND; - break; case SKIPTOEND: // skipping for /n if (ch=='\n') loopState=ANYTHING; @@ -201,3 +220,11 @@ WifiInboundHandler::INBOUND_STATE WifiInboundHandler::loop2() { } // available return (loopState==ANYTHING) ? INBOUND_IDLE: INBOUND_BUSY; } + +void WifiInboundHandler::purgeCurrentCIPSEND() { + // A CIPSEND was sent but errored... or the client closed just toss it away + if (Diag::WIFI) DIAG(F("Wifi: DROPPING CIPSEND=%d,%d\n"),clientPendingCIPSEND,currentReplySize); + for (int i=0;i<=currentReplySize;i++) outboundRing->read(); + pendingCipsend=false; + clientPendingCIPSEND=-1; +} diff --git a/WifiInboundHandler.h b/WifiInboundHandler.h index fa41aa4..1979565 100644 --- a/WifiInboundHandler.h +++ b/WifiInboundHandler.h @@ -2,6 +2,7 @@ #define WifiInboundHandler_h #include "RingStream.h" +#include "WiThrottle.h" #include "DIAG.h" class WifiInboundHandler { @@ -32,6 +33,7 @@ class WifiInboundHandler { 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 GOT_CLIENT_ID, // clientid prefix to CONNECTED / CLOSED GOT_CLIENT_ID2, // clientid prefix to CONNECTED / CLOSED @@ -42,10 +44,11 @@ class WifiInboundHandler { WifiInboundHandler(Stream * ESStream); void loop1(); INBOUND_STATE loop2(); + void purgeCurrentCIPSEND(); Stream * wifiStream; - static const int INBOUND_RING = 200; - static const int OUTBOUND_RING = 1024; + static const int INBOUND_RING = 512; + static const int OUTBOUND_RING = 2048; RingStream * inboundRing; RingStream * outboundRing; diff --git a/WifiInterface.cpp b/WifiInterface.cpp index b718b4b..be17b02 100644 --- a/WifiInterface.cpp +++ b/WifiInterface.cpp @@ -22,7 +22,7 @@ #include #include "DIAG.h" #include "StringFormatter.h" -#include "WiThrottle.h" + #include "WifiInboundHandler.h" const char PROGMEM READY_SEARCH[] = "\r\nready\r\n"; @@ -90,6 +90,11 @@ bool WifiInterface::setup(long serial_link_speed, } #endif +DCCEXParser::setAtCommandCallback(ATCommand); + +// CAUTION... ONLY CALL THIS ONCE +WifiInboundHandler::setup(wifiStream); + return wifiUp; } @@ -109,8 +114,6 @@ bool WifiInterface::setup(Stream & setupStream, const __FlashStringHelper* SSid checkForOK(200, OK_SEARCH, true); } - DCCEXParser::setAtCommandCallback(ATCommand); - WifiInboundHandler::setup(wifiStream); DIAG(F("\n++ Wifi Setup %S ++\n"), connected ? F("OK") : F("FAILED")); return connected; @@ -297,7 +300,6 @@ bool WifiInterface::checkForOK( const unsigned int timeout, const char * waitfor void WifiInterface::loop() { if (connected) { - WiThrottle::loop(); WifiInboundHandler::loop(); } }