[fix] 修正客户端keepalive期间发送ping命令超时问题

api/mqtt_api.c

@@ -92,7 +92,7 @@ MQTTStatus_t mqttConnect(NetworkContext_t *networkContext)
         return status;
     }
 
-    rt_kprintf("MQTT broker connected\n");
+    rt_kprintf("[%d] MQTT broker connected\n", getCurrentTime());
     return MQTTSuccess;
 }
 
@@ -128,21 +128,6 @@ MQTTStatus_t mqttPublish(MQTTPublishInfo_t *publishInfo)
     return MQTTSuccess;
 }
 
-static bool isSocketReadable(int socket, int timeout_ms)
-{
-    fd_set readfds;
-    struct timeval timeout;
-
-    FD_ZERO(&readfds);
-    FD_SET(socket, &readfds);
-
-    timeout.tv_sec = timeout_ms / 1000;
-    timeout.tv_usec = (timeout_ms % 1000) * 1000;
-
-    int result = select(socket + 1, &readfds, NULL, NULL, &timeout);
-    return (result > 0 && FD_ISSET(socket, &readfds));
-}
-
 const char *mqttStatus(MQTTStatus_t status)
 {
     const char *const statusStrings[] = {
@@ -252,15 +237,14 @@ void mqttClientTask(void *parameter)
 
         while (1)
         {
-            if (isSocketReadable(networkContext.socket, 100))
+            status = MQTT_ProcessLoop(&mqttContext);
+            if (status != MQTTSuccess && status != MQTTNeedMoreBytes)
             {
-                status = MQTT_ProcessLoop(&mqttContext);
-                if (status != MQTTSuccess)
-                {
-                    MQTT_PRINT("MQTT_ProcessLoop failed: %d (%s)\n", status, mqttStatus(status));
-                    break;
-                }
+                MQTT_PRINT("MQTT_ProcessLoop failed: %d (%s)\n", status, mqttStatus(status));
+                status = MQTT_Disconnect(&mqttContext);
+                break;
             }
+
             rt_thread_mdelay(MQTT_LOOP_CNT);
         }
 

api/mqtt_api.h

@@ -8,6 +8,7 @@
  * 2025-06-03     RV           the first version
  */
 #ifndef APPLICATIONS_FIREMQTT_PORT_MQTT_USR_API_H_
+
 #define APPLICATIONS_FIREMQTT_PORT_MQTT_USR_API_H_
 #include <rtthread.h>
 #include <core_mqtt.h>
@@ -21,7 +22,7 @@
 #include <unistd.h>      // 添加close等系统调用定义
 #include "port.h"
 #include <rtdbg.h>
-#include "config.h"
+#include <core_mqtt_config.h>
 
 #ifdef RT_USING_ULOG
 #define MQTT_PRINT(fmt, ...) LOG_D(fmt, ##__VA_ARGS__)

core/include/core_mqtt_config_defaults.h

@@ -46,7 +46,7 @@
 /* MQTT_DO_NOT_USE_CUSTOM_CONFIG allows building the MQTT library
  * without a custom config. If a custom config is provided, the
  * MQTT_DO_NOT_USE_CUSTOM_CONFIG macro should not be defined. */
-#define MQTT_DO_NOT_USE_CUSTOM_CONFIG
+
 #ifndef MQTT_DO_NOT_USE_CUSTOM_CONFIG
 /* Include custom config file before other headers. */
     #include "core_mqtt_config.h"

config.h → core_mqtt_config.h

@@ -10,16 +10,18 @@
 #ifndef APPLICATIONS_FIREMQTT_PORT_CONFIG_H_
 #define APPLICATIONS_FIREMQTT_PORT_CONFIG_H_
 #include "mqtt_api.h"
-#define MQTT_BROKER_ADDRESS  "broker.emqx.io" // MQTT 代理地址
-#define MQTT_BROKER_PORT     1883                 // MQTT 代理端口
+#define MQTT_BROKER_ADDRESS  "broker.emqx.io"       // MQTT 代理地址
+#define MQTT_BROKER_PORT     1883                   // MQTT 代理端口
 #define MQTT_CLIENT_ID       "rtthread_mqtt_client" // 客户端 ID
-#define MQTT_TOPIC_SUB       "rtthread/test/sub"   // 订阅主题
-#define MQTT_TOPIC_PUB       "rtthread/test/pub"   // 发布主题
+#define MQTT_TOPIC_SUB       "rtthread/test/sub"    // 订阅主题
+#define MQTT_TOPIC_PUB       "rtthread/test/pub"    // 发布主题
 #define MQTT_KEEP_ALIVE      60
 #define MQTT_LOOP_CNT        60
+#define MQTT_RECV_POLLING_TIMEOUT_MS    ( 0U )
+#define MQTT_PINGRESP_TIMEOUT_MS    ( 10000U )
 
-#define MAX_RETRY_ATTEMPTS   5                     // 最大重试次数
-#define INITIAL_BACKOFF_MS   1000                  // 初始重连退避时间（毫秒）
+#define MAX_RETRY_ATTEMPTS   5                      // 最大重试次数
+#define INITIAL_BACKOFF_MS   1000                   // 初始重连退避时间（毫秒）
 #define MAX_BACKOFF_MS       60000
 
 #define MQTT_USERCALLBACK mqttEventCallback