muserial.h

// muserial.h
#pragma once
 
#include "ustd_platform.h"
#include "ustd_array.h"
#include "ustd_map.h"
 
#include "scheduler.h"
//#include <Arduino_JSON.h>
 
#ifdef __ATTINY__
#define HardwareSerial TinySoftwareSerial
#endif
 
namespace ustd {
 
class MuSerial {
  private:
    Scheduler *pSched;
    int tID;
 
    String name;
    HardwareSerial *pSerial;
    unsigned long baudRate;
    uint8_t connectionLed;
    unsigned long ledTimer = 0;
 
    enum LinkState { SYNC, HEADER, MSG, MUCRC };
    bool bCheckLink = false;
    uint8_t blockNum = 0;
    LinkState linkState;
    unsigned long lastRead = 0;
    unsigned long lastMsg = 0;
    unsigned long lastPingSent = 0;
    bool linkConnected = false;
    unsigned long readTimeout = 5;          // sec
    unsigned long pingReceiveTimeout = 10;  // sec
    unsigned long pingPeriod = 5;           // sec
    String remoteName = "";
    ustd::array<String> outgoingBlockList;
    ustd::array<String> incomingBlockList;
 
    const uint8_t SOH = 0x01, STX = 0x02, ETX = 0x03, EOT = 0x04;
    const uint8_t VER = 0x01;
 
    enum LinkCmd {
        MUPING,  
        MQTT     
    };
 
    typedef struct t_header {
        uint8_t soh;   
        uint8_t ver;   
        uint8_t num;   
        uint8_t cmd;   
        uint8_t hLen;  
        uint8_t lLen;  
        uint8_t stx;   
        uint8_t pad;   
    } T_HEADER;
 
    typedef struct t_footer {
        uint8_t etx;   
        uint8_t pad2;  
        uint8_t crc;   
        uint8_t eot;   
    } T_FOOTER;
 
  public:
    bool activeLogic = false;  
    unsigned long connectionLedBlinkDurationMs =
        200;  
 
    MuSerial(String name, HardwareSerial *pSerial, unsigned long baudRate = 115200,
             uint8_t connectionLed = -1)
        : name(name), pSerial(pSerial), baudRate(baudRate), connectionLed(connectionLed) {
    }
 
    ~MuSerial() {
    }
 
    void begin(Scheduler *_pSched) {
        pSched = _pSched;
        pSerial->begin(baudRate);
#ifdef __ARDUINO__
        while (!*pSerial) {
        }
#endif
 
        auto ft = [=]() { this->loop(); };
        tID = pSched->add(ft, "serlink", 20000L);  // check every 20ms
        auto fnall = [=](String topic, String msg, String originator) {
            this->subsMsg(topic, msg, originator);
        };
        pSched->subscribe(tID, "#", fnall);
        bCheckLink = true;
        linkState = SYNC;
        if (connectionLed != -1) {
            pinMode(connectionLed, OUTPUT);
            digitalWrite(connectionLed, activeLogic);
        }
        ping();
    }
 
    bool outgoingBlockSet(String topic) {
        for (unsigned int i = 0; i < outgoingBlockList.length(); i++) {
            if (outgoingBlockList[i] == topic)
                return false;
        }
        if (outgoingBlockList.add(topic) == -1)
            return false;
        return true;
    }
 
    bool outgoingBlockRemove(String topic) {
        for (unsigned int i = 0; i < outgoingBlockList.length(); i++) {
            if (outgoingBlockList[i] == topic) {
                if (!outgoingBlockList.erase(i))
                    return false;
                return true;
            }
        }
        return false;
    }
 
    bool incomingBlockSet(String topic) {
        for (unsigned int i = 0; i < incomingBlockList.length(); i++) {
            if (incomingBlockList[i] == topic)
                return false;
        }
        if (incomingBlockList.add(topic) == -1)
            return false;
        return true;
    }
 
    bool incomingBlockRemove(String topic) {
        for (unsigned int i = 0; i < incomingBlockList.length(); i++) {
            if (incomingBlockList[i] == topic) {
                if (!incomingBlockList.erase(i))
                    return false;
                return true;
            }
        }
        return false;
    }
 
  private:
    unsigned char crc(const unsigned char *buf, unsigned int len, unsigned char init = 0) {
        unsigned char c = init;
        for (unsigned int i = 0; i < len; i++)
            c = c ^ buf[i];
        return c;
    }
 
    void ping() {
        char strTime[16];
#if defined(__ARDUINO__) || defined(__ARM__) || defined(__RISC_V__)
        ltoa(pSched->getUptime(), strTime, 15);
#else
        ltoa(time(nullptr), strTime, 15);
#endif
        sendOut(strTime, name, LinkCmd::MUPING);
        lastPingSent = pSched->getUptime();
    }
 
    void handleTime(uint64_t remoteTime) {
        // XXX do maybe something?
    }
 
    void sendOut(String topic, String msg, LinkCmd cmd = LinkCmd::MQTT) {
        T_HEADER th = {};
        T_FOOTER tf = {};
        unsigned char ccrc;
        unsigned char nul = 0x0;
 
        // Serial.println("Sending " + topic + ", " + msg);
        th.soh = SOH;
        th.ver = VER;
        th.num = blockNum++;
        th.cmd = cmd;
        unsigned int len = topic.length() + msg.length() + 2;
        th.hLen = len / 256;
        th.lLen = len % 256;
        th.stx = STX;
 
        tf.etx = ETX;
        tf.eot = EOT;
 
        ccrc = crc((const uint8_t *)&(th.ver), sizeof(th) - 1);
        ccrc = crc((const uint8_t *)topic.c_str(), topic.length(), ccrc);
        ccrc = crc((const uint8_t *)&nul, 1, ccrc);
        ccrc = crc((const uint8_t *)msg.c_str(), msg.length(), ccrc);
        ccrc = crc((const uint8_t *)&nul, 1, ccrc);
        ccrc = crc((const uint8_t *)&tf, 2, ccrc);
 
        tf.crc = ccrc;
 
        pSerial->write((unsigned char *)&th, sizeof(th));
        pSerial->write((unsigned char *)topic.c_str(), topic.length());
        pSerial->write((uint8_t)nul);
        pSerial->write((unsigned char *)msg.c_str(), msg.length());
        pSerial->write((uint8_t)nul);
        pSerial->write((unsigned char *)&tf, sizeof(tf));
    }
 
  private:
    T_HEADER hd;
    unsigned char *pHd;
    uint16_t hLen;
    uint16_t msgLen, curMsg;
    unsigned char *msgBuf = nullptr;
    bool allocated = false;
    T_FOOTER fo;
    unsigned char *pFo;
    uint16_t cLen;
 
    bool internalPub(String topic, String msg) {
        for (unsigned int i = 0; i < incomingBlockList.length(); i++) {
            if (Scheduler::mqttmatch(topic, incomingBlockList[i])) {
                return false;
            }
        }
 
        // Serial.println("In: " + topic);
        String pre2 = remoteName + "/";
        String pre1 = name + "/";
        if (topic.substring(0, pre1.length()) == pre1) {
            topic = topic.substring(pre1.length());
        }
        if (topic.substring(0, pre2.length()) == pre1) {
            topic = topic.substring(pre2.length());
        }
 
        // Serial.println("InPub: " + topic);
        pSched->publish(topic, msg, remoteName);
        return true;
    }
 
    bool ld = false;
    void loop() {
        unsigned char ccrc;
        unsigned char c;
        if (bCheckLink) {
            if (ledTimer) {
                if (timeDiff(ledTimer, millis()) > connectionLedBlinkDurationMs) {
                    ledTimer = 0;
                    if (connectionLed != -1) {
                        digitalWrite(connectionLed, activeLogic);
                    }
                }
            }
            if (pSched->getUptime() - lastPingSent > pingPeriod) {
                ping();
            }
            while (pSerial->available() > 0) {
                c = pSerial->read();
                lastRead = pSched->getUptime();
                switch (linkState) {
                case SYNC:
                    if (c == SOH) {
                        hd.soh = SOH;
                        linkState = HEADER;
                        pHd = (unsigned char *)&hd;
                        hLen = 1;
                    }
                    continue;
                    break;
                case HEADER:
                    pHd[hLen] = c;
                    hLen++;
                    if (hLen == sizeof(hd)) {
                        // XXX: check block number
                        if (hd.ver != VER || hd.stx != STX) {
                            linkState = SYNC;
                        } else {
                            msgLen = 256 * hd.hLen + hd.lLen;
                            if (msgLen < 1024) {
                                msgBuf = (unsigned char *)malloc(msgLen);
                                curMsg = 0;
                                if (msgBuf) {
                                    linkState = MSG;
                                    allocated = true;
                                } else {
                                    linkState = SYNC;
                                }
                            } else {
                                linkState = SYNC;
                            }
                        }
                    }
                    continue;
                    break;
                case MSG:
                    msgBuf[curMsg] = c;
                    ++curMsg;
                    if (curMsg == msgLen) {
                        linkState = MUCRC;
                        pFo = (unsigned char *)&fo;
                        cLen = 0;
                    }
                    continue;
                    break;
                case MUCRC:
                    pFo[cLen] = c;
                    ++cLen;
                    if (cLen == sizeof(fo)) {
                        if (fo.etx != ETX || fo.eot != EOT) {
                            if (allocated && msgBuf != nullptr) {
                                free(msgBuf);
                                msgBuf = nullptr;
                                allocated = false;
                            }
                            linkState = SYNC;
                            continue;
                        } else {
                            ccrc = crc((const uint8_t *)&(hd.ver), sizeof(hd) - 1);
                            ccrc = crc(msgBuf, msgLen, ccrc);
                            ccrc = crc((const uint8_t *)&fo, 2, ccrc);
                            if (ccrc != fo.crc) {
                                if (allocated && msgBuf != nullptr) {
                                    free(msgBuf);
                                    msgBuf = nullptr;
                                    allocated = false;
                                }
                                linkState = SYNC;
                                continue;
                            } else {
                                uint64_t remoteTime = 0;
                                // Serial.println("Msg received");
                                if (strlen((const char *)msgBuf) + 2 <= msgLen) {
                                    lastMsg = pSched->getUptime();
                                    const char *pM =
                                        (const char *)&msgBuf[strlen((const char *)msgBuf) + 1];
                                    if (strlen(pM) + strlen((const char *)msgBuf) + 2 <= msgLen) {
                                        switch ((LinkCmd)hd.cmd) {
                                        case LinkCmd::MUPING:
                                            remoteName = pM;
                                            // XXX: Y2031?
                                            remoteTime = atol((const char *)msgBuf);
                                            handleTime(remoteTime);
                                            lastMsg = pSched->getUptime();
                                            if (connectionLed != -1) {
                                                digitalWrite(connectionLed, !activeLogic);
                                                ledTimer = millis();
                                            }
                                            if (!linkConnected) {
                                                linkConnected = true;
                                                pSched->publish(name + "/link/" + remoteName,
                                                                "connected", name);
                                            }
                                            break;
                                        case LinkCmd::MQTT:
                                            internalPub((const char *)msgBuf, pM);
                                            break;
                                        }
                                    }
                                }
                                if (allocated && msgBuf != nullptr) {
                                    free(msgBuf);
                                    msgBuf = nullptr;
                                    allocated = false;
                                }
                                linkState = SYNC;
                            }
                        }
                    }
                    continue;
                    break;
                }
            }
            if (linkConnected || linkState != SYNC) {
                if (linkState != SYNC) {
                    if ((unsigned long)(pSched->getUptime() - lastRead) > readTimeout) {
                        linkState = SYNC;
                        if (linkConnected) {
                            pSched->publish(name + "/link/" + remoteName, "disconnected", name);
                        }
                        linkConnected = false;
                        if (allocated && msgBuf != nullptr) {
                            free(msgBuf);
                            msgBuf = nullptr;
                            allocated = false;
                        }
                    }
                } else {
                    if ((unsigned long)(pSched->getUptime() - lastMsg) > pingReceiveTimeout) {
                        if (linkConnected) {
                            pSched->publish(name + "/link/" + remoteName, "disconnected", name);
                        }
                        linkConnected = false;
                    }
                }
            }
        }
    }
 
    void subsMsg(String topic, String msg, String originator) {
        if (originator == remoteName) {
            // prevent loops;
            // Serial.println("Loop prevented: " + topic + " - " + msg + " from: " + originator);
            return;
        }
        // Serial.println("MQ-in: " + topic + " - " + msg + " from: " + originator);
        for (unsigned int i = 0; i < outgoingBlockList.length(); i++) {
            if (Scheduler::mqttmatch(topic, outgoingBlockList[i])) {
                // Serial.println("blocked: " + topic + " - " + msg + " from: " + originator);
                return;
            }
        }
        String pre = remoteName + "/";
        if (topic.substring(0, pre.length()) == pre) {
            sendOut(topic, msg);
        } else {
            sendOut(remoteName + "/" + topic, msg);
        }
    };
};
 
}  // namespace ustd