PubSubClient.h

Go to the documentation of this file.

 
#ifndef PubSubClient_h
#define PubSubClient_h
 
#include <Arduino.h>
 
#include "Client.h"
#include "IPAddress.h"
#include "Stream.h"
 
#define MQTT_VERSION_3_1   3  
#define MQTT_VERSION_3_1_1 4  
 
//< @note The following #define directives can be used to configure the library.

#ifndef MQTT_VERSION
#define MQTT_VERSION MQTT_VERSION_3_1_1
#endif

#ifndef MQTT_MAX_POSSIBLE_PACKET_SIZE
#define MQTT_MAX_POSSIBLE_PACKET_SIZE ((size_t)268435455)  // might be limited to 65535 if size_t is 16-bit (unsigned int)
#endif

#ifndef MQTT_MAX_PACKET_SIZE
#define MQTT_MAX_PACKET_SIZE 256
#endif

#ifndef MQTT_KEEPALIVE
#define MQTT_KEEPALIVE 15
#endif

#ifndef MQTT_SOCKET_TIMEOUT
#define MQTT_SOCKET_TIMEOUT 15
#endif

#ifndef MQTT_MAX_TRANSFER_SIZE  // just a hack that it gets shown in Doxygen
#define MQTT_MAX_TRANSFER_SIZE 80
#undef MQTT_MAX_TRANSFER_SIZE
#endif

#define MQTT_CONNECTION_TIMEOUT      -4  
#define MQTT_CONNECTION_LOST         -3  
#define MQTT_CONNECT_FAILED          -2  
#define MQTT_DISCONNECTED            -1  
#define MQTT_CONNECTED               0   
#define MQTT_CONNECT_BAD_PROTOCOL    1   
#define MQTT_CONNECT_BAD_CLIENT_ID   2   
#define MQTT_CONNECT_UNAVAILABLE     3   
#define MQTT_CONNECT_BAD_CREDENTIALS 4   
#define MQTT_CONNECT_UNAUTHORIZED    5   

#define MQTTRETAINED    1        // Retained flag in the header
#define MQTTCONNECT     1 << 4   // Client request to connect to Server
#define MQTTCONNACK     2 << 4   // Connect Acknowledgment
#define MQTTPUBLISH     3 << 4   // Publish message
#define MQTTPUBACK      4 << 4   // Publish Acknowledgment
#define MQTTPUBREC      5 << 4   // Publish Received (assured delivery part 1)
#define MQTTPUBREL      6 << 4   // Publish Release (assured delivery part 2)
#define MQTTPUBCOMP     7 << 4   // Publish Complete (assured delivery part 3)
#define MQTTSUBSCRIBE   8 << 4   // Client Subscribe request
#define MQTTSUBACK      9 << 4   // Subscribe Acknowledgment
#define MQTTUNSUBSCRIBE 10 << 4  // Client Unsubscribe request
#define MQTTUNSUBACK    11 << 4  // Unsubscribe Acknowledgment
#define MQTTPINGREQ     12 << 4  // PING Request
#define MQTTPINGRESP    13 << 4  // PING Response
#define MQTTDISCONNECT  14 << 4  // Client is Disconnecting
#define MQTTRESERVED    15 << 4  // Reserved

#define MQTT_QOS0 ((uint8_t)0)  
#define MQTT_QOS1 ((uint8_t)1)  
#define MQTT_QOS2 ((uint8_t)2)  
#define MQTT_QOS_GET_HDR(qos) (((qos) & 0x03) << 1) // Get QoS header bits from QoS value
#define MQTT_HDR_GET_QOS(header) (((header) & 0x06 ) >> 1) // Get QoS value from MQTT header


// Maximum size of fixed header and variable length size header
#define MQTT_MAX_HEADER_SIZE 5
// Minimal buffer size that can be handled, used to check if the buffer size is sufficient in setBufferSize().
// MQTT_MAX_HEADER_SIZE (5) + protocol (9) + flags (1) + keepalive (2) covers a minmal CONNECT message
#define MQTT_MIN_BUFFER_SIZE (MQTT_MAX_HEADER_SIZE + 9 + 1 + 2)


#if defined(__has_include) && __has_include(<functional>) && !defined(NOFUNCTIONAL)
#include <functional>
#define MQTT_CALLBACK_SIGNATURE std::function<void(char* topic, uint8_t* payload, size_t plength)> callback
#else
#define MQTT_CALLBACK_SIGNATURE void (*callback)(char* topic, uint8_t* payload, size_t plength)
#endif

#ifdef DEBUG_ESP_PORT
#ifdef DEBUG_PUBSUBCLIENT
#define DEBUG_PSC_PRINTF(fmt, ...) DEBUG_ESP_PORT.printf(("PubSubClient: " fmt), ##__VA_ARGS__)
#else
#define DEBUG_PSC_PRINTF(...)
#endif
 
#define ERROR_PSC_PRINTF(fmt, ...) DEBUG_ESP_PORT.printf(("PubSubClient error: " fmt), ##__VA_ARGS__)
#define ERROR_PSC_PRINTF_P(fmt, ...) DEBUG_ESP_PORT.printf_P(PSTR("PubSubClient error: " fmt), ##__VA_ARGS__)
#else  // DEBUG_ESP_PORT
#ifndef DEBUG_PSC_PRINTF
#define DEBUG_PSC_PRINTF(...)
#endif
#ifndef ERROR_PSC_PRINTF
#define ERROR_PSC_PRINTF(fmt, ...)
#endif
#ifndef ERROR_PSC_PRINTF_P
#define ERROR_PSC_PRINTF_P(fmt, ...)
#endif
#endif

class PubSubClient : public Print {
   private:
    Client* _client{};
    uint8_t* _buffer{};
    size_t _bufferSize{};
    size_t _bufferWritePos{};
    unsigned long _keepAliveMillis{};
    unsigned long _socketTimeoutMillis{};
    uint16_t _nextMsgId{};
    unsigned long _lastOutActivity{};
    unsigned long _lastInActivity{};
    bool _pingOutstanding{};
    MQTT_CALLBACK_SIGNATURE{};
    IPAddress _ip{};
    char* _domain{};
    uint16_t _port{};
    Stream* _stream{};
    int _state{MQTT_DISCONNECTED};
 
    size_t readPacket(uint8_t* hdrLen);
    bool handlePacket(uint8_t hdrLen, size_t len);
    bool readByte(uint8_t* result);
    bool readByte(uint8_t* result, size_t* pos);
    uint8_t buildHeader(uint8_t header, size_t length);
    bool writeControlPacket(uint8_t header, size_t length);
    size_t writeBuffer(size_t pos, size_t size);
    size_t writeStringImpl(bool progmem, const char* string, size_t pos);
    size_t writeString(const char* string, size_t pos);
    size_t writeNextMsgId(size_t pos);
 
    bool beginPublishImpl(bool progmem, const char* topic, size_t plength, uint8_t qos, bool retained);
    bool subscribeImpl(bool progmem, const char* topic, uint8_t qos);
    bool unsubscribeImpl(bool progmem, const char* topic);
 
    // Add to buffer and flush if full (only to be used with beginPublish/endPublish)
    size_t appendBuffer(uint8_t data);
    size_t flushBuffer();
 
   public:
    PubSubClient();

    PubSubClient(Client& client);

    PubSubClient(IPAddress addr, uint16_t port, Client& client);

    PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream);

    PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client);

    PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream);

    PubSubClient(uint8_t* ip, uint16_t port, Client& client);

    PubSubClient(uint8_t* ip, uint16_t port, Client& client, Stream& stream);

    PubSubClient(uint8_t* ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client);

    PubSubClient(uint8_t* ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream);

    PubSubClient(const char* domain, uint16_t port, Client& client);

    PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream);

    PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client);

    PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream);

    ~PubSubClient();

    PubSubClient& setServer(IPAddress ip, uint16_t port);

    PubSubClient& setServer(uint8_t* ip, uint16_t port);

    PubSubClient& setServer(const char* domain, uint16_t port);

    PubSubClient& setCallback(MQTT_CALLBACK_SIGNATURE);

    PubSubClient& setClient(Client& client);

    PubSubClient& setStream(Stream& stream);

    PubSubClient& setKeepAlive(uint16_t keepAlive);

    PubSubClient& setSocketTimeout(uint16_t timeout);

    bool setBufferSize(size_t size);

    size_t getBufferSize();

    inline bool connect(const char* id) {
        return connect(id, nullptr, nullptr, nullptr, MQTT_QOS0, false, nullptr, true);
    }

    inline bool connect(const char* id, const char* user, const char* pass) {
        return connect(id, user, pass, nullptr, MQTT_QOS0, false, nullptr, true);
    }

    inline bool connect(const char* id, const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage) {
        return connect(id, nullptr, nullptr, willTopic, willQos, willRetain, willMessage, true);
    }

    inline bool connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, bool willRetain,
                        const char* willMessage) {
        return connect(id, user, pass, willTopic, willQos, willRetain, willMessage, true);
    }

    bool connect(const char* id, const char* user, const char* pass, const char* willTopic, uint8_t willQos, bool willRetain, const char* willMessage,
                 bool cleanSession);

    void disconnect();

    inline bool publish(const char* topic, const char* payload) {
        return publish(topic, payload, MQTT_QOS0, false);
    }

    inline bool publish(const char* topic, const char* payload, bool retained) {
        return publish(topic, payload, MQTT_QOS0, retained);
    }

    inline bool publish(const char* topic, const char* payload, uint8_t qos, bool retained) {
        return publish(topic, reinterpret_cast<const uint8_t*>(payload), payload ? strlen(payload) : 0, qos, retained);
    }

    inline bool publish(const __FlashStringHelper* topic, const char* payload, uint8_t qos, bool retained) {
        return publish(topic, reinterpret_cast<const uint8_t*>(payload), payload ? strlen(payload) : 0, qos, retained);
    }

    inline bool publish(const __FlashStringHelper* topic, const __FlashStringHelper* payload, uint8_t qos, bool retained) {
        return publish_P(topic, reinterpret_cast<const uint8_t*>(payload), payload ? strlen_P(reinterpret_cast<const char*>(payload)) : 0, qos, retained);
    }

    inline bool publish(const char* topic, const uint8_t* payload, size_t plength) {
        return publish(topic, payload, plength, MQTT_QOS0, false);
    }

    inline bool publish(const char* topic, const uint8_t* payload, size_t plength, bool retained) {
        return publish(topic, payload, plength, MQTT_QOS0, retained);
    }

    bool publish(const char* topic, const uint8_t* payload, size_t plength, uint8_t qos, bool retained);

    bool publish(const __FlashStringHelper* topic, const uint8_t* payload, size_t plength, uint8_t qos, bool retained);

    inline bool publish_P(const char* topic, PGM_P payload, bool retained) {
        return publish_P(topic, payload, MQTT_QOS0, retained);
    }

    inline bool publish_P(const char* topic, PGM_P payload, uint8_t qos, bool retained) {
        return publish_P(topic, reinterpret_cast<const uint8_t*>(payload), payload ? strlen_P(payload) : 0, qos, retained);
    }

    bool publish_P(const __FlashStringHelper* topic, PGM_P payload, uint8_t qos, bool retained) {
        return publish_P(topic, reinterpret_cast<const uint8_t*>(payload), payload ? strlen_P(payload) : 0, qos, retained);
    }

    inline bool publish_P(const char* topic, const uint8_t* payload, size_t plength, bool retained) {
        return publish_P(topic, payload, plength, MQTT_QOS0, retained);
    }

    bool publish_P(const char* topic, const uint8_t* payload, size_t plength, uint8_t qos, bool retained);

    bool publish_P(const __FlashStringHelper* topic, const uint8_t* payload, size_t plength, uint8_t qos, bool retained);

    inline bool beginPublish(const char* topic, size_t plength, bool retained) {
        return beginPublishImpl(false, topic, plength, MQTT_QOS0, retained);
    }

    inline bool beginPublish(const char* topic, size_t plength, uint8_t qos, bool retained) {
        return beginPublishImpl(false, topic, plength, qos, retained);
    }

    inline bool beginPublish(const __FlashStringHelper* topic, size_t plength, uint8_t qos, bool retained) {
        // convert FlashStringHelper in PROGMEM-pointer
        return beginPublishImpl(true, reinterpret_cast<const char*>(topic), plength, qos, retained);
    }

    inline bool beginPublish_P(PGM_P topic, size_t plength, uint8_t qos, bool retained) {
        return beginPublishImpl(true, reinterpret_cast<const char*>(topic), plength, qos, retained);
    }

    bool endPublish();

    virtual size_t write(uint8_t data);

    virtual size_t write(const uint8_t* buf, size_t size);

    inline size_t write_P(PGM_P string) {
        return write_P(reinterpret_cast<const uint8_t*>(string), strlen_P(string));
    }

    size_t write_P(const uint8_t* buf, size_t size);

    inline bool subscribe(const char* topic) {
        return subscribeImpl(false, topic, MQTT_QOS0);
    }

    inline bool subscribe(const __FlashStringHelper* topic) {
        // convert FlashStringHelper in PROGMEM-pointer
        return subscribeImpl(true, reinterpret_cast<const char*>(topic), MQTT_QOS0);
    }

    inline bool subscribe_P(PGM_P topic) {
        return subscribeImpl(true, reinterpret_cast<const char*>(topic), MQTT_QOS0);
    }

    inline bool subscribe(const char* topic, uint8_t qos) {
        return subscribeImpl(false, topic, qos);
    }

    inline bool subscribe(const __FlashStringHelper* topic, uint8_t qos) {
        // convert FlashStringHelper in PROGMEM-pointer
        return subscribeImpl(true, reinterpret_cast<const char*>(topic), qos);
    }

    inline bool subscribe_P(PGM_P topic, uint8_t qos) {
        return subscribeImpl(true, reinterpret_cast<const char*>(topic), qos);
    }

    inline bool unsubscribe(const char* topic) {
        return unsubscribeImpl(false, topic);
    }

    inline bool unsubscribe(const __FlashStringHelper* topic) {
        // convert FlashStringHelper in PROGMEM-pointer
        return unsubscribeImpl(true, reinterpret_cast<const char*>(topic));
    }

    inline bool unsubscribe_P(PGM_P topic) {
        return unsubscribeImpl(true, reinterpret_cast<const char*>(topic));
    }

    bool loop();

    bool connected();

    int state();
};
 
#endif