ucloud-iot-rtthread-package/uiot/mqtt/include/mqtt_client.h

/*
* Copyright (C) 2012-2019 UCloud. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* or in the "license" file accompanying this file. This file is distributed
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
* express or implied. See the License for the specific language governing
* permissions and limitations under the License.
*/


#ifndef C_SDK_MQTT_CLIENT_H_
#define C_SDK_MQTT_CLIENT_H_

#ifdef __cplusplus
extern "C" {
#endif

#include <stddef.h>
#include <stdbool.h>
#include <stdint.h>

#include "uiot_export.h"
#include "uiot_import.h"
#include "uiot_internal.h"

#include "mqtt_client_net.h"

#include "utils_timer.h"
#include "utils_list.h"

/* 报文id最大值 */
#define MAX_PACKET_ID                                               (65535)

/* 成功订阅主题的最大个数 */
#define MAX_SUB_TOPICS                                              (10)

/* 在列表中最大的重新发布数量 */
#define MAX_REPUB_NUM                                               (20)

/* 重连最小等待时间 */
#define MIN_RECONNECT_WAIT_INTERVAL                                 (1000)

/* MQTT报文最小超时时间 */
#define MIN_COMMAND_TIMEOUT                                         (500)

/* MQTT报文最大超时时间 */
#define MAX_COMMAND_TIMEOUT                                         (5000)

/* 云端保留主题的最大长度 */
#define MAX_SIZE_OF_CLOUD_TOPIC                                     (128)

/* 动态注册相关的topic */
#define DYNAMIC_REGISTER_PUB_TEMPLATE          "/$system/%s/%s/password"

#define DYNAMIC_REGISTER_SUB_TEMPLATE          "/$system/%s/%s/password_reply"

/* 获取设备密钥的消息中的字段 */
#define PASSWORD_REPLY_RETCODE                 "RetCode"

#define PASSWORD_REPLY_REQUEST_ID              "RequestID"

#define PASSWORD_REPLY_PASSWORD                "Password"

/**
 * @brief MQTT Message Type
 */
typedef enum msgTypes {
    RESERVED    = 0,     // Reserved
    CONNECT     = 1,     // Client request to connect to Server
    CONNACK     = 2,     // Connect Acknowledgment
    PUBLISH     = 3,     // Publish message
    PUBACK      = 4,     // Publish Acknowledgment
    PUBREC      = 5,     // Publish Received
    PUBREL      = 6,     // Publish Release
    PUBCOMP     = 7,     // Publish Complete
    SUBSCRIBE   = 8,     // Client Subscribe request
    SUBACK      = 9,     // Subscribe Acknowledgment
    UNSUBSCRIBE = 10,    // Client Unsubscribe request
    UNSUBACK    = 11,    // Unsubscribe Acknowledgment
    PINGREQ     = 12,    // PING Request
    PINGRESP    = 13,    // PING Response
    DISCONNECT  = 14     // Client is Disconnecting
} MessageTypes;

typedef enum {
    DISCONNECTED = 0,
    CONNECTED    = 1
} ConnStatus;

typedef enum {
    STATIC_AUTH  = 1,        
    DYNAMIC_AUTH = 2,
} AuthStatus;

/** 
 * MQTT byte 1: fixed header
 * bits  |7654: Message Type  | 3:DUP flag |  21:QoS level | 0:RETAIN |
 */
#define MQTT_HEADER_TYPE_SHIFT          0x04
#define MQTT_HEADER_TYPE_MASK           0xF0
#define MQTT_HEADER_DUP_SHIFT           0x03
#define MQTT_HEADER_DUP_MASK            0x08
#define MQTT_HEADER_QOS_SHIFT           0x01
#define MQTT_HEADER_QOS_MASK            0x06
#define MQTT_HEADER_RETAIN_MASK         0x01


/**
 * @brief MQTT 遗嘱消息参数结构体定义
 *
 * 当客户端异常断开连接, 若客户端在连接时有设置遗嘱消息, 服务端会发布遗嘱消息。
 */
typedef struct {
    char        struct_id[4];      // The eyecatcher for this structure. must be MQTW
    uint8_t     struct_version;    // 结构体 0
    char        *topic_name;       // 遗嘱消息主题名
    char        *message;          // 遗嘱消息负载部分数据
    uint8_t     retained;          // 遗嘱消息retain标志位
    QoS         qos;               // 遗嘱消息qos标志位
} WillOptions;

/**
 * MQTT遗嘱消息结构体默认值定义
 */
#define DEFAULT_WILL_OPTIONS { {'M', 'Q', 'T', 'W'}, 0, NULL, NULL, 0, QOS0 }

/**
 * @brief MQTT 连接参数结构体定义
 *
 */
typedef struct {
    char                        *client_id;             // 客户端标识符, 请保持唯一
    char                        *username;              // 用户名
    char                        *password;                // 密码

    char                        struct_id[4];           // The eyecatcher for this structure.  must be MQTC.
    uint8_t                     struct_version;         // 结构体版本号, 必须为0
    uint8_t                     mqtt_version;           // MQTT版本协议号 4 = 3.1.1
    uint16_t                    keep_alive_interval;    // 心跳周期, 单位: s
    uint8_t                     clean_session;          // 清理会话标志位, 具体含义请参考MQTT协议说明文档3.1.2.4小结

    uint8_t                     auto_connect_enable;    // 是否开启自动重连
} MQTTConnectParams;

/**
 * MQTT连接参数结构体默认值定义
 */
#define DEFAULT_MQTT_CONNECT_PARAMS { NULL, NULL, NULL, {'M', 'Q', 'T', 'C'}, 0, 4, 240, 1 , 1,}

/**
 * @brief 订阅主题对应的消息处理结构体定义
 */
typedef struct SubTopicHandle {
    const char              *topic_filter;               // 订阅主题名, 可包含通配符
    OnMessageHandler        message_handler;             // 订阅主题消息回调:wq函数指针
    void                    *message_handler_data;       // 用户数据, 通过回调函数返回
    QoS                     qos;                         // 服务质量等级
} SubTopicHandle;

/**
 * @brief MQTT Client结构体定义
 */
typedef struct Client {
    uint8_t                  is_connected;                                  // 网络是否连接
    uint8_t                  was_manually_disconnected;                     // 是否手动断开连接
    uint8_t                  is_ping_outstanding;                           // 心跳包是否未完成, 即未收到服务器响应

    uint16_t                 next_packet_id;                                // MQTT报文标识符
    uint32_t                 command_timeout_ms;                            // MQTT消息超时时间, 单位:ms

    uint32_t                 current_reconnect_wait_interval;               // MQTT重连周期, 单位:ms
    uint32_t                 counter_network_disconnected;                  // 网络断开连接次数

    size_t                   write_buf_size;                                // 消息发送buffer长度
    size_t                   read_buf_size;                                 // 消息接收buffer长度
    unsigned char            write_buf[UIOT_MQTT_TX_BUF_LEN];               // MQTT消息发送buffer
    unsigned char            read_buf[UIOT_MQTT_RX_BUF_LEN];                // MQTT消息接收buffer

    void                     *lock_generic;                                 // client原子锁
    void                     *lock_write_buf;                               // 输出流的锁

    void                     *lock_list_pub;                                // 等待发布消息ack列表的锁
    void                     *lock_list_sub;                                // 等待订阅消息ack列表的锁

    List                     *list_pub_wait_ack;                            // 等待发布消息ack列表
    List                     *list_sub_wait_ack;                            // 等待订阅消息ack列表

    MQTTEventHandler         event_handler;                                 // 事件句柄

    MQTTConnectParams        options;                                       // 连接相关参数

    utils_network_t          network_stack;                                 // MQTT底层使用的网络参数

    Timer                    ping_timer;                                    // MQTT心跳包发送定时器
    Timer                    reconnect_delay_timer;                         // MQTT重连定时器, 判断是否已到重连时间

    SubTopicHandle           sub_handles[MAX_SUB_TOPICS];                   // 订阅主题对应的消息处理结构数组
} UIoT_Client;

/**
 * @brief MQTT协议版本
 */
typedef enum {
    MQTT_3_1_1 = 4
} MQTT_VERSION;


typedef enum MQTT_NODE_STATE {
    MQTT_NODE_STATE_NORMAL = 0,
    MQTT_NODE_STATE_INVALID,
} MQTTNodeState;

/* 记录已经发布的topic的信息 */
typedef struct REPUBLISH_INFO {
    Timer                   pub_start_time;     /* 发布的时间 */
    MQTTNodeState           node_state;         /* 节点状态 */
    uint16_t                msg_id;             /* 发布消息的packet id */
    uint32_t                len;                /* 消息长度 */
    unsigned char           *buf;               /* 消息内容 */
} UIoTPubInfo;

/* 记录已经订阅的topic的信息 */
typedef struct SUBSCRIBE_INFO {
    enum msgTypes           type;           /* 类型, (sub or unsub) */
    uint16_t                msg_id;         /* 订阅/取消订阅的 packet id */
    Timer                   sub_start_time; /* 订阅的开始时间 */
    MQTTNodeState           node_state;     /* 节点状态 */
    SubTopicHandle          handler;        /* handle of topic subscribed(unsubscribed) */
    uint16_t                len;            /* 消息长度 */
    unsigned char           *buf;           /* 消息内容 */
} UIoTSubInfo;

/**
 * @brief 对结构体Client进行初始化
 *
 * @param pClient MQTT客户端结构体
 * @param pParams MQTT客户端初始化参数
 * @return 返回SUCCESS, 表示成功
 */
int uiot_mqtt_init(UIoT_Client *pClient, MQTTInitParams *pParams);

/**
 * @brief 建立基于TLS的MQTT连接
 *
 * 注意: Client ID不能为NULL或空字符串
 *
 * @param pClient MQTT客户端结构体
 * @param pParams MQTT连接相关参数, 可参考MQTT协议说明
 * @return 返回SUCCESS, 表示成功
 */
int uiot_mqtt_connect(UIoT_Client *pClient, MQTTConnectParams *pParams);

/**
 * @brief 与服务器重新建立MQTT连接
 *
 * 1. 与服务器重新建立MQTT连接
 * 2. 连接成功后, 重新订阅之前的订阅过的主题
 *
 * @param pClient MQTT Client结构体
 *
 * @return 返回ERR_MQTT_RECONNECTED, 表示重连成功
 */
int uiot_mqtt_attempt_reconnect(UIoT_Client *pClient);

/**
 * @brief 断开MQTT连接
 *
 * @param pClient MQTT Client结构体
 * @return 返回SUCCESS, 表示成功
 */
int uiot_mqtt_disconnect(UIoT_Client *pClient);

/**
 * @brief 发布MQTT消息
 *
 * @param pClient MQTT客户端结构体
 * @param topicName 主题名
 * @param pParams 发布参数
 * @return < 0  :   表示失败
 *         >= 0 :   返回唯一的packet id 
 */
int uiot_mqtt_publish(UIoT_Client *pClient, char *topicName, PublishParams *pParams);

/**
 * @brief 订阅MQTT主题
 *
 * @param pClient MQTT客户端结构体
 * @param topicFilter 主题过滤器, 可参考MQTT协议说明 4.7
 * @param pParams 订阅参数
 * @return < 0  :   表示失败
 *         >= 0 :   返回唯一的packet id 
 */
int uiot_mqtt_subscribe(UIoT_Client *pClient, char *topicFilter, SubscribeParams *pParams);

/**
 * @brief 重新订阅断开连接之前已订阅的主题
 *
 * @param pClient MQTT客户端结构体
 * @return 返回SUCCESS, 表示成功
 */
int uiot_mqtt_resubscribe(UIoT_Client *pClient);

/**
 * @brief 取消订阅已订阅的MQTT主题
 *
 * @param pClient MQTT客户端结构体
 * @param topicFilter  主题过滤器, 可参考MQTT协议说明 4.7
 * @return < 0  :   表示失败
 *         >= 0 :   返回唯一的packet id   
 */
int uiot_mqtt_unsubscribe(UIoT_Client *pClient, char *topicFilter);

/**
 * @brief 在当前线程为底层MQTT客户端让出一定CPU执行时间
 *
 * 在这段时间内, MQTT客户端会用用处理消息接收, 以及发送PING报文, 监控网络状态
 *
 * @param pClient    MQTT Client结构体
 * @param timeout_ms Yield操作超时时间
 * @return 返回SUCCESS, 表示成功, 返回ERR_MQTT_ATTEMPTING_RECONNECT, 表示正在重连
 */
int uiot_mqtt_yield(UIoT_Client *pClient, uint32_t timeout_ms);

/**
 * @brief 客户端自动重连是否开启
 *
 * @param pClient MQTT客户端结构体
 * @return 返回true, 表示客户端已开启自动重连功能
 */
bool uiot_mqtt_is_autoreconnect_enabled(UIoT_Client *pClient);

/**
 * @brief 设置客户端是否开启自动重连
 *
 * @param pClient MQTT客户端结构体
 * @param value 是否开启该功能
 * @return 返回SUCCESS, 表示设置成功
 */
int uiot_mqtt_set_autoreconnect(UIoT_Client *pClient, bool value);

/**
 * @brief 获取网络断开的次数
 *
 * @param pClient MQTT Client结构体
 * @return 返回客户端MQTT网络断开的次数
 */
int uiot_mqtt_get_network_disconnected_count(UIoT_Client *pClient);

/**
 * @brief 重置连接断开次数
 *
 * @param pClient MQTT Client结构体
 * @return 返回SUCCESS, 表示设置成功
 */
int uiot_mqtt_reset_network_disconnected_count(UIoT_Client *pClient);

/**
 * @brief 获取报文标识符
 *
 * @param pClient
 * @return
 */
uint16_t get_next_packet_id(UIoT_Client *pClient);

/**
 *
 * @param header
 * @param message_type
 * @param qos
 * @param dup
 * @param retained
 * @return
 */
int mqtt_init_packet_header(unsigned char *header, MessageTypes message_type, QoS qos, uint8_t dup, uint8_t retained);

/**
 * @brief 接收服务端消息
 *
 * @param pClient
 * @param timer
 * @param packet_type
 * @param qos
 * @return
 */
int cycle_for_read(UIoT_Client *pClient, Timer *timer, uint8_t *packet_type, QoS qos);

/**
 * @brief 发送报文数据
 *
 * @param pClient       Client结构体
 * @param length        报文长度
 * @param timer         定时器
 * @return
 */
int send_mqtt_packet(UIoT_Client *pClient, size_t length, Timer *timer);

/**
 * @brief 等待指定类型的MQTT控制报文
 *
 * only used in single-threaded mode where one command at a time is in process
 *
 * @param pClient       MQTT Client结构体
 * @param packet_type   控制报文类型
 * @param timer         定时器
 * @return
 */
int wait_for_read(UIoT_Client *pClient, uint8_t packet_type, Timer *timer, QoS qos);

/**
 * @brief 设置MQTT当前连接状态
 *
 * @param pClient       Client结构体
 * @param connected     0: 连接断开状态  1: 已连接状态
 * @return
 */
void set_client_conn_state(UIoT_Client *pClient, uint8_t connected);

/**
 * @brief 获取MQTT当前连接状态
 *
 * @param pClient         Client结构体
 * @return                0: 连接断开状态  1: 已连接状态
 */
uint8_t get_client_conn_state(UIoT_Client *pClient);

/**
 * @brief 检查 Publish ACK 等待列表，若有成功接收或者超时，则将对应节点从列表中移除
 *
 * @param pClient MQTT客户端
 * @return 返回SUCCESS, 表示成功
 */
int uiot_mqtt_pub_info_proc(UIoT_Client *pClient);

/**
 * @brief 检查 Subscribe ACK 等待列表，若有成功接收或者超时，则将对应节点从列表中移除
 *
 * @param pClient MQTT客户端
 * @return 返回SUCCESS, 表示成功
 */
int uiot_mqtt_sub_info_proc(UIoT_Client *pClient);

int push_sub_info_to(UIoT_Client *c, int len, unsigned short msgId, MessageTypes type,
                                   SubTopicHandle *handler, ListNode **node);

int serialize_pub_ack_packet(unsigned char *buf, size_t buf_len, MessageTypes packet_type, uint8_t dup,
                             uint16_t packet_id,
                             uint32_t *serialized_len);

int serialize_packet_with_zero_payload(unsigned char *buf, size_t buf_len, MessageTypes packetType, uint32_t *serialized_len);

int deserialize_publish_packet(unsigned char *dup, QoS *qos, uint8_t *retained, uint16_t *packet_id, char **topicName,
                               uint16_t *topicNameLen, unsigned char **payload, size_t *payload_len, unsigned char *buf, size_t buf_len);

int deserialize_suback_packet(uint16_t *packet_id, uint32_t max_count, uint32_t *count,
                                     QoS *grantedQoSs, unsigned char *buf, size_t buf_len);

int deserialize_unsuback_packet(uint16_t *packet_id, unsigned char *buf, size_t buf_len);

int deserialize_ack_packet(uint8_t *packet_type, uint8_t *dup, uint16_t *packet_id, unsigned char *buf, size_t buf_len);

bool parse_mqtt_payload_retcode_type(char *pJsonDoc, uint32_t *pRetCode); 

bool parse_mqtt_state_request_id_type(char *pJsonDoc, char **pType);

bool parse_mqtt_state_password_type(char *pJsonDoc, char **pType);

#ifdef MQTT_CHECK_REPEAT_MSG

void reset_repeat_packet_id_buffer(void);

#endif
/**
 * @brief 根据剩余长度计算整个MQTT报文的长度
 *
 * @param rem_len    剩余长度
 * @return           整个MQTT报文的长度
 */
size_t get_mqtt_packet_len(size_t rem_len);

size_t mqtt_write_packet_rem_len(unsigned char *buf, uint32_t length);

int mqtt_read_packet_rem_len_form_buf(unsigned char *buf, uint32_t *value, uint32_t *readBytesLen);

uint16_t mqtt_read_uint16_t(unsigned char **pptr);

unsigned char mqtt_read_char(unsigned char **pptr);

void mqtt_write_char(unsigned char **pptr, unsigned char c);

void mqtt_write_uint_16(unsigned char **pptr, uint16_t anInt);

void mqtt_write_utf8_string(unsigned char **pptr, const char *string);

#ifdef __cplusplus
}
#endif

#endif //C_SDK_MQTT_CLIENT_H_