MQTT.h

/*
MQTT.h - MQTT packet classes
Copyright (C) 2015 Ian Tester

This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library 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
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/

#pragma once

#include <stdint.h>
#ifdef ESP8266
#include <pgmspace.h>
#include <functional>
#endif
#include <Client.h>

// MQTT_KEEPALIVE : keepAlive interval in Seconds
#define MQTT_KEEPALIVE 15

class PubSubClient;

namespace MQTT {

  enum message_type {
    None,
    CONNECT,        // Client request to connect to Server
    CONNACK,        // Connect Acknowledgment
    PUBLISH,        // Publish message
    PUBACK,     // Publish Acknowledgment
    PUBREC,     // Publish Received (assured delivery part 1)
    PUBREL,     // Publish Release (assured delivery part 2)
    PUBCOMP,        // Publish Complete (assured delivery part 3)
    SUBSCRIBE,      // Client Subscribe request
    SUBACK,     // Subscribe Acknowledgment
    UNSUBSCRIBE,    // Client Unsubscribe request
    UNSUBACK,       // Unsubscribe Acknowledgment
    PINGREQ,        // PING Request
    PINGRESP,       // PING Response
    DISCONNECT,     // Client is Disconnecting
    Reserved,       // Reserved
  };

#ifdef _GLIBCXX_FUNCTIONAL
  typedef std::function<bool(Client&)> payload_callback_t;
#else
  typedef bool(*payload_callback_t)(Client&);
#endif

  class Message {
  protected:
    message_type _type;
    uint8_t _flags;
    uint16_t _packet_id;    
    bool _need_packet_id;
    Client* _stream_client;
    payload_callback_t _payload_callback;

    Message(message_type t, uint8_t f = 0) :
      _type(t), _flags(f),
      _packet_id(0), _need_packet_id(false),
      _stream_client(NULL)
    {}

    virtual ~Message() {}


    uint8_t fixed_header_length(uint32_t rlength) const;


    void write_fixed_header(uint8_t *buf, uint32_t& bufpos, uint32_t rlength) const;

    bool need_packet_id(void) const { return _need_packet_id; }

    void set_packet_id(uint16_t pid) { _packet_id = pid; }

    uint16_t packet_id(void) const { return _packet_id; }


    void write_packet_id(uint8_t *buf, uint32_t& bufpos) const;

    virtual uint32_t variable_header_length(void) const { return 0; }


    virtual void write_variable_header(uint8_t *buf, uint32_t& bufpos) const { }

    virtual uint32_t payload_length(void) const { return 0; }


    virtual void write_payload(uint8_t *buf, uint32_t& bufpos) const { }

    virtual message_type response_type(void) const { return None; }

    friend PubSubClient;    // Just to allow it to call response_type()

  public:
    bool send(Client& client);

    message_type type(void) const { return _type; }

    bool has_stream(void) const { return _stream_client != NULL; }

  };


  Message* readPacket(Client& client);


  class Connect : public Message {
  private:
    bool _clean_session;
    uint8_t _will_qos;
    bool _will_retain;

    String _clientid;
    String _will_topic;
    String _will_message;
    String _username, _password;

    uint16_t _keepalive;

    uint32_t variable_header_length(void) const;
    void write_variable_header(uint8_t *buf, uint32_t& bufpos) const;
    uint32_t payload_length(void) const;
    void write_payload(uint8_t *buf, uint32_t& bufpos) const;

    message_type response_type(void) const { return CONNACK; }

  public:
    Connect(String cid);

    Connect& set_clean_session(bool cs = true)  { _clean_session = cs; return *this; }
    Connect& unset_clean_session(void)      { _clean_session = false; return *this; }

    Connect& set_will(String willTopic, String willMessage, uint8_t willQos = 0, bool willRetain = false) {
      _will_topic = willTopic; _will_message = willMessage; _will_qos = willQos; _will_retain = willRetain;
      return *this;
    }
    Connect& unset_will(void)           { _will_topic = ""; return *this; }

    Connect& set_auth(String u, String p)   { _username = u; _password = p; return *this; }
    Connect& unset_auth(void)           { _username = ""; _password = ""; return *this; }

    uint16_t keepalive(void) const  { return _keepalive; }
    Connect& set_keepalive(uint16_t k)  { _keepalive = k; return *this; }

  };


  class ConnectAck : public Message {
  private:
    bool _session_present;
    uint8_t _rc;

    ConnectAck(uint8_t* data, uint32_t length);

    friend Message* readPacket(Client& client);
  };


  class Publish : public Message {
  private:
    String _topic;
    uint8_t *_payload;
    uint32_t _payload_len;
    bool _payload_mine;

    uint32_t variable_header_length(void) const;
    void write_variable_header(uint8_t *buf, uint32_t& bufpos) const;
    uint32_t payload_length(void) const;
    void write_payload(uint8_t *buf, uint32_t& bufpos) const;

    message_type response_type(void) const;

    Publish(String topic, uint8_t* payload, uint32_t length, bool mine) :
      Message(PUBLISH),
      _topic(topic),
      _payload(payload), _payload_len(length),
      _payload_mine(mine)
    {}

    Publish(uint8_t flags, uint8_t* data, uint32_t length);

    Publish(uint8_t flags, Client& client, uint32_t remaining_length);

    friend Message* readPacket(Client& client);

  public:

    Publish(String topic, String payload);


    Publish(String topic, uint8_t* payload, uint32_t length) :
      Publish(topic, payload, length, false)
    {}


    Publish(String topic, payload_callback_t pcb, uint32_t length);

    Publish(String topic, const __FlashStringHelper* payload);

    friend Publish Publish_P(String topic, PGM_P payload, uint32_t length);

    ~Publish();

    bool retain(void) const     { return _flags & 0x01; }
    Publish& set_retain(bool r = true)  { _flags = (_flags & ~0x01) | r; return *this; }
    Publish& unset_retain(void)     { _flags = _flags & ~0x01; return *this; }

    uint8_t qos(void) const     { return (_flags >> 1) & 0x03; }
    Publish& set_qos(uint8_t q);
    Publish& unset_qos(void)        { _flags &= ~0x06; _need_packet_id = false; return *this; }

    bool dup(void) const        { return (_flags >> 3) & 0x01; }
    Publish& set_dup(bool d = true) { _flags = (_flags & ~0x08) | (d ? 0x08 : 0); return *this; }
    Publish& unset_dup(void)        { _flags = _flags & ~0x08; return *this; }

    String topic(void) const { return _topic; }

    String payload_string(void) const;

    uint8_t* payload(void) const { return _payload; }
    uint32_t payload_len(void) const { return _payload_len; }

    Client* payload_stream(void) const { return _stream_client; }
  };

  Publish Publish_P(String topic, PGM_P payload, uint32_t length);


  class PublishAck : public Message {
  private:
    PublishAck(uint8_t* data, uint32_t length);

    friend Message* readPacket(Client& client);

  public:
    PublishAck(uint16_t pid);

  };


  class PublishRec : public Message {
  private:
    uint32_t variable_header_length(void) const;
    void write_variable_header(uint8_t *buf, uint32_t& bufpos) const;

    message_type response_type(void) const { return PUBREL; }

    PublishRec(uint8_t* data, uint32_t length);

    friend Message* readPacket(Client& client);

  public:
    PublishRec(uint16_t pid);

  };


  class PublishRel : public Message {
  private:
    uint32_t variable_header_length(void) const;
    void write_variable_header(uint8_t *buf, uint32_t& bufpos) const;

    message_type response_type(void) const { return PUBCOMP; }

    PublishRel(uint8_t* data, uint32_t length);

    friend Message* readPacket(Client& client);

  public:
    PublishRel(uint16_t pid);

  };


  class PublishComp : public Message {
  private:
    uint32_t variable_header_length(void) const;
    void write_variable_header(uint8_t *buf, uint32_t& bufpos) const;

    PublishComp(uint8_t* data, uint32_t length);

    friend Message* readPacket(Client& client);

  public:
    PublishComp(uint16_t pid);

  };


  class Subscribe : public Message {
  private:
    uint8_t *_buffer;
    uint32_t _buflen;

    uint32_t variable_header_length(void) const;
    void write_variable_header(uint8_t *buf, uint32_t& bufpos) const;
    uint32_t payload_length(void) const;
    void write_payload(uint8_t *buf, uint32_t& bufpos) const;

    message_type response_type(void) const { return SUBACK; }

  public:
    Subscribe();

    Subscribe(String topic, uint8_t qos = 0);

    ~Subscribe();

    Subscribe& add_topic(String topic, uint8_t qos = 0);

  };


  class SubscribeAck : public Message {
  private:
    uint8_t *_rcs;
    uint32_t _num_rcs;

    SubscribeAck(uint8_t* data, uint32_t length);

    SubscribeAck(Client& client, uint32_t remaining_length);

    friend Message* readPacket(Client& client);

  public:
    ~SubscribeAck();

    uint32_t num_rcs(void) const { return _num_rcs; }

    uint8_t rc(uint8_t i) const { return _rcs[i]; }

    uint8_t next_rc(void) const;

  };


  class Unsubscribe : public Message {
  private:
    uint8_t *_buffer;
    uint32_t _buflen;

    uint32_t variable_header_length(void) const;
    void write_variable_header(uint8_t *buf, uint32_t& bufpos) const;
    uint32_t payload_length(void) const;
    void write_payload(uint8_t *buf, uint32_t& bufpos) const;

    message_type response_type(void) const { return UNSUBACK; }

  public:
    Unsubscribe();

    Unsubscribe(String topic);

    ~Unsubscribe();

    Unsubscribe& add_topic(String topic);

  };


  class UnsubscribeAck : public Message {
  private:
    UnsubscribeAck(uint8_t* data, uint32_t length);

    friend Message* readPacket(Client& client);

  };


  class Ping : public Message {
  private:
    message_type response_type(void) const { return PINGRESP; }

  public:
    Ping() :
      Message(PINGREQ)
    {}

  };


  class PingResp : public Message {
  public:
    PingResp() :
      Message(PINGRESP)
    {}

  };


  class Disconnect : public Message {
  public:
    Disconnect() :
      Message(DISCONNECT)
    {}

  };

}