mqttclient源码的实现
API
mqttclient拥有非常简洁的api接口
int mqtt_keep_alive(mqtt_client_t* c);
int mqtt_init(mqtt_client_t* c, client_init_params_t* init);
int mqtt_release(mqtt_client_t* c);
int mqtt_connect(mqtt_client_t* c);
int mqtt_disconnect(mqtt_client_t* c);
int mqtt_subscribe(mqtt_client_t* c, const char* topic_filter, mqtt_qos_t qos, message_handler_t msg_handler);
int mqtt_unsubscribe(mqtt_client_t* c, const char* topic_filter);
int mqtt_publish(mqtt_client_t* c, const char* topic_filter, mqtt_message_t* msg);
int mqtt_yield(mqtt_client_t* c, int timeout_ms);
核心
mqtt_client_t 结构
typedef struct mqtt_client {
unsigned short packet_id;
unsigned char ping_outstanding;
unsigned char ack_handler_number;
unsigned char *read_buf;
unsigned char *write_buf;
unsigned int cmd_timeout;
unsigned int read_buf_size;
unsigned int write_buf_size;
unsigned int reconnect_try_duration;
void *reconnect_date;
reconnect_handler_t reconnect_handler;
client_state_t client_state;
platform_mutex_t write_lock;
platform_mutex_t global_lock;
list_t msg_handler_list;
list_t ack_handler_list;
network_t *network;
platform_thread_t *thread;
platform_timer_t reconnect_timer;
platform_timer_t last_sent;
platform_timer_t last_received;
connect_params_t *connect_params;
} mqtt_client_t;
该结构主要维护以下内容：
读写数据缓冲区
read_buf、write_buf命令超时时间
cmd_timeout（主要是读写阻塞时间、等待响应的时间、重连等待时间）维护
ack链表ack_handler_list，这是异步实现的核心，所有等待响应的报文都会被挂载到这个链表上维护消息处理列表
msg_handler_list，这是mqtt协议必须实现的内容，所有来自服务器的publish报文都会被处理（前提是订阅了对应的消息）维护一个网卡接口
network维护一个内部线程
thread，所有来自服务器的mqtt包都会在这里被处理！两个定时器，分别是掉线重连定时器与保活定时器
reconnect_timer、last_sent、last_received一些连接的参数
connect_params
mqttclient实现
以下是整个框架的实现方式，方便大家更容易理解mqttclient的代码与设计思想，让大家能够修改源码与使用，还可以提交pr或者issues，开源的世界期待各位大神的参与，感谢！
除此之外以下代码的记录机制与其超时处理机制是非常好的编程思想，大家有兴趣一定要看源代码！
初始化
int mqtt_init(mqtt_client_t* c, client_init_params_t* init)
主要是配置mqtt_client_t结构的相关信息，如果没有指定初始化参数，则系统会提供默认的参数。
但连接部分的参数则必须指定：
init_params.connect_params.network_params.addr = "[你的mqtt服务器IP地址或者是域名]";
init_params.connect_params.network_params.port = 1883; //端口号
init_params.connect_params.user_name = "jiejietop";
init_params.connect_params.password = "123456";
init_params.connect_params.client_id = "clientid";
mqtt_init(&client, &init_params);
连接服务器
int mqtt_connect(mqtt_client_t* c);
参数只有 mqtt_client_t 类型的指针，字符串类型的主题（支持通配符"#" "+"），主题的服务质量，以及收到报文的处理函数，如不指定则有默认处理函数。连接服务器则是使用非异步的方式设计，因为必须等待连接上服务器才能进行下一步操作。
过程如下:
调用底层的连接函数连接上服务器：
c->network->connect(c->network);
序列化
mqtt的CONNECT报文并且发送
MQTTSerialize_connect(c->write_buf, c->write_buf_size, &connect_data)
mqtt_send_packet(c, len, &connect_timer)
等待来自服务器的
CONNACK报文
mqtt_wait_packet(c, CONNACK, &connect_timer)
连接成功后创建一个内部线程
mqtt_yield_thread
platform_thread_init("mqtt_yield_thread", mqtt_yield_thread, c, MQTT_THREAD_STACK_SIZE, MQTT_THREAD_PRIO, MQTT_THREAD_TICK)
订阅报文
int mqtt_subscribe(mqtt_client_t* c, const char* topic_filter, mqtt_qos_t qos, message_handler_t handler)
订阅报文使用异步设计来实现的：
过程如下：
序列化订阅报文并且发送给服务器
MQTTSerialize_subscribe(c->write_buf, c->write_buf_size, 0, mqtt_get_next_packet_id(c), 1, &topic, (int*)&qos)
mqtt_send_packet(c, len, &timer)
创建对应的消息处理节点，这个消息节点在收到服务器的
SUBACK订阅应答报文后会挂载到消息处理列表msg_handler_list上
mqtt_msg_handler_create(topic_filter, qos, handler)
在发送了报文给服务器那就要等待服务器的响应了，记录这个等待
SUBACK
mqtt_ack_list_record(c, SUBACK, mqtt_get_next_packet_id(c), len, msg_handler)
取消订阅
与订阅报文的逻辑基本差不多的~
序列化订阅报文并且发送给服务器
MQTTSerialize_unsubscribe(c->write_buf, c->write_buf_size, 0, packet_id, 1, &topic)
mqtt_send_packet(c, len, &timer)
创建对应的消息处理节点，这个消息节点在收到服务器的
UNSUBACK取消订阅应答报文后将消息处理列表msg_handler_list上的已经订阅的主题消息节点销毁
mqtt_msg_handler_create((const char*)topic_filter, QOS0, NULL)
在发送了报文给服务器那就要等待服务器的响应了，先记录这个等待
UNSUBACK
mqtt_ack_list_record(c, UNSUBACK, packet_id, len, msg_handler)
发布报文
int mqtt_publish(mqtt_client_t* c, const char* topic_filter, mqtt_message_t* msg)
参数只有 mqtt_client_t 类型的指针，字符串类型的主题（支持通配符），要发布的消息（包括服务质量、消息主体）。
mqtt_message_t msg;
msg.qos = 2;
msg.payload = (void *) buf;
mqtt_publish(&client, "testtopic1", &msg);
核心思想都差不多，过程如下：
先序列化发布报文，然后发送到服务器
MQTTSerialize_publish(c->write_buf, c->write_buf_size, 0, msg->qos, msg->retained, msg->id,
topic, (unsigned char*)msg->payload, msg->payloadlen);
mqtt_send_packet(c, len, &timer)
对于QOS0的逻辑，不做任何处理，对于QOS1和QOS2的报文则需要记录下来，在没收到服务器应答的时候进行重发
if (QOS1 == msg->qos) {
rc = mqtt_ack_list_record(c, PUBACK, mqtt_get_next_packet_id(c), len, NULL);
} else if (QOS2 == msg->qos) {
rc = mqtt_ack_list_record(c, PUBREC, mqtt_get_next_packet_id(c), len, NULL);
}
还有非常重要的一点，重发报文的MQTT报文头部需要设置DUP标志位，这是MQTT协议的标准，因此，在重发的时候作者直接操作了报文的DUP标志位，因为修改DUP标志位的函数我没有从MQTT库中找到，所以我封装了一个函数，这与LwIP中的交叉存取思想是一个道理，它假设我知道MQTT报文的所有操作，所以我可以操作它，这样子可以提高很多效率：
mqtt_set_publish_dup(c,1); /* may resend this data, set the udp flag in advance */
内部线程
static void mqtt_yield_thread(void *arg)
主要是对mqtt_yield函数的返回值做处理，比如在disconnect的时候销毁这个线程。
核心的处理函数mqtt_yield
数据包的处理
mqtt_packet_handle
static int mqtt_packet_handle(mqtt_client_t* c, platform_timer_t* timer)
对不同的包使用不一样的处理：
switch (packet_type) {
case 0: /* timed out reading packet */
break;
case CONNACK:
break;
case PUBACK:
case PUBCOMP:
rc = mqtt_puback_and_pubcomp_packet_handle(c, timer);
break;
case SUBACK:
rc = mqtt_suback_packet_handle(c, timer);
break;
case UNSUBACK:
rc = mqtt_unsuback_packet_handle(c, timer);
break;
case PUBLISH:
rc = mqtt_publish_packet_handle(c, timer);
break;
case PUBREC:
case PUBREL:
rc = mqtt_pubrec_and_pubrel_packet_handle(c, timer);
break;
case PINGRESP:
c->ping_outstanding = 0;
break;
default:
goto exit;
}
并且做保活的处理：
mqtt_keep_alive(c)
当发生超时后，
if (platform_timer_is_expired(&c->last_sent) || platform_timer_is_expired(&c->last_received))
序列号一个心跳包并且发送给服务器
MQTTSerialize_pingreq(c->write_buf, c->write_buf_size);
mqtt_send_packet(c, len, &timer);
当再次发生超时后，表示与服务器的连接已断开，需要重连的操作，设置客户端状态为断开连接
mqtt_set_client_state(c, CLIENT_STATE_DISCONNECTED);
ack链表的扫描，当收到服务器的报文时，对ack列表进行扫描操作
mqtt_ack_list_scan(c);
当超时后就销毁ack链表节点：
mqtt_ack_handler_destroy(ack_handler);
当然下面这几种报文则需要重发操作：（PUBACK 、PUBREC、 PUBREL 、PUBCOMP，保证QOS1 QOS2的服务质量）
if ((ack_handler->type == PUBACK) || (ack_handler->type == PUBREC) || (ack_handler->type == PUBREL) || (ack_handler->type == PUBCOMP))
mqtt_ack_handler_resend(c, ack_handler);
保持活性的时间过去了，可能掉线了，需要重连操作
mqtt_try_reconnect(c);
重连成功后尝试重新订阅报文，保证恢复原始状态~
mqtt_try_resubscribe(c)
发布应答与发布完成报文的处理
static int mqtt_puback_and_pubcomp_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
反序列化报文
MQTTDeserialize_ack(&packet_type, &dup, &packet_id, c->read_buf, c->read_buf_size)
取消对应的ack记录
mqtt_ack_list_unrecord(c, packet_type, packet_id, NULL);
订阅应答报文的处理
static int mqtt_suback_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
反序列化报文
MQTTDeserialize_suback(&packet_id, 1, &count, (int*)&granted_qos, c->read_buf, c->read_buf_size)
取消对应的ack记录
mqtt_ack_list_unrecord(c, packet_type, packet_id, NULL);
安装对应的订阅消息处理函数，如果是已存在的则不会安装
mqtt_msg_handlers_install(c, msg_handler);
取消订阅应答报文的处理
static int mqtt_unsuback_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
反序列化报文
MQTTDeserialize_unsuback(&packet_id, c->read_buf, c->read_buf_size)
取消对应的ack记录
mqtt_ack_list_unrecord(c, UNSUBACK, packet_id, &msg_handler)
销毁对应的订阅消息处理函数
mqtt_msg_handler_destory(msg_handler);
来自服务器的发布报文的处理
static int mqtt_publish_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
反序列化报文
MQTTDeserialize_publish(&msg.dup, &qos, &msg.retained, &msg.id, &topic_name,
(unsigned char**)&msg.payload, (int*)&msg.payloadlen, c->read_buf, c->read_buf_size)
对于QOS0、QOS1的报文，直接去处理消息
mqtt_deliver_message(c, &topic_name, &msg);
对于QOS1的报文，还需要发送一个
PUBACK应答报文给服务器
MQTTSerialize_ack(c->write_buf, c->write_buf_size, PUBACK, 0, msg.id);
而对于QOS2的报文则需要发送
PUBREC报文给服务器，除此之外还需要记录PUBREL到ack链表上，等待服务器的发布释放报文，最后再去处理这个消息
MQTTSerialize_ack(c->write_buf, c->write_buf_size, PUBREC, 0, msg.id);
mqtt_ack_list_record(c, PUBREL, msg.id + 1, len, NULL)
mqtt_deliver_message(c, &topic_name, &msg);
说明：一旦注册到ack列表上的报文，当具有重复的报文是不会重新被注册的，它会通过
mqtt_ack_list_node_is_exist函数判断这个节点是否存在，主要是依赖等待响应的消息类型与msgid。
发布收到与发布释放`报文的处理
static int mqtt_pubrec_and_pubrel_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
反序列化报文
MQTTDeserialize_ack(&packet_type, &dup, &packet_id, c->read_buf, c->read_buf_size)
产生一个对应的应答报文
mqtt_publish_ack_packet(c, packet_id, packet_type);
取消对应的ack记录
mqtt_ack_list_unrecord(c, UNSUBACK, packet_id, &msg_handler)