mqtt.h

/*NB_REVISION*/
 
/*NB_COPYRIGHT*/
 
#ifndef  __MQTT_API_H
#define  __MQTT_API_H
 
#include <predef.h>
#include <constants.h>
#include <config_obj.h>
#include <config_fs_bulkstore.h>
#include <nettimer.h>
#include <mqtt/msg_types.h>
#include <mqtt/mqtt_msg_reqs.h>
#include <webclient/http_funcs.h>
#include <json_lexer.h>
#include <pins.h>
#ifdef NB_SSL_SUPPORTED
#include <crypto/NetBurner/NbSslCtx.h>
#endif
 
void UserMain(void*pd); // leaving this out of the group to avoid a documentation mess
 
#define MQTT_WAIT (TICKS_PER_SECOND*10)                  
#define MQTT_TLS_NEGOTIATE_TIMEOUT (TICKS_PER_SECOND*50) 
 
#define MQTT_OBJ_PRIORITY_COUNT (16)                     
#define MQTT_OBJ_PRIORITY_MAX 0                          
#define MQTT_OBJ_PRIORITY_MIN 15                         
 
#define MQTT_MAX_PKT_SIZE (0x2FFFFFFF)                   
#define MQTT_MAX_MAX_SUPPORTED_IN_FLIGHT (256)           
#define MQTT_MAX_SUPPORTED_IN_FLIGHT (16)                
#define MQTT_MAX_HANDLERS_PER_CONNECTION (256)           
#define MQTT_MAX_SUPPORTED_RX_TOPIC_ALIASES (128)        
#define MQTT_DEFAULT_SESSION_EXPIRY_INTERVAL (3600)      
#define MQTT_DEFAULT_KEEP_ALIVE (30)                     
 
#define MQTT_MAX_SUB_IDs     (128)                       
 
namespace MQTT {
enum eResult_t {
    eResult_Success                      = 0, 
 
    eResult_Mqtt_Unspecified             = -Pkt::eReason_Error_Unspecified, 
    eResult_Mqtt_MalformedPkt            = -Pkt::eReason_Error_MalformedPkt, 
    eResult_Mqtt_Protocol                = -Pkt::eReason_Error_Protocol, 
    eResult_Mqtt_Implementation          = -Pkt::eReason_Error_Implementation, 
    eResult_Mqtt_UnsupportedProtoVer     = -Pkt::eReason_Error_UnsupportedProtoVer, 
    eResult_Mqtt_ClientIdInvalid         = -Pkt::eReason_Error_ClientIdInvalid, 
    eResult_Mqtt_UserPassword            = -Pkt::eReason_Error_UserPassword, 
    eResult_Mqtt_NotAuthorized           = -Pkt::eReason_Error_NotAuthorized, 
    eResult_Mqtt_ServerUnavail           = -Pkt::eReason_Error_ServerUnavail, 
    eResult_Mqtt_ServerBusy              = -Pkt::eReason_Error_ServerBusy, 
    eResult_Mqtt_Banned                  = -Pkt::eReason_Error_Banned, 
    eResult_Mqtt_ServerShutdown          = -Pkt::eReason_Error_ServerShutdown, 
    eResult_Mqtt_BadAuthMethod           = -Pkt::eReason_Error_BadAuthMethod, 
    eResult_Mqtt_KATimeout               = -Pkt::eReason_Error_KATimeout, 
    eResult_Mqtt_SessTakeover            = -Pkt::eReason_Error_SessTakeover, 
    eResult_Mqtt_FilterInvalid           = -Pkt::eReason_Error_FilterInvalid, 
    eResult_Mqtt_NameInvalid             = -Pkt::eReason_Error_NameInvalid, 
    eResult_Mqtt_PktId_InUse             = -Pkt::eReason_Error_PktId_InUse, 
    eResult_Mqtt_PktId_NotFound          = -Pkt::eReason_Error_PktId_NotFound, 
    eResult_Mqtt_RcvMax_Exceeded         = -Pkt::eReason_Error_RcvMax_Exceeded, 
    eResult_Mqtt_TopicAlias_Invalid      = -Pkt::eReason_Error_TopicAlias_Invalid, 
    eResult_Mqtt_TooLarge                = -Pkt::eReason_Error_TooLarge, 
    eResult_Mqtt_MsgRate                 = -Pkt::eReason_Error_MsgRate, 
    eResult_Mqtt_Quota                   = -Pkt::eReason_Error_Quota, 
    eResult_Mqtt_Admin                   = -Pkt::eReason_Error_Admin, 
    eResult_Mqtt_PayFmt_Invalid          = -Pkt::eReason_Error_PayFmt_Invalid, 
    eResult_Mqtt_NotSupported_Retain     = -Pkt::eReason_Error_NotSupported_Retain, 
    eResult_Mqtt_NotSupported_QoS        = -Pkt::eReason_Error_NotSupported_QoS, 
    eResult_Mqtt_UseOther                = -Pkt::eReason_Error_UseOther, 
    eResult_Mqtt_ServerMoved             = -Pkt::eReason_Error_ServerMoved, 
    eResult_Mqtt_NotSupported_SharedSubs = -Pkt::eReason_Error_NotSupported_SharedSubs, 
    eResult_Mqtt_ConnRate                = -Pkt::eReason_Error_ConnRate, 
    eResult_Mqtt_MaxConnTime             = -Pkt::eReason_Error_MaxConnTime, 
    eResult_Mqtt_NotSupported_SubIds     = -Pkt::eReason_Error_NotSupported_SubIds, 
    eResult_Mqtt_NotSupported_WildSubs   = -Pkt::eReason_Error_NotSupported_WildSubs, 
    eResult_Mqtt_Transport_Error         = -Pkt::eReason_Error_Transport_Error, 
    eResult_Error_NetErr                 = -Pkt::eReason_Error_Transport_Error, 
    eResult_Socket_TryAgain              = -256, 
    eResult_Socket_NeedDNS               = -257, 
    eResult_Cont_Auth                    = -258, 
    eResult_Error_InvalidURI             = -259, 
    eResult_Error_NoneAvail              = -260, 
    eResult_Error_Malformed              = -261, 
    eResult_Error_Timeout                = -262, 
    eResult_Error_BadProp                = -263, 
    eResult_Error_NoSuchConn             = -264, 
    eResult_Error_BadArg                 = -265, 
};
 
typedef int (*MsgCallback_t)(MQTT::Client *mConn, int mqttFd, MQTT::PacketInfo::Msg *msg, void *ctx);
 
struct TopicHandler {
    NBString topicFilter; 
    MsgCallback_t handleMsg; 
    void *ctx; 
};
 
 
}
 
#include <mqtt/mqtt_internal.h>
#include <mqtt/config_mqtt_obj.h>
 
namespace MQTT {
 
enum {
    eObj_Flag_PubOnWrite          = 0x0001, 
    eObj_Flag_PubQoS_AtLeastOnce  = 0x0002, 
    eObj_Flag_PubQoS_ExactlyOnce  = 0x0004, 
    eObj_Flag_PubRetain           = 0x0008, 
    eObj_Flag_Subscribe           = 0x0010, 
    eObj_Flag_Sub_HandlerOnly     = 0x0020, 
    eObj_Flag_Sub_IncludeLocal    = 0x0040, 
    eObj_Flag_Sub_QoS_AtLeastOnce = 0x0080, 
    eObj_Flag_Sub_QoS_ExactlyOnce = 0x0100, 
    eObj_Flag_Sub_NoLocal         = 0x0200, 
    eObj_Flag_Sub_Retained_IfNew  = 0x0400, 
    eObj_Flag_Sub_Retained_Never  = 0x0800, 
 
    eObj_Flag_WaitForPub          = 0x1000, 
    eObj_Flag_DynTopicName        = 0x2000, 
};
 
typedef enum {
    eObj_Serialize_Disabled     = 0x00, 
    eObj_Serialize_String       = 0x01, 
    eObj_Serialize_JSON         = 0x02, 
    eObj_Serialize_JSON_ValOnly = 0x04, 
    eObj_Serialize_MqttString   = 0x08, 
    eObj_Serialize_BigEndian    = 0x10, 
    eObj_Serialize_LittleEndian = 0x20, 
    eObj_Serialize_NativeEndian = 0x40, 
    eObj_Serialize_CURRENT_FORM = 0x80  
} eObj_Serialize_t;
 
class mqtt_leaf;
namespace internal {
class WrCritObj;
}
 
const char * GetPacketTypeString(Pkt::eType_t pkt);
 
const char * GetReturnCodeString(MQTT::eResult_t result);
 
const char * GetConnectionStatusString(int status);
 
#define MqttEndMarker                                \
    virtual int Missing_ConfigEndMarker(void *&startp) override \
    {                                                  \
        startp = this;                                 \
        return sizeof(*this);                          \
    };
 
 
class Client : public TimeOutManager{
    friend void ::UserMain(void *pd);
public:
    typedef enum {
        eLimit_Size_Tx,         
        eLimit_Size_Rx,         
        eLimit_InFlight_Tx,     
        eLimit_InFlight_Rx,     
        eLimit_Subscriptions,   
        eLimit_SubAliases,      
        eLimit_PublishAliases,  
        eLimit_QoS,             
        eLimit_SessionExpirary, 
    } eSessionLimit;
 
    typedef enum {
        eRateLimit_Publish,     
        eRateLimit_Subscribe,   
    } eRateLimitType;
 
    typedef enum {
        eRateLimitAdj_None, 
        eRateLimitAdj_Set, 
        eRateLimitAdj_Add, 
        eRateLimitAdj_Subtract 
    } eRateLimitAdjustMethod;
 
    typedef enum {
        eConnStat_Error,        
        eConnStat_No_Broker_Configured, 
        eConnStat_Resolving,    
        eConnStat_InProgress,   
        eConnStat_Established,  
        eConnStat_Refused,      
        eConnStat_Disconnected, // Client is disconnected
    } eConnectionStatus;
 
private:
    struct leaf_list_t {
        mqtt_leaf *head;
        mqtt_leaf *tail;
    };
 
    typedef leaf_list_t leaf_queue_t;
 
    static Client *s_clients; // global list of all clients
    static Client *s_clientsTail; // tail of global list of all clients
    Client *pNextClient; // next client in the global list
    leaf_list_t regList; // Registered objects
 
    void dequeue_active(mqtt_leaf **ppNextReq, int activeQueue,
            uint32_t &waiting, leaf_queue_t *q);
 
    OS_CRIT queueLock;
 
    ParsedURI broker;
    NBString clientId;
    NBString username;
    NBString password;
protected:
    NBString lwtTopic;  // Topic Name for Last Will and Testament message
    NBString lwtMessage; // Last Will and Testament message contents
    NBString connMsg;   // Message to Publish to LWT topic after connecting
    NBString disconnMsg; // Message to Publish to LWT topic when intentionally disconnecting
#ifdef NB_SSL_SUPPORTED
    uint8_t *clientCert;
    uint32_t certLen;
    uint8_t *privKey;
    uint32_t privKeyLen;
    NbSslCtx *pSslCtx;
#endif
private:
    // Registered Object message event queues
    uint32_t pubQueueWaiting;
    uint32_t subQueueWaiting;
    leaf_queue_t pubQueue[MQTT_OBJ_PRIORITY_COUNT];
    leaf_queue_t subQueue[MQTT_OBJ_PRIORITY_COUNT];
 
    PublishRequest nextPubReq;
    AckCbCtx nextSubAckCtx;
 
    fd_set rd_fds;
    fd_set wr_fds;
    fd_set er_fds;
 
    int connect();
    int SendConnect();
    // Client has messages to send?
    inline bool ReadyToSend()
    {
        return mSess.txNeedSpace || pubQueueWaiting || subQueueWaiting
            || mSess.pubReqQueueWaiting || mSess.subReqQueue.head;
    }
 
protected:
    ConnectRequest connReq;
    PublishRequest lwtMsg;
 
    internal::Session mSess;
    void initSessionInfo(const ParsedURI &brokerURI, bool bCleanStart, uint16_t maxKeepAlive, PublishRequest *pLwtMsg, RequestProperty *addProps);
    virtual internal::eClientState_t SetState_Rx(internal::eClientState_t newState);
    virtual internal::eClientState_t SetState_Tx(internal::eClientState_t newState);
 
    void RegisterClient();
    void DeregisterClient();
 
    void InterpolateLeafTopics();
 
    void readyConnMsgsFromConfig(config_MqttClient &clientCfg,
            config_MqttLWT *lwtCfg, RequestProperty *addProps);
    Client(config_MqttClient &clientCfg,
            config_MqttLWT *lwtCfg = NULL, RequestProperty *addProps = NULL,
            bool ConnectOnStartup = true);
public:
    Client(const char *brokerURI, const char *pClientId, uint16_t maxKeepAlive = MQTT_DEFAULT_KEEP_ALIVE,
            bool bCleanStart = true, PublishRequest *lwtMsg = NULL,
            const char *pConnMsg = NULL, const char *pDisconnMsg = NULL,
            const char *pUsername = NULL, const char *pPassword = NULL,
            RequestProperty *addProps = NULL, bool ConnectOnStartup = true);
    Client(const ParsedURI *brokerURI, const char *pClientId, uint16_t maxKeepAlive = MQTT_DEFAULT_KEEP_ALIVE,
            bool bCleanStart = true, PublishRequest *lwtMsg = NULL,
            const char *pConnMsg = NULL, const char *pDisconnMsg = NULL,
            const char *pUsername = NULL, const char *pPassword = NULL,
            RequestProperty *addProps = NULL, bool ConnectOnStartup = true);
 
    int Connect(bool bWaitForConnection, const char *pNewBrokerURI,
            ConnAckCb_t notifyCb = NULL, void *notifyCtx = NULL);
 
    int Connect(bool bWaitForConnection = true, const ParsedURI *pNewBrokerURI = NULL,
            ConnAckCb_t notifyCb = NULL, void *notifyCtx = NULL);
 
    int Disconnect();
    void SetConnectNotify(ConnAckCb_t notifyCb, void *notifyCtx);
 
    /*
     * @brief Get the current configured value for various connection limits.
     *
     * @param limit The limit type to get the value of.
     *
     * @returns The value of the requested connection limit;
     * */
    uint32_t GetLimit(eSessionLimit limit);
 
    /*
     * @brief Set the value for various connection limits.
     *
     * @param limit The limit type to get the value of.
     *
     * @returns Whether setting the limit was successful or not.
     * */
    bool     SetLimit(eSessionLimit limit, uint32_t value);
 
    int      SetRateLimit(eRateLimitType type, uint8_t bucketMax, uint8_t fillCount,
                            uint8_t fillTicks, eRateLimitAdjustMethod adjMethod,
                            uint8_t adjCount);
    int      SetRateLimit_Publish(int reqPerSecond);
 
    int      SetRateLimit_Subscribe(int reqPerSecond);
 
    eConnectionStatus GetConnectionStatus();
 
    inline void SetClientCert(uint8_t *cert, uint32_t len, uint8_t *privkey, uint32_t privlen)
    {
#ifdef NB_SSL_SUPPORTED
        clientCert = cert;
        certLen = len;
        privKey = privkey;
        privKeyLen = privlen;
#endif
    }
 
    inline void GetClientCert(uint8_t **cert, uint32_t *len, uint8_t **privkey, uint32_t *privlen)
    {
#ifdef NB_SSL_SUPPORTED
        *cert = clientCert;
        *len = certLen;
        *privkey = privKey;
        *privlen = privKeyLen;
#else
        *cert = NULL;
        *len = 0;
        *privkey = NULL;
        *privlen = 0;
#endif
    }
 
    int QueuePublish(mqtt_leaf &leaf);
 
    int QueueSubscribe(mqtt_leaf &leaf);
 
    int QueueUnsubscribe(mqtt_leaf &leaf);
 
    int QueueAllSubscriptions();
 
    int Publish(const char *topic, uint8_t *pData, uint32_t len, int prio, TickTimeout &t, uint8_t qos = 0);
 
    int Publish(const char *topic, const char *pStr, int prio, TickTimeout &t, uint8_t qos = 0);
 
    int Publish(PublishRequest *req, int prio, TickTimeout &t);
 
    int QueuePublish(PublishRequest *req, int *prio);
 
    int Subscribe(TopicHandler &handler, int *handlerIdx, TickTimeout &t, uint8_t subOpts = Pkt::eSubOpt_NoLocal);
 
    int Subscribe(SubscribeRequest *req, int *handlerIdx, TickTimeout &t);
 
    int QueueSubscribe(SubUnsubQueueMsg *qmsg);
 
    int Unsubscribe(int handlerIdx, TickTimeout &t, const char *topicFilter = NULL);
 
    int Unsubscribe(const char *topicFilter, TickTimeout &t);
 
    int QueueUnsubscribe(SubUnsubQueueMsg *qmsg);
 
    int RegisterTopicHandler(TopicHandler *newHandler, int *handlerIdx);
 
    int RemoveTopicHandler(TopicHandler *handler);
 
    int RemoveTopicHandler(int handlerIdx);
 
    void RegisterMqttObj(mqtt_leaf &leaf, uint32_t interval=0);
 
    void RemoveObj(mqtt_leaf &leaf);
 
    int SendNextPubReq();
 
    int SendNextSubReq();
 
    int AddToSelectSet(fd_set &rd_set, fd_set &wr_set, fd_set &er_set);
    TickTimeout ProcessSelectResult(fd_set &rd_set, fd_set &wr_set, fd_set &er_set);
 
    const ParsedURI & getBrokerURI() { return broker; }
 
protected:
    static volatile OS_TCB *TaskTcb;
    static OS_CRIT s_ListCrit;      // Critical to protect global static list values
    static OS_SEM  s_TaskWakeSem;   // Explicit task wake signalling semaphore
    static void Task(void *);
    friend class internal::WrCritObj;
    friend struct internal::Session;
 
public:
    static void StartTask(uint8_t taskPrio = MQTT_TASK_PRIO);
 
    static inline Client *GetFirstClient() { return s_clients; }
};
 
class ConfiguredClient : public config_MqttClient, public Client {
//    config_MqttClient clientCfg;
    config_MqttLWT    clientCfgLWT{*this,"LWT","Last Will and Testament message configuration"}; 
    MqttEndMarker;
#ifdef NB_SSL_SUPPORTED
    NbSslCtx sslConCtx; 
#endif
    void init(RequestProperty *addProps);
 
    virtual internal::eClientState_t SetState_Tx(internal::eClientState_t newState) override;
public:
    ConfiguredClient(config_obj &owner = sys, const char* name = NULL,
            const char *desc = "Mqtt client connection configuration",
            RequestProperty *addProps = NULL, bool ConnectOnStartup = true);
 
    virtual void CommitTestedValue(uint32_t permission_mask) override;
};
 
typedef void(LeafCallBack)(mqtt_leaf *p, void *pextra);
 
enum ValueTestResult
{
    eOk,
    eBad
};
 
class mqtt_leaf : public TimeOutElement{
public:
    enum eObj_State_t {
        eObj_State_PubReady         = 0x00,
        eObj_State_PubPendTimer     = 0x01,
        eObj_State_PubQueued        = 0x02,
        eObj_State_PubNeedAck       = 0x03,
 
        eObj_State_SubNone          = 0x00,
        eObj_State_SubSendSub       = 0x10,
        eObj_State_SubNeedAck       = 0x20,
        eObj_State_SubReady         = 0x30,
        eObj_State_SubSendUnsub     = 0x40,
        eObj_State_SubUnsubNeedAck  = 0x50,
 
        eObj_State_PubState_MASK    = 0x0F,
        eObj_State_SubState_MASK    = 0xF0,
    };
 
private:
    OS_SEM *pSemDone;
    Client *mClient;
    mqtt_leaf *pParent;
    mqtt_leaf *pSibling;
    mqtt_leaf *pNextLeaf;
 
    mqtt_leaf *pNextPubReq;
    AckCbCtx pubAckCtx;
    mqtt_leaf *pNextSubReq;
    AckCbCtx subAckCtx;
 
    uint16_t subHandlerId;
    uint8_t state;
    uint8_t priority;
 
    static OS_CRIT    s_needClientCrit;
    static mqtt_leaf *s_needClient;
 
protected:
    NBInterpolatedString topicName; 
    eObj_Serialize_t publishForm; 
    uint16_t mFlags; 
 
    mqtt_leaf(const char *topic, Client *mClient, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
 
    mqtt_leaf(mqtt_leaf &parent, const char *topic, Client *mClient, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
 
    /*
     * @brief Set the Client this object is registered with.
     *
     * @param *client   Pointer to the Client to register with.
     *
     * @returns eResult_t return code
     * */
//    int SetClient(Client *client);
 
    /*
     * @brief Get the Client this object is registered with.
     *
     * @returns Pointer to the Client this oject is registered with
     * */
//    Client *GetClient();
 
    virtual mqtt_leaf *GetChildren();
 
    virtual mqtt_leaf *AddChild(mqtt_leaf *child);
 
    virtual mqtt_leaf *RemoveFirstChild();
 
    void ForEachLeaf(LeafCallBack *p, void *pd);
 
    void RootParse(ParsedJsonDataSet &pjs);
 
    bool MatchId(ParsedJsonDataSet &pjs);
 
    int QueuePublish();
 
    int QueueSubscribe();
 
    int QueueUnsubscribe();
 
    int Assign();
 
    uint8_t PubState() { return state & eObj_State_PubState_MASK; }
 
    uint8_t SubState() { return state & eObj_State_SubState_MASK; }
 
    virtual void SetPubState(uint8_t newState);
 
    virtual void SetSubState(uint8_t newState);
 
    virtual int ParseFd(int mqttFd) = 0;
 
    int HandleSubAck(int mqttFd, PacketInfo::Msg *msg);
 
    int HandleUnsubAck(int mqttFd, PacketInfo::Msg *msg);
 
    int HandlePubAck(int mqttFd, PacketInfo::Msg *msg);
public:
    ~mqtt_leaf();
 
    void RemoveFromTree();
 
    bool NameMatch(const char *cp);
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) = 0;
 
    virtual int GetPubTopicName(NBString &s);
 
    virtual int GetSubTopicFilter(NBString &s);
 
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg);
 
    virtual int GetRenderLength(eObj_Serialize_t form = eObj_Serialize_CURRENT_FORM);
 
    int Publish(int waitForCompletion = -1);
 
    void SetPublishInterval(uint32_t intervalPeriodTicks);
 
    virtual void SetSerializationForm(eObj_Serialize_t form);
 
    inline eObj_Serialize_t GetSerializationForm()
        { return publishForm; }
 
    virtual void SetFlags(uint16_t newFlags);
 
    virtual void ClearFlags(uint16_t delFlags);
 
    inline uint16_t GetFlags() { return mFlags; }
 
    inline int getPriority() { return priority; }
 
    static int MqttRxPublish(Client *mClient, int mqttFd, PacketInfo::Msg *msg, void *ctx);
 
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) = 0;
 
    virtual int RenderToFd(eObj_Serialize_t form, int fd) = 0;
    virtual void TimeElementEvent() override;
 
    virtual int Missing_ConfigEndMarker(void *&startp) = 0;
private:
    struct leaf_queue_t {
        mqtt_leaf *head;
        mqtt_leaf *tail;
    };
    static int puback_cb(Client *mSess, void *ctx, PacketInfo::PublishAck *ack);
    static int suback_cb(Client *mSess, void *ctx, PacketInfo::PublishAck *ack);
 
    static void dequeue_active(mqtt_leaf **ppNextReq, int activeQueue,
                                uint32_t &waiting, leaf_queue_t *q);
    friend class Client;
    friend class mqtt_obj;
    friend void mqttParseFd(mqtt_leaf *pl, void *pd);
};
 
class mqtt_obj : public mqtt_leaf {
    mqtt_leaf *pChildren;
protected:
    mqtt_obj(const char *topic, Client *mClient, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_obj(mqtt_leaf &parent, const char *topic, Client *mClient, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    ~mqtt_obj();
 
    mqtt_leaf *FindChild(const char *topic);
 
    virtual mqtt_leaf *GetChildren() override;
 
    virtual mqtt_leaf *AddChild(mqtt_leaf *child) override;
 
    virtual mqtt_leaf *RemoveFirstChild() override;
 
    virtual void    SetPubState(uint8_t newState) override;
 
    virtual void    SetSubState(uint8_t newState) override;
 
    virtual int ParseFd(int mqttFd) override;
 
    virtual int GetSubTopicFilter(NBString &s) override;
 
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
 
    virtual int GetRenderLength(eObj_Serialize_t form = eObj_Serialize_CURRENT_FORM) override;
public:
 
    virtual int Missing_ConfigEndMarker(void *&startp) override
    {
        startp = this;
        return sizeof(*this);
    };
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
 
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
 
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
};
 
class mqtt_int : public mqtt_leaf {
private:
    int32_t val;
protected:
    virtual int ParseFd(int mqttFd) override;
 
public:
    mqtt_int(int val, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_int(const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_int(mqtt_leaf &parent, int val, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_int(mqtt_leaf &parent, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
 
    mqtt_int &operator=(int i);
 
    inline operator int() const { return val; }
    mqtt_int &operator++() { ++val; Assign(); return *this; }
    int32_t operator++(int) { int32_t ret = val++; Assign(); return ret; }
    mqtt_int &operator--() { --val; Assign(); return *this; }
    int32_t operator--(int) { int32_t ret = val--; Assign(); return ret; }
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
 
    // This is named so error messages make sense to user, old name was GetExtent
    virtual int Missing_ConfigEndMarker(void *&startp) override
    {
        startp = this;
        return sizeof(*this);
    };
};
 
class mqtt_uint : public mqtt_leaf {
private:
    uint32_t val;
protected:
    virtual int ParseFd(int mqttFd) override;
 
public:
    mqtt_uint(uint32_t val, const char *topic, Client *client, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_uint(const char *topic, Client *client, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_uint(mqtt_leaf &parent, uint32_t val, const char *topic, Client *client, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_uint(mqtt_leaf &parent, const char *topic, Client *client, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
 
    mqtt_uint &operator=(uint32_t i);
    inline operator unsigned() const { return val; }
    mqtt_uint &operator++() { ++val; Assign(); return *this; }
    uint32_t operator++(int) { uint32_t ret = val++; Assign(); return ret; }
    mqtt_uint &operator--() { --val; Assign(); return *this; }
    uint32_t operator--(int) { uint32_t ret = val--; Assign(); return ret; }
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
 
    virtual int Missing_ConfigEndMarker(void *&startp) override
    {
        startp = this;
        return sizeof(*this);
    };
};
 
class mqtt_float : public mqtt_leaf {
private:
    double val;
    uint8_t serPrecision;
    char serStrFormat;
protected:
    virtual int ParseFd(int mqttFd) override;
 
public:
    mqtt_float(double val, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_float(const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_float(mqtt_leaf &parent, double val, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_float(mqtt_leaf &parent, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
 
    mqtt_float &operator=(double f);
    inline operator double() const { return val; }
 
    void SetStringFormat(char formatChar, uint8_t precision)
    {
        switch (formatChar)
        {
            case 'f':
            case 'F':
            case 'e':
            case 'E':
            case 'g':
            case 'G':
            case 'a':
            case 'A':
                serStrFormat = formatChar;
                break;
            default:
                break;
        }
        serPrecision = precision;
    }
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
 
    virtual int Missing_ConfigEndMarker(void *&startp) override
    {
        startp = this;
        return sizeof(*this);
    };
};
 
class mqtt_string : public mqtt_leaf {
    NBString val;
protected:
    virtual int ParseFd(int mqttFd) override;
 
public:
    mqtt_string(const char *val, const char *topic, Client *mClient = NULL, uint16_t flags = 0);
    mqtt_string(const NBString &val, const char *topic, Client *mClient = NULL, uint16_t flags = 0);
    mqtt_string(const char *topic, Client *mClient = NULL, uint16_t flags = 0);
    mqtt_string(mqtt_leaf &parent, const char *val, const char *topic, Client *mClient = NULL, uint16_t flags = 0);
    mqtt_string(mqtt_leaf &parent, const NBString &val, const char *topic, Client *mClient = NULL, uint16_t flags = 0);
    mqtt_string(mqtt_leaf &parent, const char *topic, Client *mClient = NULL, uint16_t flags = 0);
 
    mqtt_string &operator=(const NBString &s);
    mqtt_string &operator=(const char *s);
 
    inline operator NBString() const { return val; }
    inline const char operator[](size_t pos) const { return val[pos]; }
    inline char &operator[](size_t pos) { return val.posRef(pos); }
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
 
    virtual int Missing_ConfigEndMarker(void *&startp) override
    {
        startp = this;
        return sizeof(*this);
    };
};
 
class mqtt_bool : public mqtt_leaf {
private:
    bool val;
protected:
    virtual int ParseFd(int mqttFd) override;
 
public:
    mqtt_bool(bool val, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_bool(const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_bool(mqtt_leaf &parent, bool val, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_bool(mqtt_leaf &parent, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
 
    mqtt_bool &operator=(bool i);
    inline operator bool() const { return val; }
    bool  operator!() const { return !val; }
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
 
    virtual int Missing_ConfigEndMarker(void *&startp) override
    {
        startp = this;
        return sizeof(*this);
    };
};
 
class mqtt_IO : public mqtt_leaf {
private:
    FastIO pin;
    int fixDir;
    MqttEndMarker;
 
protected:
    virtual int ParseFd(int mqttFd) override;
public:
    mqtt_IO(PinIO ioPin, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
    mqtt_IO(mqtt_leaf &parent, PinIO ioPin, const char *topic, Client *mClient = NULL, uint16_t flags = (eObj_Flag_PubOnWrite | eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal));
 
    virtual void SetFlags(uint16_t newFlags) override;
    virtual void ClearFlags(uint16_t delFlags) override;
 
    int SetFixedDirection(int direction);
 
    mqtt_IO &operator=(bool val);
    inline operator bool() { return pin; }
    bool  operator!() { return !pin; }
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
 
    static void PinNotify(int port_num, int pin, void *ctx);
};
 
class mqtt_IO_Input : public mqtt_IO {
public:
    using mqtt_IO::operator=;
    mqtt_IO_Input(PinIO ioPin, const char *topic, Client *client = NULL, uint16_t flags = 0)
        : mqtt_IO(ioPin, topic, client, flags | eObj_Flag_PubOnWrite)
    {SetFixedDirection(-1);}
    mqtt_IO_Input(mqtt_leaf &parent, PinIO ioPin, const char *topic, Client *client = NULL, uint16_t flags = 0)
        : mqtt_IO(parent, ioPin, topic, client, flags | eObj_Flag_PubOnWrite)
    {SetFixedDirection(-1);}
};
 
class mqtt_IO_Output : public mqtt_IO {
public:
    using mqtt_IO::operator=;
    mqtt_IO_Output(PinIO ioPin, const char *topic, Client *client = NULL, uint16_t flags = 0)
        : mqtt_IO(ioPin, topic, client, flags | (eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal))
    {SetFixedDirection(1);}
    mqtt_IO_Output(mqtt_leaf &parent, PinIO ioPin, const char *topic, Client *client = NULL, uint16_t flags = 0)
        : mqtt_IO(parent, ioPin, topic, client, flags | (eObj_Flag_Subscribe | eObj_Flag_Sub_NoLocal))
    {SetFixedDirection(1);}
};
 
class ConfigExposer : public mqtt_leaf {
public:
    enum eRenderFlag_t {
        eRender_Hidden,         
        eRender_ModifiedOnly,   
        eRender_Pretty          
    };
private:
    config_leaf *pLeaf;
    config_leaf *pendingPubLeaf;
 
    uint32_t renderFlags;
protected:
    virtual int ParseFd(int mqttFd) override;
 
public:
    ConfigExposer(config_leaf &leaf, const char *topicRoot, Client *mClient = NULL, uint16_t initFlags = 0, uint32_t renderFlags = 0);
    ConfigExposer(mqtt_leaf &parent, config_leaf &leaf, const char *topicRoot, Client *mClient = NULL, uint16_t initFlags = 0, uint32_t renderFlags = 0);
//    MqttConfigExposer(const char *topicRoot, Session *mClient);
//    MqttConfigExposer(const char *topicRoot, Session *mClient);
 
    virtual int GetPubTopicName(NBString &s) override;
    virtual int GetSubTopicFilter(NBString &s) override;
 
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
    virtual int GetRenderLength(eObj_Serialize_t form = eObj_Serialize_CURRENT_FORM) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
 
    virtual int Missing_ConfigEndMarker(void *&startp) override
    {
        startp = this;
        return sizeof(*this);
    };
};
 
typedef int (*SubHandlerCb)(int mqttFd, const char *pathExt, uint32_t msgLen);
 
class SubHandler : public mqtt_leaf {
    SubHandlerCb handler;
    MqttEndMarker;
protected:
    virtual int ParseFd(int mqttFd) override;
    virtual int RenderToBuffer(eObj_Serialize_t form, uint8_t *buf, int buflen) override;
    virtual int RenderToFd(eObj_Serialize_t form, int fd) override;
public:
    SubHandler(SubHandlerCb handler, const char * topicFilter, Client *mClient = NULL, uint16_t flags = 0);
    virtual ValueTestResult TestOrSetValue(ParsedJsonDataSet &pjs, bool setVal) override;
    virtual int SubscribeEvent(int mqttFd, PacketInfo::Msg *msg) override;
 
};
 
class PropMarker {
    uint32_t markers[8];
    int num;
public:
    PropMarker()
        : markers{0,0,0,0,0,0,0,0}, num(0)
    {
    }
 
    PropMarker(uint32_t val_0, uint32_t val_1, uint32_t val_2, uint32_t val_3,
            uint32_t val_4, uint32_t val_5, uint32_t val_6, uint32_t val_7)
        : markers{val_0, val_1, val_2, val_3, val_4, val_5, val_6, val_7}, num(0)
    {
        for (int i = 0; i < 8; i++)
            for (uint32_t v = 1; v; v <<= 1)
                if (markers[i]&v)
                    num++;
    }
 
    inline void reset() { for(int i = 0; i < 8; i++) markers[i] = 0; }
 
    inline void set(int prop)
    {
        prop--;
        num += ((markers[prop>>5] & (1UL << (prop&0x1F))) == 0);
        markers[prop>>5] |= 1UL << (prop&0x1F);
    }
    inline void clr(int prop)
    {
        prop--;
        num -= ((markers[prop>>5] & (1UL << (prop&0x1F))) != 0);
        markers[prop>>5] &= ~(1UL << (prop&0x1F));
    }
    inline bool isset(int prop) { prop--; return ((markers[prop>>5] & (1UL << (prop&0x1F))) != 0); }
    inline int  count() { return num; }
    void show();
};
 
 
 
}
#endif   /* ----- #ifndef __MQTT_API_H  ----- */