1.优化MQTT keep alive连接机制。

2.增加一个MQTT错误类型用于unsub不存在的主题。

Author: spikelin

src/mqtt/src/mqtt_client_common.c

@@ -1272,7 +1272,8 @@ static void _handle_pingresp_packet(Qcloud_IoT_Client *pClient) {
     IOT_FUNC_ENTRY;
 
     HAL_MutexLock(pClient->lock_generic);
-    pClient->is_ping_outstanding = 0;    
+    pClient->is_ping_outstanding = 0;
+    countdown(&pClient->ping_timer, pClient->options.keep_alive_interval);
     HAL_MutexUnlock(pClient->lock_generic);
 
     IOT_FUNC_EXIT;
@@ -1296,10 +1297,6 @@ int cycle_for_read(Qcloud_IoT_Client *pClient, Timer *timer, uint8_t *packet_typ
         IOT_FUNC_EXIT_RC(rc);
     }
 
-    HAL_MutexLock(pClient->lock_generic);    
-    countdown(&pClient->ping_timer, pClient->options.keep_alive_interval);
-    HAL_MutexUnlock(pClient->lock_generic);
-    
     switch (*packet_type) {
         case CONNACK:
             break;
@@ -1326,10 +1323,8 @@ int cycle_for_read(Qcloud_IoT_Client *pClient, Timer *timer, uint8_t *packet_typ
         }
         case PUBCOMP:
             break;
-        case PINGRESP: {
-            _handle_pingresp_packet(pClient);
+        case PINGRESP: 
             break;
-        }
         default: {
             /* Either unknown packet type or Failure occurred
              * Should not happen */
@@ -1339,6 +1334,23 @@ int cycle_for_read(Qcloud_IoT_Client *pClient, Timer *timer, uint8_t *packet_typ
         }
     }
 
+    /* Receiving below msgs are all considered as PING OK */
+    switch (*packet_type) {        
+        case PUBACK:
+        case SUBACK:
+        case UNSUBACK:        
+        case PINGRESP: {
+            _handle_pingresp_packet(pClient);
+            break;
+        }
+        case PUBLISH: {
+            HAL_MutexLock(pClient->lock_generic);    
+            pClient->is_ping_outstanding = 0;
+            HAL_MutexUnlock(pClient->lock_generic);
+            break;
+        }
+    }
+
     IOT_FUNC_EXIT_RC(rc);
 }
 

src/mqtt/src/mqtt_client_unsubscribe.c

@@ -126,7 +126,7 @@ int qcloud_iot_mqtt_unsubscribe(Qcloud_IoT_Client *pClient, char *topicFilter) {
 
     if (suber_exists == false) {
         Log_e("subscription does not exists: %s", topicFilter);
-        IOT_FUNC_EXIT_RC(QCLOUD_ERR_FAILURE);
+        IOT_FUNC_EXIT_RC(QCLOUD_ERR_MQTT_UNSUB_FAIL);
     }
 
     if (!get_client_conn_state(pClient)) {

src/mqtt/src/mqtt_client_yield.c

@@ -130,7 +130,10 @@ static int _handle_reconnect(Qcloud_IoT_Client *pClient) {
  * @param pClient
  * @return
  */
-static int _mqtt_keep_alive(Qcloud_IoT_Client *pClient) {
+static int _mqtt_keep_alive(Qcloud_IoT_Client *pClient)
+{
+#define MQTT_PING_TIMES_IN_KEEPALIVE_INTERVAL   2
+
     IOT_FUNC_ENTRY;
 
     int rc;
@@ -145,14 +148,14 @@ static int _mqtt_keep_alive(Qcloud_IoT_Client *pClient) {
         IOT_FUNC_EXIT_RC(QCLOUD_ERR_SUCCESS);
     }
 
-    if (pClient->is_ping_outstanding) {
+    if (pClient->is_ping_outstanding >= MQTT_PING_TIMES_IN_KEEPALIVE_INTERVAL) {
+        //Reaching here means we haven't received any MQTT packet for a long time (keep_alive_interval)
+        Log_e("Fail to recv MQTT msg. Something wrong with the connection.");
         rc = _handle_disconnect(pClient);
         IOT_FUNC_EXIT_RC(rc);
     }
 
-    /* there is no ping outstanding - send one */
-    InitTimer(&timer);
-    countdown_ms(&timer, pClient->command_timeout_ms);
+    /* there is no ping outstanding - send one */    
     HAL_MutexLock(pClient->lock_write_buf);
     rc = serialize_packet_with_zero_payload(pClient->write_buf, pClient->write_buf_size, PINGREQ, &serialized_len);
     if (QCLOUD_ERR_SUCCESS != rc) {
@@ -161,10 +164,15 @@ static int _mqtt_keep_alive(Qcloud_IoT_Client *pClient) {
     }
 
     /* send the ping packet */
-    rc = send_mqtt_packet(pClient, serialized_len, &timer);
+    int i = 0;
+    InitTimer(&timer);    
+    do {
+        countdown_ms(&timer, pClient->command_timeout_ms);
+        rc = send_mqtt_packet(pClient, serialized_len, &timer);
+    } while (QCLOUD_ERR_SUCCESS != rc && (i++ < 3));
+    
     if (QCLOUD_ERR_SUCCESS != rc) {
-        HAL_MutexUnlock(pClient->lock_write_buf);
-        //Reaching here means we haven't received any MQTT packet for a long time (keep_alive_interval)
+        HAL_MutexUnlock(pClient->lock_write_buf);        
         //If sending a PING fails, propably the connection is not OK and we decide to disconnect and begin reconnection attempts
         Log_e("Fail to send PING request. Something wrong with the connection.");
         rc = _handle_disconnect(pClient);
@@ -173,10 +181,11 @@ static int _mqtt_keep_alive(Qcloud_IoT_Client *pClient) {
     HAL_MutexUnlock(pClient->lock_write_buf);
 
     HAL_MutexLock(pClient->lock_generic);
-    pClient->is_ping_outstanding = 1;
+    pClient->is_ping_outstanding++;
     /* start a timer to wait for PINGRESP from server */
-    countdown(&pClient->ping_timer, pClient->options.keep_alive_interval/2);
+    countdown(&pClient->ping_timer, pClient->options.keep_alive_interval/MQTT_PING_TIMES_IN_KEEPALIVE_INTERVAL);
     HAL_MutexUnlock(pClient->lock_generic);
+    Log_d("PING request %u has been sent...", pClient->is_ping_outstanding);
 
     IOT_FUNC_EXIT_RC(QCLOUD_ERR_SUCCESS);
 }

src/sdk-impl/exports/qcloud_iot_export_error.h

@@ -63,6 +63,7 @@ typedef enum {
     QCLOUD_ERR_RX_MESSAGE_INVAL                              = -118,    // 表示收到的消息无效
     QCLOUD_ERR_BUF_TOO_SHORT                                 = -119,    // 表示消息接收缓冲区的长度小于消息的长度
     QCLOUD_ERR_MQTT_QOS_NOT_SUPPORT                          = -120,    // 表示该QOS级别不支持
+    QCLOUD_ERR_MQTT_UNSUB_FAIL                               = -121,    // 表示取消订阅主题失败,比如该主题不存在
 
     QCLOUD_ERR_JSON_PARSE                                    = -132,    // 表示JSON解析错误
     QCLOUD_ERR_JSON_BUFFER_TRUNCATED                         = -133,    // 表示JSON文档会被截断

src/sdk-impl/qcloud_iot_export.h

@@ -24,7 +24,7 @@ extern "C" {
 #define QCLOUD_IOT_DEVICE_SDK_APPID                                	"21010406"
 
 /* IoT C-SDK version info */
-#define QCLOUD_IOT_DEVICE_SDK_VERSION                               "2.3.2"
+#define QCLOUD_IOT_DEVICE_SDK_VERSION                               "2.3.3"
 
 /* MQTT心跳消息发送周期, 单位:ms */
 #define QCLOUD_IOT_MQTT_KEEP_ALIVE_INTERNAL                         (240 * 1000)