Merge branch 'mqttdev' into add/iotcloud

examples/mqtt/mqtt_connect.py

@@ -3,7 +3,7 @@ import mqtt
 client = mqtt.MQTT('broker.emqx.io',port=1883,clinetID='clientid',username='name_',password='passwd_')
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
 
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)

examples/mqtt/mqtt_publish.py

@@ -7,15 +7,15 @@ client.setPort(1883)
 client.setClientID('123456dddecetdc')
 client.setUsername('test1')
 client.setPassword('aabbccdd')
-client.setVersion('3')
+client.setVersion('3.1')
-client.setKeepAlive('10')
+client.setKeepAlive(10)
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
 
-client.publish(0,'topic_pikapy', 'hello pikascript qos=0')
+client.publish('topic_pikapy', 'hello pikascript qos=0', 0)
-client.publish(1,'topic_pikapy', 'hello pikascript qos=1')
+client.publish('topic_pikapy', 'hello pikascript qos=1', 1)
-client.publish(2,'topic_pikapy', 'hello pikascript qos=2')
+client.publish('topic_pikapy', 'hello pikascript qos=2', 2)
 
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)

examples/mqtt/mqtt_set_para.py

@@ -35,11 +35,11 @@ client.setPort(1883)
 client.setClientID('123456dddecetdc')
 client.setUsername('j6npr4w/mqtt-client-dev')
 client.setPassword('lcUhUs5VYLMSbrnB')
-client.setVersion('3')
+client.setVersion('3.1.1')
-client.setKeepAlive('10')
+client.setKeepAlive(10)
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
 
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)

examples/mqtt/mqtt_subscribe.py

@@ -4,36 +4,54 @@ import PikaStdDevice
 client = mqtt.MQTT('broker.emqx.io', port=1883, clinetID='clientid', username='name_', password='passwd_')
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
+
 
 def callback0(signal):
-    print("py cb: %s:%s" % (client.recv_topic, client.recv_msg))
+    recv_msg = client.getMsg(signal)
+    recv_topic = client.getTopic(signal)
+    recv_qos = client.getQos(signal)
+    print("py0 cb: %s-qos:%d-->>%s" % (recv_topic, recv_qos, recv_msg))
 
 
-ret = client.subscribe('topic_pikapy_qos0', 0, callback0)
+def callback1(signal):
-print("ret:%d" % ret)
+    recv_msg = client.getMsg(signal)
-ret = client.subscribe('topic_pikapy_qos1', 1,0)
+    recv_topic = client.getTopic(signal)
-print("ret:%d" % ret)
+    recv_qos = client.getQos(signal)
-ret = client.subscribe('topic_pikapy_qos2', 2,0)
+    print("py1 cb: %s-qos:%d-->>%s" % (recv_topic, recv_qos, recv_msg))
-print("ret:%d" % ret)
 
 
+def callback2(signal):
+    recv_msg = client.getMsg(signal)
+    recv_topic = client.getTopic(signal)
+    recv_qos = client.getQos(signal)
+    print("py2 cb: %s-qos:%d-->>%s" % (recv_topic, recv_qos, recv_msg))
+
+
+ret = client.subscribe('topic_pikapy_qos0', callback0, 0)
+print("subscribe ret:%d" % ret)
+ret = client.subscribe('topic_pikapy_qos1', callback1, 1)
+print("subscribe ret:%d" % ret)
+ret = client.subscribe('topic_pikapy_qos2', callback2, 2)
+print("subscribe ret:%d" % ret)
+
 # sleep wait for recv data
 T = PikaStdDevice.Time()
 T.sleep_s(5)
 
-out = client.listSubscribrTopic()
+out = client.listSubscribeTopic()
-print('out',out)
+print('listSubscribeTopic out', out)
 
 # client.unsubscribe('topic_pikapy_qos0');
 # client.unsubscribe('topic_pikapy_qos1');
 # client.unsubscribe('topic_pikapy_qos2');
 # T.sleep_s(5)
-# client.listSubscribrTopic()
+# out2 = client.listSubscribeTopic()
+# print('listSubscribeTopic out2',out2)
 
 
 # ret = client.setWill(1,'topic_will',1,'lost mqtt connect')
 T.sleep_s(10)
 # exit()
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)

package/mqtt/_mqtt.pyi

@@ -26,11 +26,11 @@ class _MQTT:
         pass
         """Set the Ca of the MQTTClient."""
 
-    def setKeepAlive(self, time: str) -> int:
+    def setKeepAlive(self, time: int) -> int:
         pass
         """Set the KeepAlive of the MQTTClient."""
 
-    def setWill(self, qos: int, topic: str, retain: int, payload: str) -> int:
+    def setWill(self, topic: str, payload: str, qos: int, retain: int) -> int:
         pass
         """Set the Will of the MQTTClient."""
 
@@ -42,7 +42,7 @@ class _MQTT:
         pass
         """disconnect to the mqtt-server."""
 
-    def subscribe(self, topic: str, qos: int, cb: any) -> int:
+    def subscribe(self, topic: str, cb: any, qos: int) -> int:
         pass
         """subscribe to the mqtt-server."""
 
@@ -50,11 +50,11 @@ class _MQTT:
         pass
         """unsubscribe to the mqtt-server."""
 
-    def listSubscribrTopic(self) -> list:
+    def listSubscribeTopic(self) -> list:
         pass
-        """listSubscribrTopic """
+        """listSubscribeTopic """
 
-    def publish(self, qos:int, topic: str, payload: str) -> int:
+    def publish(self,topic: str, payload: str, qos:int) -> int:
         pass
         """publish to the mqtt-server."""
 
@@ -64,6 +64,22 @@ class _MQTT:
     def setHost(self, host_url: str) -> int:
         """Set the host_url of the MQTTClient."""
 
+    def getMsg(self,signal:int) -> str:
+        pass
+        """callback fun get msg"""
+
+    def getTopic(self,signal:int) -> str:
+        pass
+        """callback fun get topic"""
+        
+    def getQos(self,signal:int) -> int:
+        pass
+        """callback fun get qos"""
+        
+    def setDisconnectHandler(self,cb: any) -> int:
+        pass
+        """set disconnect callback fun."""
+
 
 def __del__():
     pass

package/mqtt/_mqtt__MQTT.c

@@ -1,6 +1,8 @@
 #include "_mqtt__MQTT.h"
+#include "PikaStdData_List.h"
+#include "TinyObj.h"
 #include "mqttclient.h"
-
+#include "PikaObj.h"
 PikaEventListener* g_mqtt_event_listener = NULL;
 
 void Subscribe_Handler(void* client, message_data_t* msg);
@@ -31,8 +33,7 @@ void _mqtt__MQTT___init__(PikaObj* self,
     if (strlen(ip) > 0) {
         obj_setStr(self, "host_str", ip);
         mqtt_set_host(_client, obj_getStr(self, "host_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input ip none\r\n");
     }
 
@@ -44,47 +45,41 @@ void _mqtt__MQTT___init__(PikaObj* self,
     if (strlen(clinetID) > 0) {
         obj_setStr(self, "id_str", clinetID);
         mqtt_set_client_id(_client, obj_getStr(self, "id_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input clinetID none\r\n");
     }
 
     if (strlen(username) > 0) {
         obj_setStr(self, "username_str", username);
         mqtt_set_user_name(_client, obj_getStr(self, "username_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input username none\r\n");
     }
 
     if (strlen(password) > 0) {
         obj_setStr(self, "password_str", password);
         mqtt_set_password(_client, obj_getStr(self, "password_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input password none\r\n");
     }
 
     tmp = atoi(version);
     if (tmp > 0) {
         mqtt_set_version(_client, tmp);
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input version none\r\n");
     }
 
     if (strlen(ca) > 0) {
         obj_setStr(self, "ca_str", ca);
         mqtt_set_ca(_client, obj_getStr(self, "ca_str"));
-    }
+    } else {
-    else {
+        // __platform_printf("mqtt_init input ca none\r\n");
-        __platform_printf("mqtt_init input ca none\r\n");
     }
 
     if (keepalive > 0) {
         mqtt_set_keep_alive_interval(_client, keepalive);
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input keepalive none\r\n");
     }
 
@@ -92,7 +87,7 @@ void _mqtt__MQTT___init__(PikaObj* self,
 
     obj_setPtr(self, "_client",
                _client);  // 这里要再保存一次mqtt结构体的内容到python环境
-    __platform_printf("Mqtt_Lib buildtime:%s-%s\r\n", __DATE__, __TIME__);
+    // __platform_printf("Mqtt_Lib buildtime:%s-%s\r\n", __DATE__, __TIME__);
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -111,7 +106,7 @@ void _mqtt__MQTT___del__(PikaObj* self) {
     if (_connected) {
         _mqtt__MQTT_disconnect(self);
     }
-    MQTT_LOG_E("%s:%d %s() >_<", __FILE__, __LINE__, __FUNCTION__);
+    // MQTT_LOG_E("%s:%d %s() >_<", __FILE__, __LINE__, __FUNCTION__);
     mqtt_release_free(_client);
 }
 
@@ -130,8 +125,8 @@ int _mqtt__MQTT_connect(PikaObj* self) {
     if (ret != 0)
         __platform_printf("mqtt connect ERROR! :%d\r\n", ret);
 
-    if (ret == 0)
+    // if (ret == 0)
-        __platform_printf("mqtt connect OK\r\n");
+    //     __platform_printf("mqtt connect OK\r\n");
     return ret;
 }
 
@@ -150,31 +145,55 @@ int _mqtt__MQTT_disconnect(PikaObj* self) {
     if (ret != 0)
         __platform_printf("mqtt disconnect ERROR! :%d\r\n", ret);
 
-    if (ret == 0)
+    // if (ret == 0)
-        __platform_printf("mqtt disconnect OK\r\n");
+    //     __platform_printf("mqtt disconnect OK\r\n");
     return ret;
 }
 
 ////////////////////////////////////////////////////////////////////
-// 函 数 名：_mqtt__MQTT_listSubscribrTopic
+// 函 数 名：_mqtt__MQTT_listSubscribeTopic
 // 功能说明：罗列出当前订阅的主题
 // 输入参数：无
 // 返 回 值：对象指针
 ///////////////////////////////////////////////////////////////////
-PikaObj* _mqtt__MQTT_listSubscribrTopic(PikaObj* self) {
+PikaObj* _mqtt__MQTT_listSubscribeTopic(PikaObj* self) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
-    int ret;
+    int i = 0;
-    PikaObj* pt_out = NULL;
+    mqtt_list_t *curr, *next;
+    message_handlers_t* msg_handler;
+    PikaObj* list = NULL;
 
-    ret = mqtt_list_subscribe_topic(_client);
+    if (NULL == _client) {
-    if (ret == 0)
+        return NULL;
-        __platform_printf("MQTT_listSubscribrTopic OK\r\n");
-    else {
-        pt_out = NULL;
-        __platform_printf("MQTT_listSubscribrTopic ERROR\r\n");
     }
 
-    return pt_out;
+    if (mqtt_list_is_empty(&_client->mqtt_msg_handler_list)) {
+        MQTT_LOG_I("%s:%d %s()... there are no subscribed topics...", __FILE__,
+                   __LINE__, __FUNCTION__);
+        return NULL;
+    }
+
+    /* 创建 list 对象 */
+    list = newNormalObj(New_PikaStdData_List);
+    /* 初始化 list */
+    PikaStdData_List___init__(list);
+
+    LIST_FOR_EACH_SAFE(curr, next, &_client->mqtt_msg_handler_list) {
+        msg_handler = LIST_ENTRY(curr, message_handlers_t, list);
+        /* determine whether a node already exists by mqtt topic, but wildcards
+         * are not supported */
+        if (NULL != msg_handler->topic_filter) {
+            MQTT_LOG_I("%s:%d %s()...[%d] subscribe topic: %s", __FILE__,
+                       __LINE__, __FUNCTION__, ++i, msg_handler->topic_filter);
+            __platform_printf("[%d]subscribe topic: %s\n",++i, msg_handler->topic_filter);
+            /* 用 arg_new<type> 的 api 创建 arg */
+            Arg* str_arg1 = arg_newStr((char*)msg_handler->topic_filter);
+            /* 添加到 list 对象 */
+            PikaStdData_List_append(list, str_arg1);
+            arg_deinit(str_arg1);
+        }
+    }
+    return list;
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -183,7 +202,7 @@ PikaObj* _mqtt__MQTT_listSubscribrTopic(PikaObj* self) {
 // 输入参数：主题名称，有效数据
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
-int _mqtt__MQTT_publish(PikaObj *self, int qos, char* topic, char* payload) {
+int _mqtt__MQTT_publish(PikaObj *self, char* topic, char* payload, int qos) {
     int ret;
     mqtt_message_t msg;
 
@@ -206,11 +225,12 @@ int _mqtt__MQTT_publish(PikaObj *self, int qos, char* topic, char* payload) {
 
     msg.payload = (void*)payload;
     msg.qos = qos;
-    __platform_printf("msg.qos:%d\r\n",msg.qos);
+    __platform_printf("msg.qos:%d\r\n",
+                      msg.qos);  // 这里为了防止被优化，导致运行异常
     ret = mqtt_publish(_client, topic, &msg);
-    if (ret == 0)
+    if (ret == 0) {
-        __platform_printf("MQTT_publish OK\r\n");
+        // __platform_printf("MQTT_publish OK\r\n");
-    else
+    } else
         __platform_printf("MQTT_publish ERROR\r\n");
     return ret;
 }
@@ -236,7 +256,7 @@ int _mqtt__MQTT_setCa(PikaObj* self, char* ca) {
     obj_setStr(self, "ca_str", ca);
     mqtt_set_ca(_client, obj_getStr(self, "ca_str"));
 
-    __platform_printf("MQTT_setCa len:%d\r\n", strlen(ca));
+    // __platform_printf("MQTT_setCa len:%d\r\n", strlen(ca));
     return 0;
 }
 
@@ -261,7 +281,7 @@ int _mqtt__MQTT_setClientID(PikaObj* self, char* id) {
     obj_setStr(self, "id_str", id);
     mqtt_set_client_id(_client, obj_getStr(self, "id_str"));
 
-    __platform_printf("MQTT_setClientID :%s\r\n", id);
+    // __platform_printf("MQTT_setClientID :%s\r\n", id);
     return 0;
 }
 
@@ -283,9 +303,13 @@ int _mqtt__MQTT_setHost(PikaObj* self, char* host_url) {
         return -2;
     }
 
-    obj_setStr(self, "host_str",host_url);  // python 环境创建一个全局变量存放 host
+    obj_setStr(self, "host_str",
-    mqtt_set_host(_client,obj_getStr(self,"host_str"));  //从python环境中取出 host的指针 赋值给结构体
+               host_url);  // python 环境创建一个全局变量存放 host
-    __platform_printf("MQTT_setHost :%s\r\n", host_url);
+    mqtt_set_host(
+        _client,
+        obj_getStr(self,
+                   "host_str"));  // 从python环境中取出 host的指针 赋值给结构体
+    // __platform_printf("MQTT_setHost :%s\r\n", host_url);
 
     return 0;
 }
@@ -296,11 +320,12 @@ int _mqtt__MQTT_setHost(PikaObj* self, char* host_url) {
 // 输入参数：字符串格式
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
-int _mqtt__MQTT_setKeepAlive(PikaObj* self, char* time) {
+int _mqtt__MQTT_setKeepAlive(PikaObj* self, int time) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
     int tmp;
 
-    tmp = atoi(time);
+    // tmp = atoi(time);
+    tmp = time;
     if (tmp > 0) {
         mqtt_set_keep_alive_interval(_client, tmp);
     } else {
@@ -308,7 +333,7 @@ int _mqtt__MQTT_setKeepAlive(PikaObj* self, char* time) {
         return -2;
     }
 
-    __platform_printf("MQTT_setKeepAlive :%d\r\n", tmp);
+    // __platform_printf("MQTT_setKeepAlive :%d\r\n", tmp);
     return 0;
 }
 
@@ -333,7 +358,7 @@ int _mqtt__MQTT_setPassword(PikaObj* self, char* passwd) {
     obj_setStr(self, "password_str", passwd);
     mqtt_set_password(_client, obj_getStr(self, "password_str"));
 
-    __platform_printf("MQTT_setPassword :%s\r\n", passwd);
+    // __platform_printf("MQTT_setPassword :%s\r\n", passwd);
     return 0;
 }
 
@@ -356,7 +381,7 @@ int _mqtt__MQTT_setPort(PikaObj* self, int port) {
     obj_setStr(self, "port", port_str);
 
     mqtt_set_port(_client, obj_getStr(self, "port"));
-    __platform_printf("MQTT_setPort :%s\r\n", port_str);
+    // __platform_printf("MQTT_setPort :%s\r\n", port_str);
 
     return 0;
 }
@@ -382,7 +407,7 @@ int _mqtt__MQTT_setUsername(PikaObj* self, char* name) {
     obj_setStr(self, "username_str", name);
     mqtt_set_user_name(_client, obj_getStr(self, "username_str"));
 
-    __platform_printf("MQTT_setUsername :%s\r\n", name);
+    // __platform_printf("MQTT_setUsername :%s\r\n", name);
     return 0;
 }
 
@@ -394,17 +419,20 @@ int _mqtt__MQTT_setUsername(PikaObj* self, char* name) {
 ///////////////////////////////////////////////////////////////////
 int _mqtt__MQTT_setVersion(PikaObj* self, char* version) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
-    int tmp;
+    // int tmp;
 
-    tmp = atoi(version);
+    if (version == NULL) {
-    if (tmp > 0) {
+        __platform_printf("input version str error\n");
-        mqtt_set_version(_client, tmp);
+        return -1;
+    }
+    if ((strcmp(version, "3.1") == 0) || (strcmp(version, "3.1.1") == 0)) {
+        mqtt_set_version(_client, 3);
     } else {
-        __platform_printf("input version data error \r\n");
+        __platform_printf("input version data error\n");
         return -2;
     }
 
-    __platform_printf("MQTT_setVersion :%d\r\n", tmp);
+    // __platform_printf("MQTT_setVersion :%d\r\n", tmp);
     return 0;
 }
 
@@ -415,15 +443,15 @@ int _mqtt__MQTT_setVersion(PikaObj* self, char* version) {
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
 int _mqtt__MQTT_setWill(PikaObj* self,
-                        int qos,
                         char* topic,
-                        int retain,
+                        char* payload,
-                        char* payload) {
+                        int qos,
+                        int retain) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
     int ret;
     char topic_str[MQTT_TOPIC_LEN_MAX];
 
-    __platform_printf("\r\n");
+    // __platform_printf("\r\n");
     if (topic == NULL) {
         __platform_printf("input topic error\r\n");
         return -1;
@@ -449,7 +477,7 @@ int _mqtt__MQTT_setWill(PikaObj* self,
         return -1;
     }
 
-    __platform_printf("input retain :%d\r\n", (uint8_t)retain);
+    // __platform_printf("input retain :%d\r\n", (uint8_t)retain);
 
     // 必须转换成python环境的变量，否则函数退出后，topic里的是个空指针
     memset(topic_str, 0, sizeof(topic_str));
@@ -457,10 +485,12 @@ int _mqtt__MQTT_setWill(PikaObj* self,
     obj_setStr(self, topic_str, topic);
     obj_setStr(self, "Will_payload", payload);
 
-    ret = mqtt_set_will_options(_client, obj_getStr(self, topic_str), qos, (uint8_t)retain, obj_getStr(self, "Will_payload"));
+    ret = mqtt_set_will_options(_client, obj_getStr(self, topic_str), qos,
+                                (uint8_t)retain,
+                                obj_getStr(self, "Will_payload"));
 
     if (ret == 0) {
-        __platform_printf("MQTT_setWill OK\r\n", topic);
+        // __platform_printf("MQTT_setWill OK\r\n", topic);
     } else
         __platform_printf("MQTT_setWill ERROR\r\n");
 
@@ -473,24 +503,26 @@ int _mqtt__MQTT_setWill(PikaObj* self,
 // 输入参数：
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
-int _mqtt__MQTT_subscribe(PikaObj* self, char* topic, int qos, Arg* cb) {
+int _mqtt__MQTT_subscribe(PikaObj *self, char* topic, Arg* cb, int qos) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
     int ret;
     char topic_str[MQTT_TOPIC_LEN_MAX + 24];
 
+    // __platform_printf("topic_str:%s \r\n",topic_str);
     if (topic == NULL) {
         __platform_printf("input topic error\r\n");
         return -1;
     }
 
     if ((strlen(topic) > MQTT_TOPIC_LEN_MAX) || (strlen(topic) <= 0)) {
-        __platform_printf("input topic error\r\n");
+        __platform_printf("input topic data error strlen(topic):%d\r\n",
-        return -1;
+                          strlen(topic));
+        return -2;
     }
 
     if ((qos < 0) || (qos > 2)) {
         __platform_printf("input qos error\r\n");
-        return -1;
+        return -3;
     }
 
     // 必须转换成python环境的变量，否则函数退出后，topic里的是个空指针
@@ -498,21 +530,25 @@ int _mqtt__MQTT_subscribe(PikaObj* self, char* topic, int qos, Arg* cb) {
     sprintf(topic_str, "%s", topic);
     obj_setStr(self, topic_str, topic);
 
-    ret = mqtt_subscribe(_client, obj_getStr(self, topic_str), qos, Subscribe_Handler);
+    ret = mqtt_subscribe(_client, obj_getStr(self, topic_str), qos,
+                         Subscribe_Handler);
     if (ret == 0) {
-        __platform_printf("MQTT_subscribe Topic :%s Qos:%d OK\r\n", topic,qos);
+        // __platform_printf("MQTT_subscribe Topic :%s Qos:%d OK\r\n", topic,qos);
         //注册mqtt订阅主题的 回调函数
         if(cb != NULL) {
-            memset(topic_str,0,sizeof(topic_str));
+            char hash_str[32] = {0};
-            sprintf(topic_str,"eventCallBack_%s",topic);
+            memset(hash_str,0,sizeof(hash_str));
-            __platform_printf("topic_str:%s \r\n",topic_str);
+            sprintf(hash_str,"C%d",hash_time33(topic_str));
-            obj_setArg(self, topic_str, cb);
+            obj_newDirectObj(self,hash_str,New_TinyObj);//新建一个对象来放CB
+            PikaObj* eventHandler = obj_getPtr(self,hash_str);
+            obj_setArg(eventHandler, "eventCallBack", cb);
             /* init event_listener for the first time */
             if (NULL == g_mqtt_event_listener) {
-                pks_eventLisener_init(&g_mqtt_event_listener);
+                pks_eventListener_init(&g_mqtt_event_listener);
             }
-            uint32_t eventId = hash_time33(topic);
+            uint32_t eventId = hash_time33(topic_str);
-            pks_eventLicener_registEvent(g_mqtt_event_listener, eventId, self);
+            // __platform_printf("hash_time33(topic_str):%d \r\n",hash_time33(topic_str));
+            pks_eventListener_registEvent(g_mqtt_event_listener, eventId, eventHandler);
         }
 
     } else
@@ -538,7 +574,7 @@ int _mqtt__MQTT_unsubscribe(PikaObj* self, char* topic) {
 
     ret = mqtt_unsubscribe(_client, topic);
     if (ret == 0) {
-        __platform_printf("MQTT_unsubscribe :%s OK\r\n", topic);
+        // __platform_printf("MQTT_unsubscribe :%s OK\r\n", topic);
     } else
         __platform_printf("MQTT_unsubscribe :%s ERROR\r\n", topic);
 
@@ -554,31 +590,102 @@ int _mqtt__MQTT_unsubscribe(PikaObj* self, char* topic) {
 void Subscribe_Handler(void* client, message_data_t* msg) {
     char topic_str[MQTT_TOPIC_LEN_MAX + 24];
     PikaObj* self = ((mqtt_client_t*)client)->user_data;
+    char hash_str[32] = {0};
 
-    //防止数组约界
     memset(topic_str,0,sizeof(topic_str));
     if(strlen(msg->topic_name) <= MQTT_TOPIC_LEN_MAX) 
-        sprintf(topic_str,"eventCallBack_%s",msg->topic_name);
+        sprintf(topic_str,"%s",msg->topic_name);
     else {
-        sprintf(topic_str,"eventCallBack_");
+        __platform_printf("Subscribe Topic recv data topic length ERROR\r\n");
-        memcpy((topic_str+strlen("eventCallBack_")),msg->topic_name,MQTT_TOPIC_LEN_MAX);
+        return ;
-    }
-    __platform_printf("topic_str:%s \r\n",topic_str);
-
-    Arg* cb = obj_getArg(self, topic_str);
-    // obj_setStr(self, "recv_topic", msg->topic_name);
-    // obj_setStr(self, "recv_msg", msg->message->payload);
-    pks_eventLisener_sendSignal(g_mqtt_event_listener,
-                                hash_time33(msg->topic_name), 1);
-
-    MQTT_LOG_I("\n>>>------------------");
-    MQTT_LOG_I("Topic:%s \nlen:%d,message: %s", msg->topic_name,
-               (int)msg->message->payloadlen, (char*)msg->message->payload);
-    MQTT_LOG_I("------------------<<<");
     }
     
+    memset(hash_str,0,sizeof(hash_str));
+    sprintf(hash_str,"M%d",hash_time33(msg->topic_name));
+    obj_setStr(self, hash_str, (char*)msg->message->payload);
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "T%d", hash_time33(msg->topic_name));
+    obj_setStr(self, hash_str, (char*)msg->topic_name);
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "Q%d", hash_time33(msg->topic_name));
+    obj_setInt(self, hash_str, msg->message->qos);
+
+    //存好数据后，再发送事件信号，防止信号收到了但是需要传输的数据没准备好
+    pks_eventListener_sendSignal(g_mqtt_event_listener,
+                                hash_time33(msg->topic_name), hash_time33(msg->topic_name));
+    
+    // MQTT_LOG_I("\n>>>------------------");
+    // MQTT_LOG_I("Topic:%s \nlen:%d,message: %s", msg->topic_name,
+    //            (int)msg->message->payloadlen, (char*)msg->message->payload);
+    // MQTT_LOG_I("------------------<<<");
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt___del__
+// 功能说明：释放事件处理器
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
 void _mqtt___del__(PikaObj* self) {
     if (NULL != g_mqtt_event_listener) {
-        pks_eventLisener_deinit(&g_mqtt_event_listener);
+        pks_eventListener_deinit(&g_mqtt_event_listener);
     }
 }
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_getMsg
+// 功能说明：在回调函数中取出返回的数据
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+char* _mqtt__MQTT_getMsg(PikaObj* self, int signal) {
+    // mqtt_client_t* _client = obj_getPtr(self, "_client");
+    char hash_str[32];
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "M%d", signal);
+    return (obj_getStr(self, hash_str));
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_getTopic
+// 功能说明：在回调函数中取出返回的数据,主题
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+char* _mqtt__MQTT_getTopic(PikaObj* self, int signal) {
+    char hash_str[32];
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "T%d", signal);
+    return (obj_getStr(self, hash_str));
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_getQos
+// 功能说明：在回调函数中取出返回的数据,消息类型
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+int _mqtt__MQTT_getQos(PikaObj* self, int signal) {
+    char hash_str[32];
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "Q%d", signal);
+    return (obj_getInt(self, hash_str));
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_setDisconnectHandler
+// 功能说明：设置断开连接的回调函数
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+int _mqtt__MQTT_setDisconnectHandler(PikaObj* self, Arg* cb) {
+    // mqtt_client_t* _client = obj_getPtr(self, "_client");
+
+    __platform_printf("_mqtt__MQTT_setDisconnectHandler\r\n");
+    return 0;
+}

package/mqtt/mqtt.py

@@ -10,7 +10,19 @@ class MQTT(_mqtt._MQTT):
                  password='',
                  version='3.1.1',
                  ca='',
-                 keepalive=10):
+                 keepalive=60):
         super().__init__(ip, port, clinetID,
                          username, password, version,
                          ca, keepalive)
+
+    def subscribe(self, topic, cb, qos=0):
+        return super().subscribe(topic, cb, qos)
+
+    def publish(self, topic, payload, qos=0):
+        return super().publish(topic, payload, qos)
+
+    def setWill(self, topic, payload, qos=0, retain=0):
+        return super().setWill(topic, payload, qos, retain)
+
+    def unsubscribe(self, topic=''):
+        return super().unsubscribe(topic)

package/mqtt/mqtt_config.h

@@ -9,7 +9,7 @@
 
 // #define             MQTT_LOG_IS_SALOF
 
-#define             MQTT_LOG_LEVEL                      MQTT_LOG_INFO_LEVEL   //MQTT_LOG_WARN_LEVEL MQTT_LOG_DEBUG_LEVEL
+#define             MQTT_LOG_LEVEL                      MQTT_LOG_WARN_LEVEL   //MQTT_LOG_WARN_LEVEL MQTT_LOG_DEBUG_LEVEL
 
 #ifdef MQTT_LOG_IS_SALOF
     #define         SALOF_USING_LOG                     (1U)

package/mqtt/mqttclient.c

@@ -1022,7 +1022,7 @@ static void mqtt_yield_thread(void* arg) {
     while (1) {
         rc = mqtt_yield(c, c->mqtt_cmd_timeout);
         if (MQTT_CLEAN_SESSION_ERROR == rc) {
-            MQTT_LOG_W("%s:%d %s()..., mqtt clean session....", __FILE__,
+            MQTT_LOG_I("%s:%d %s()..., mqtt clean session....", __FILE__,
                        __LINE__, __FUNCTION__);
             network_disconnect(c->mqtt_network);
             mqtt_clean_session(c);
@@ -1168,6 +1168,7 @@ static uint32_t mqtt_read_buf_malloc(mqtt_client_t* c, uint32_t size) {
                    __LINE__, __FUNCTION__);
         RETURN_ERROR(MQTT_MEM_NOT_ENOUGH_ERROR);
     }
+    memset(c->mqtt_read_buf,0,c->mqtt_read_buf_size);//清空申请的内存
     return c->mqtt_read_buf_size;
 }
 
@@ -1569,9 +1570,10 @@ int mqtt_list_subscribe_topic(mqtt_client_t* c) {
     if (NULL == c)
         RETURN_ERROR(MQTT_NULL_VALUE_ERROR);
 
-    if (mqtt_list_is_empty(&c->mqtt_msg_handler_list))
+    if (mqtt_list_is_empty(&c->mqtt_msg_handler_list)) {
         MQTT_LOG_I("%s:%d %s()... there are no subscribed topics...", __FILE__,
                    __LINE__, __FUNCTION__);
+    }
     
     LIST_FOR_EACH_SAFE(curr, next, &c->mqtt_msg_handler_list) {
         msg_handler = LIST_ENTRY(curr, message_handlers_t, list);
@@ -1629,9 +1631,9 @@ int mqtt_release_free(mqtt_client_t* c) {
        mqtt_clean_session(c);
     }
 
-    MQTT_LOG_E("%s:%d %s() 1", __FILE__,__LINE__, __FUNCTION__);
+    MQTT_LOG_I("%s:%d %s() 1", __FILE__,__LINE__, __FUNCTION__);
     mqtt_release(c);
-    MQTT_LOG_E("%s:%d %s() 2", __FILE__,__LINE__, __FUNCTION__);
+    MQTT_LOG_I("%s:%d %s() 2", __FILE__,__LINE__, __FUNCTION__);
     platform_memory_free(c);
     return 0;
 }

port/linux/.vscode/settings.json

@@ -93,7 +93,8 @@
         "__instruction_def.h": "c",
         "pika_hal_def.h": "c",
         "pika_hal_table.h": "c",
-        "pika_hal_table_rule.h": "c"
+        "pika_hal_table_rule.h": "c",
+        "pikaobj.h": "c"
     },
     "python.formatting.provider": "autopep8",
     "C_Cpp.errorSquiggles": "Disabled"

port/linux/package/pikascript/_mqtt.pyi

@@ -26,11 +26,11 @@ class _MQTT:
         pass
         """Set the Ca of the MQTTClient."""
 
-    def setKeepAlive(self, time: str) -> int:
+    def setKeepAlive(self, time: int) -> int:
         pass
         """Set the KeepAlive of the MQTTClient."""
 
-    def setWill(self, qos: int, topic: str, retain: int, payload: str) -> int:
+    def setWill(self, topic: str, payload: str, qos: int, retain: int) -> int:
         pass
         """Set the Will of the MQTTClient."""
 
@@ -42,7 +42,7 @@ class _MQTT:
         pass
         """disconnect to the mqtt-server."""
 
-    def subscribe(self, topic: str, qos: int, cb: any) -> int:
+    def subscribe(self, topic: str, cb: any, qos: int) -> int:
         pass
         """subscribe to the mqtt-server."""
 
@@ -50,11 +50,11 @@ class _MQTT:
         pass
         """unsubscribe to the mqtt-server."""
 
-    def listSubscribrTopic(self) -> list:
+    def listSubscribeTopic(self) -> list:
         pass
-        """listSubscribrTopic """
+        """listSubscribeTopic """
 
-    def publish(self, qos:int, topic: str, payload: str) -> int:
+    def publish(self,topic: str, payload: str, qos:int) -> int:
         pass
         """publish to the mqtt-server."""
 
@@ -64,6 +64,22 @@ class _MQTT:
     def setHost(self, host_url: str) -> int:
         """Set the host_url of the MQTTClient."""
 
+    def getMsg(self,signal:int) -> str:
+        pass
+        """callback fun get msg"""
+
+    def getTopic(self,signal:int) -> str:
+        pass
+        """callback fun get topic"""
+        
+    def getQos(self,signal:int) -> int:
+        pass
+        """callback fun get qos"""
+        
+    def setDisconnectHandler(self,cb: any) -> int:
+        pass
+        """set disconnect callback fun."""
+
 
 def __del__():
     pass

port/linux/package/pikascript/mqtt.py

@@ -10,7 +10,19 @@ class MQTT(_mqtt._MQTT):
                  password='',
                  version='3.1.1',
                  ca='',
-                 keepalive=10):
+                 keepalive=60):
         super().__init__(ip, port, clinetID,
                          username, password, version,
                          ca, keepalive)
+
+    def subscribe(self, topic, cb, qos=0):
+        return super().subscribe(topic, cb, qos)
+
+    def publish(self, topic, payload, qos=0):
+        return super().publish(topic, payload, qos)
+
+    def setWill(self, topic, payload, qos=0, retain=0):
+        return super().setWill(topic, payload, qos, retain)
+
+    def unsubscribe(self, topic=''):
+        return super().unsubscribe(topic)

port/linux/package/pikascript/pikascript-lib/PikaStdDevice/PikaStdDevice_BaseDev.c

@@ -15,7 +15,7 @@ void PikaStdDevice_BaseDev_addEventCallBack(PikaObj* self, Arg* eventCallBack) {
     obj_setArg(self, "eventCallBack", eventCallBack);
     /* init event_listener for the first time */
     if (NULL == g_pika_device_event_listener) {
-        pks_eventLisener_init(&g_pika_device_event_listener);
+        pks_eventListener_init(&g_pika_device_event_listener);
     }
     if (PIKA_RES_OK != obj_runNativeMethod(self, "platformGetEventId", NULL)) {
         obj_setErrorCode(self, 1);
@@ -23,7 +23,7 @@ void PikaStdDevice_BaseDev_addEventCallBack(PikaObj* self, Arg* eventCallBack) {
                           "platformGetEventId");
     }
     uint32_t eventId = obj_getInt(self, "eventId");
-    pks_eventLicener_registEvent(g_pika_device_event_listener, eventId, self);
+    pks_eventListener_registEvent(g_pika_device_event_listener, eventId, self);
 #else
     obj_setErrorCode(self, 1);
     obj_setSysOut(self, "[error] PIKA_EVENT_ENABLE is disabled.");

port/linux/package/pikascript/pikascript-lib/TemplateDevice/TemplateDevice.c

@@ -3,5 +3,5 @@
 extern PikaEventListener* g_pika_device_event_listener;
 
 void TemplateDevice___del__(PikaObj *self){
-    pks_eventLisener_deinit(&g_pika_device_event_listener);
+    pks_eventListener_deinit(&g_pika_device_event_listener);
 }

port/linux/package/pikascript/pikascript-lib/TemplateDevice/TemplateDevice_GPIO.c

@@ -21,8 +21,8 @@ extern PikaEventListener* g_pika_device_event_listener;
 #define GPIO_PA8_EVENT_ID 0x08
 
 void TemplateDevice_GPIO_eventTest(PikaObj* self) {
-    pks_eventLisener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
+    pks_eventListener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
                                 EVENT_SIGNAL_IO_FALLING_EDGE);
-    pks_eventLisener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
+    pks_eventListener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
                                 EVENT_SIGNAL_IO_RISING_EDGE);
 }

port/linux/package/pikascript/pikascript-lib/mqtt/_mqtt__MQTT.c

@@ -1,6 +1,8 @@
 #include "_mqtt__MQTT.h"
+#include "PikaStdData_List.h"
+#include "TinyObj.h"
 #include "mqttclient.h"
-
+#include "PikaObj.h"
 PikaEventListener* g_mqtt_event_listener = NULL;
 
 void Subscribe_Handler(void* client, message_data_t* msg);
@@ -31,8 +33,7 @@ void _mqtt__MQTT___init__(PikaObj* self,
     if (strlen(ip) > 0) {
         obj_setStr(self, "host_str", ip);
         mqtt_set_host(_client, obj_getStr(self, "host_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input ip none\r\n");
     }
 
@@ -44,47 +45,41 @@ void _mqtt__MQTT___init__(PikaObj* self,
     if (strlen(clinetID) > 0) {
         obj_setStr(self, "id_str", clinetID);
         mqtt_set_client_id(_client, obj_getStr(self, "id_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input clinetID none\r\n");
     }
 
     if (strlen(username) > 0) {
         obj_setStr(self, "username_str", username);
         mqtt_set_user_name(_client, obj_getStr(self, "username_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input username none\r\n");
     }
 
     if (strlen(password) > 0) {
         obj_setStr(self, "password_str", password);
         mqtt_set_password(_client, obj_getStr(self, "password_str"));
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input password none\r\n");
     }
 
     tmp = atoi(version);
     if (tmp > 0) {
         mqtt_set_version(_client, tmp);
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input version none\r\n");
     }
 
     if (strlen(ca) > 0) {
         obj_setStr(self, "ca_str", ca);
         mqtt_set_ca(_client, obj_getStr(self, "ca_str"));
-    }
+    } else {
-    else {
+        // __platform_printf("mqtt_init input ca none\r\n");
-        __platform_printf("mqtt_init input ca none\r\n");
     }
 
     if (keepalive > 0) {
         mqtt_set_keep_alive_interval(_client, keepalive);
-    }
+    } else {
-    else {
         __platform_printf("mqtt_init input keepalive none\r\n");
     }
 
@@ -92,7 +87,7 @@ void _mqtt__MQTT___init__(PikaObj* self,
 
     obj_setPtr(self, "_client",
                _client);  // 这里要再保存一次mqtt结构体的内容到python环境
-    __platform_printf("Mqtt_Lib buildtime:%s-%s\r\n", __DATE__, __TIME__);
+    // __platform_printf("Mqtt_Lib buildtime:%s-%s\r\n", __DATE__, __TIME__);
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -111,7 +106,7 @@ void _mqtt__MQTT___del__(PikaObj* self) {
     if (_connected) {
         _mqtt__MQTT_disconnect(self);
     }
-    MQTT_LOG_E("%s:%d %s() >_<", __FILE__, __LINE__, __FUNCTION__);
+    // MQTT_LOG_E("%s:%d %s() >_<", __FILE__, __LINE__, __FUNCTION__);
     mqtt_release_free(_client);
 }
 
@@ -130,8 +125,8 @@ int _mqtt__MQTT_connect(PikaObj* self) {
     if (ret != 0)
         __platform_printf("mqtt connect ERROR! :%d\r\n", ret);
 
-    if (ret == 0)
+    // if (ret == 0)
-        __platform_printf("mqtt connect OK\r\n");
+    //     __platform_printf("mqtt connect OK\r\n");
     return ret;
 }
 
@@ -150,31 +145,55 @@ int _mqtt__MQTT_disconnect(PikaObj* self) {
     if (ret != 0)
         __platform_printf("mqtt disconnect ERROR! :%d\r\n", ret);
 
-    if (ret == 0)
+    // if (ret == 0)
-        __platform_printf("mqtt disconnect OK\r\n");
+    //     __platform_printf("mqtt disconnect OK\r\n");
     return ret;
 }
 
 ////////////////////////////////////////////////////////////////////
-// 函 数 名：_mqtt__MQTT_listSubscribrTopic
+// 函 数 名：_mqtt__MQTT_listSubscribeTopic
 // 功能说明：罗列出当前订阅的主题
 // 输入参数：无
 // 返 回 值：对象指针
 ///////////////////////////////////////////////////////////////////
-PikaObj* _mqtt__MQTT_listSubscribrTopic(PikaObj* self) {
+PikaObj* _mqtt__MQTT_listSubscribeTopic(PikaObj* self) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
-    int ret;
+    int i = 0;
-    PikaObj* pt_out = NULL;
+    mqtt_list_t *curr, *next;
+    message_handlers_t* msg_handler;
+    PikaObj* list = NULL;
 
-    ret = mqtt_list_subscribe_topic(_client);
+    if (NULL == _client) {
-    if (ret == 0)
+        return NULL;
-        __platform_printf("MQTT_listSubscribrTopic OK\r\n");
-    else {
-        pt_out = NULL;
-        __platform_printf("MQTT_listSubscribrTopic ERROR\r\n");
     }
 
-    return pt_out;
+    if (mqtt_list_is_empty(&_client->mqtt_msg_handler_list)) {
+        MQTT_LOG_I("%s:%d %s()... there are no subscribed topics...", __FILE__,
+                   __LINE__, __FUNCTION__);
+        return NULL;
+    }
+
+    /* 创建 list 对象 */
+    list = newNormalObj(New_PikaStdData_List);
+    /* 初始化 list */
+    PikaStdData_List___init__(list);
+
+    LIST_FOR_EACH_SAFE(curr, next, &_client->mqtt_msg_handler_list) {
+        msg_handler = LIST_ENTRY(curr, message_handlers_t, list);
+        /* determine whether a node already exists by mqtt topic, but wildcards
+         * are not supported */
+        if (NULL != msg_handler->topic_filter) {
+            MQTT_LOG_I("%s:%d %s()...[%d] subscribe topic: %s", __FILE__,
+                       __LINE__, __FUNCTION__, ++i, msg_handler->topic_filter);
+            __platform_printf("[%d]subscribe topic: %s\n",++i, msg_handler->topic_filter);
+            /* 用 arg_new<type> 的 api 创建 arg */
+            Arg* str_arg1 = arg_newStr((char*)msg_handler->topic_filter);
+            /* 添加到 list 对象 */
+            PikaStdData_List_append(list, str_arg1);
+            arg_deinit(str_arg1);
+        }
+    }
+    return list;
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -183,7 +202,7 @@ PikaObj* _mqtt__MQTT_listSubscribrTopic(PikaObj* self) {
 // 输入参数：主题名称，有效数据
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
-int _mqtt__MQTT_publish(PikaObj *self, int qos, char* topic, char* payload) {
+int _mqtt__MQTT_publish(PikaObj *self, char* topic, char* payload, int qos) {
     int ret;
     mqtt_message_t msg;
 
@@ -206,11 +225,12 @@ int _mqtt__MQTT_publish(PikaObj *self, int qos, char* topic, char* payload) {
 
     msg.payload = (void*)payload;
     msg.qos = qos;
-    __platform_printf("msg.qos:%d\r\n",msg.qos);
+    __platform_printf("msg.qos:%d\r\n",
+                      msg.qos);  // 这里为了防止被优化，导致运行异常
     ret = mqtt_publish(_client, topic, &msg);
-    if (ret == 0)
+    if (ret == 0) {
-        __platform_printf("MQTT_publish OK\r\n");
+        // __platform_printf("MQTT_publish OK\r\n");
-    else
+    } else
         __platform_printf("MQTT_publish ERROR\r\n");
     return ret;
 }
@@ -236,7 +256,7 @@ int _mqtt__MQTT_setCa(PikaObj* self, char* ca) {
     obj_setStr(self, "ca_str", ca);
     mqtt_set_ca(_client, obj_getStr(self, "ca_str"));
 
-    __platform_printf("MQTT_setCa len:%d\r\n", strlen(ca));
+    // __platform_printf("MQTT_setCa len:%d\r\n", strlen(ca));
     return 0;
 }
 
@@ -261,7 +281,7 @@ int _mqtt__MQTT_setClientID(PikaObj* self, char* id) {
     obj_setStr(self, "id_str", id);
     mqtt_set_client_id(_client, obj_getStr(self, "id_str"));
 
-    __platform_printf("MQTT_setClientID :%s\r\n", id);
+    // __platform_printf("MQTT_setClientID :%s\r\n", id);
     return 0;
 }
 
@@ -283,9 +303,13 @@ int _mqtt__MQTT_setHost(PikaObj* self, char* host_url) {
         return -2;
     }
 
-    obj_setStr(self, "host_str",host_url);  // python 环境创建一个全局变量存放 host
+    obj_setStr(self, "host_str",
-    mqtt_set_host(_client,obj_getStr(self,"host_str"));  //从python环境中取出 host的指针 赋值给结构体
+               host_url);  // python 环境创建一个全局变量存放 host
-    __platform_printf("MQTT_setHost :%s\r\n", host_url);
+    mqtt_set_host(
+        _client,
+        obj_getStr(self,
+                   "host_str"));  // 从python环境中取出 host的指针 赋值给结构体
+    // __platform_printf("MQTT_setHost :%s\r\n", host_url);
 
     return 0;
 }
@@ -296,11 +320,12 @@ int _mqtt__MQTT_setHost(PikaObj* self, char* host_url) {
 // 输入参数：字符串格式
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
-int _mqtt__MQTT_setKeepAlive(PikaObj* self, char* time) {
+int _mqtt__MQTT_setKeepAlive(PikaObj* self, int time) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
     int tmp;
 
-    tmp = atoi(time);
+    // tmp = atoi(time);
+    tmp = time;
     if (tmp > 0) {
         mqtt_set_keep_alive_interval(_client, tmp);
     } else {
@@ -308,7 +333,7 @@ int _mqtt__MQTT_setKeepAlive(PikaObj* self, char* time) {
         return -2;
     }
 
-    __platform_printf("MQTT_setKeepAlive :%d\r\n", tmp);
+    // __platform_printf("MQTT_setKeepAlive :%d\r\n", tmp);
     return 0;
 }
 
@@ -333,7 +358,7 @@ int _mqtt__MQTT_setPassword(PikaObj* self, char* passwd) {
     obj_setStr(self, "password_str", passwd);
     mqtt_set_password(_client, obj_getStr(self, "password_str"));
 
-    __platform_printf("MQTT_setPassword :%s\r\n", passwd);
+    // __platform_printf("MQTT_setPassword :%s\r\n", passwd);
     return 0;
 }
 
@@ -356,7 +381,7 @@ int _mqtt__MQTT_setPort(PikaObj* self, int port) {
     obj_setStr(self, "port", port_str);
 
     mqtt_set_port(_client, obj_getStr(self, "port"));
-    __platform_printf("MQTT_setPort :%s\r\n", port_str);
+    // __platform_printf("MQTT_setPort :%s\r\n", port_str);
 
     return 0;
 }
@@ -382,7 +407,7 @@ int _mqtt__MQTT_setUsername(PikaObj* self, char* name) {
     obj_setStr(self, "username_str", name);
     mqtt_set_user_name(_client, obj_getStr(self, "username_str"));
 
-    __platform_printf("MQTT_setUsername :%s\r\n", name);
+    // __platform_printf("MQTT_setUsername :%s\r\n", name);
     return 0;
 }
 
@@ -394,17 +419,20 @@ int _mqtt__MQTT_setUsername(PikaObj* self, char* name) {
 ///////////////////////////////////////////////////////////////////
 int _mqtt__MQTT_setVersion(PikaObj* self, char* version) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
-    int tmp;
+    // int tmp;
 
-    tmp = atoi(version);
+    if (version == NULL) {
-    if (tmp > 0) {
+        __platform_printf("input version str error\n");
-        mqtt_set_version(_client, tmp);
+        return -1;
+    }
+    if ((strcmp(version, "3.1") == 0) || (strcmp(version, "3.1.1") == 0)) {
+        mqtt_set_version(_client, 3);
     } else {
-        __platform_printf("input version data error \r\n");
+        __platform_printf("input version data error\n");
         return -2;
     }
 
-    __platform_printf("MQTT_setVersion :%d\r\n", tmp);
+    // __platform_printf("MQTT_setVersion :%d\r\n", tmp);
     return 0;
 }
 
@@ -415,15 +443,15 @@ int _mqtt__MQTT_setVersion(PikaObj* self, char* version) {
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
 int _mqtt__MQTT_setWill(PikaObj* self,
-                        int qos,
                         char* topic,
-                        int retain,
+                        char* payload,
-                        char* payload) {
+                        int qos,
+                        int retain) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
     int ret;
     char topic_str[MQTT_TOPIC_LEN_MAX];
 
-    __platform_printf("\r\n");
+    // __platform_printf("\r\n");
     if (topic == NULL) {
         __platform_printf("input topic error\r\n");
         return -1;
@@ -449,7 +477,7 @@ int _mqtt__MQTT_setWill(PikaObj* self,
         return -1;
     }
 
-    __platform_printf("input retain :%d\r\n", (uint8_t)retain);
+    // __platform_printf("input retain :%d\r\n", (uint8_t)retain);
 
     // 必须转换成python环境的变量，否则函数退出后，topic里的是个空指针
     memset(topic_str, 0, sizeof(topic_str));
@@ -457,10 +485,12 @@ int _mqtt__MQTT_setWill(PikaObj* self,
     obj_setStr(self, topic_str, topic);
     obj_setStr(self, "Will_payload", payload);
 
-    ret = mqtt_set_will_options(_client, obj_getStr(self, topic_str), qos, (uint8_t)retain, obj_getStr(self, "Will_payload"));
+    ret = mqtt_set_will_options(_client, obj_getStr(self, topic_str), qos,
+                                (uint8_t)retain,
+                                obj_getStr(self, "Will_payload"));
 
     if (ret == 0) {
-        __platform_printf("MQTT_setWill OK\r\n", topic);
+        // __platform_printf("MQTT_setWill OK\r\n", topic);
     } else
         __platform_printf("MQTT_setWill ERROR\r\n");
 
@@ -473,24 +503,26 @@ int _mqtt__MQTT_setWill(PikaObj* self,
 // 输入参数：
 // 返 回 值：0=成功；非0=错误码
 ///////////////////////////////////////////////////////////////////
-int _mqtt__MQTT_subscribe(PikaObj* self, char* topic, int qos, Arg* cb) {
+int _mqtt__MQTT_subscribe(PikaObj *self, char* topic, Arg* cb, int qos) {
     mqtt_client_t* _client = obj_getPtr(self, "_client");
     int ret;
     char topic_str[MQTT_TOPIC_LEN_MAX + 24];
 
+    // __platform_printf("topic_str:%s \r\n",topic_str);
     if (topic == NULL) {
         __platform_printf("input topic error\r\n");
         return -1;
     }
 
     if ((strlen(topic) > MQTT_TOPIC_LEN_MAX) || (strlen(topic) <= 0)) {
-        __platform_printf("input topic error\r\n");
+        __platform_printf("input topic data error strlen(topic):%d\r\n",
-        return -1;
+                          strlen(topic));
+        return -2;
     }
 
     if ((qos < 0) || (qos > 2)) {
         __platform_printf("input qos error\r\n");
-        return -1;
+        return -3;
     }
 
     // 必须转换成python环境的变量，否则函数退出后，topic里的是个空指针
@@ -498,21 +530,25 @@ int _mqtt__MQTT_subscribe(PikaObj* self, char* topic, int qos, Arg* cb) {
     sprintf(topic_str, "%s", topic);
     obj_setStr(self, topic_str, topic);
 
-    ret = mqtt_subscribe(_client, obj_getStr(self, topic_str), qos, Subscribe_Handler);
+    ret = mqtt_subscribe(_client, obj_getStr(self, topic_str), qos,
+                         Subscribe_Handler);
     if (ret == 0) {
-        __platform_printf("MQTT_subscribe Topic :%s Qos:%d OK\r\n", topic,qos);
+        // __platform_printf("MQTT_subscribe Topic :%s Qos:%d OK\r\n", topic,qos);
         //注册mqtt订阅主题的 回调函数
         if(cb != NULL) {
-            memset(topic_str,0,sizeof(topic_str));
+            char hash_str[32] = {0};
-            sprintf(topic_str,"eventCallBack_%s",topic);
+            memset(hash_str,0,sizeof(hash_str));
-            __platform_printf("topic_str:%s \r\n",topic_str);
+            sprintf(hash_str,"C%d",hash_time33(topic_str));
-            obj_setArg(self, topic_str, cb);
+            obj_newDirectObj(self,hash_str,New_TinyObj);//新建一个对象来放CB
+            PikaObj* eventHandler = obj_getPtr(self,hash_str);
+            obj_setArg(eventHandler, "eventCallBack", cb);
             /* init event_listener for the first time */
             if (NULL == g_mqtt_event_listener) {
-                pks_eventLisener_init(&g_mqtt_event_listener);
+                pks_eventListener_init(&g_mqtt_event_listener);
             }
-            uint32_t eventId = hash_time33(topic);
+            uint32_t eventId = hash_time33(topic_str);
-            pks_eventLicener_registEvent(g_mqtt_event_listener, eventId, self);
+            // __platform_printf("hash_time33(topic_str):%d \r\n",hash_time33(topic_str));
+            pks_eventListener_registEvent(g_mqtt_event_listener, eventId, eventHandler);
         }
 
     } else
@@ -538,7 +574,7 @@ int _mqtt__MQTT_unsubscribe(PikaObj* self, char* topic) {
 
     ret = mqtt_unsubscribe(_client, topic);
     if (ret == 0) {
-        __platform_printf("MQTT_unsubscribe :%s OK\r\n", topic);
+        // __platform_printf("MQTT_unsubscribe :%s OK\r\n", topic);
     } else
         __platform_printf("MQTT_unsubscribe :%s ERROR\r\n", topic);
 
@@ -554,31 +590,102 @@ int _mqtt__MQTT_unsubscribe(PikaObj* self, char* topic) {
 void Subscribe_Handler(void* client, message_data_t* msg) {
     char topic_str[MQTT_TOPIC_LEN_MAX + 24];
     PikaObj* self = ((mqtt_client_t*)client)->user_data;
+    char hash_str[32] = {0};
 
-    //防止数组约界
     memset(topic_str,0,sizeof(topic_str));
     if(strlen(msg->topic_name) <= MQTT_TOPIC_LEN_MAX) 
-        sprintf(topic_str,"eventCallBack_%s",msg->topic_name);
+        sprintf(topic_str,"%s",msg->topic_name);
     else {
-        sprintf(topic_str,"eventCallBack_");
+        __platform_printf("Subscribe Topic recv data topic length ERROR\r\n");
-        memcpy((topic_str+strlen("eventCallBack_")),msg->topic_name,MQTT_TOPIC_LEN_MAX);
+        return ;
-    }
-    __platform_printf("topic_str:%s \r\n",topic_str);
-
-    Arg* cb = obj_getArg(self, topic_str);
-    // obj_setStr(self, "recv_topic", msg->topic_name);
-    // obj_setStr(self, "recv_msg", msg->message->payload);
-    pks_eventLisener_sendSignal(g_mqtt_event_listener,
-                                hash_time33(msg->topic_name), 1);
-
-    MQTT_LOG_I("\n>>>------------------");
-    MQTT_LOG_I("Topic:%s \nlen:%d,message: %s", msg->topic_name,
-               (int)msg->message->payloadlen, (char*)msg->message->payload);
-    MQTT_LOG_I("------------------<<<");
     }
     
+    memset(hash_str,0,sizeof(hash_str));
+    sprintf(hash_str,"M%d",hash_time33(msg->topic_name));
+    obj_setStr(self, hash_str, (char*)msg->message->payload);
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "T%d", hash_time33(msg->topic_name));
+    obj_setStr(self, hash_str, (char*)msg->topic_name);
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "Q%d", hash_time33(msg->topic_name));
+    obj_setInt(self, hash_str, msg->message->qos);
+
+    //存好数据后，再发送事件信号，防止信号收到了但是需要传输的数据没准备好
+    pks_eventListener_sendSignal(g_mqtt_event_listener,
+                                hash_time33(msg->topic_name), hash_time33(msg->topic_name));
+    
+    // MQTT_LOG_I("\n>>>------------------");
+    // MQTT_LOG_I("Topic:%s \nlen:%d,message: %s", msg->topic_name,
+    //            (int)msg->message->payloadlen, (char*)msg->message->payload);
+    // MQTT_LOG_I("------------------<<<");
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt___del__
+// 功能说明：释放事件处理器
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
 void _mqtt___del__(PikaObj* self) {
     if (NULL != g_mqtt_event_listener) {
-        pks_eventLisener_deinit(&g_mqtt_event_listener);
+        pks_eventListener_deinit(&g_mqtt_event_listener);
     }
 }
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_getMsg
+// 功能说明：在回调函数中取出返回的数据
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+char* _mqtt__MQTT_getMsg(PikaObj* self, int signal) {
+    // mqtt_client_t* _client = obj_getPtr(self, "_client");
+    char hash_str[32];
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "M%d", signal);
+    return (obj_getStr(self, hash_str));
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_getTopic
+// 功能说明：在回调函数中取出返回的数据,主题
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+char* _mqtt__MQTT_getTopic(PikaObj* self, int signal) {
+    char hash_str[32];
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "T%d", signal);
+    return (obj_getStr(self, hash_str));
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_getQos
+// 功能说明：在回调函数中取出返回的数据,消息类型
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+int _mqtt__MQTT_getQos(PikaObj* self, int signal) {
+    char hash_str[32];
+
+    memset(hash_str, 0, sizeof(hash_str));
+    sprintf(hash_str, "Q%d", signal);
+    return (obj_getInt(self, hash_str));
+}
+
+////////////////////////////////////////////////////////////////////
+// 函 数 名：_mqtt__MQTT_setDisconnectHandler
+// 功能说明：设置断开连接的回调函数
+// 输入参数：
+// 返 回 值：
+///////////////////////////////////////////////////////////////////
+int _mqtt__MQTT_setDisconnectHandler(PikaObj* self, Arg* cb) {
+    // mqtt_client_t* _client = obj_getPtr(self, "_client");
+
+    __platform_printf("_mqtt__MQTT_setDisconnectHandler\r\n");
+    return 0;
+}

port/linux/package/pikascript/pikascript-lib/mqtt/mqtt_config.h

@@ -9,7 +9,7 @@
 
 // #define             MQTT_LOG_IS_SALOF
 
-#define             MQTT_LOG_LEVEL                      MQTT_LOG_INFO_LEVEL   //MQTT_LOG_WARN_LEVEL MQTT_LOG_DEBUG_LEVEL
+#define             MQTT_LOG_LEVEL                      MQTT_LOG_WARN_LEVEL   //MQTT_LOG_WARN_LEVEL MQTT_LOG_DEBUG_LEVEL
 
 #ifdef MQTT_LOG_IS_SALOF
     #define         SALOF_USING_LOG                     (1U)

port/linux/package/pikascript/pikascript-lib/mqtt/mqttclient.c

@@ -1022,7 +1022,7 @@ static void mqtt_yield_thread(void* arg) {
     while (1) {
         rc = mqtt_yield(c, c->mqtt_cmd_timeout);
         if (MQTT_CLEAN_SESSION_ERROR == rc) {
-            MQTT_LOG_W("%s:%d %s()..., mqtt clean session....", __FILE__,
+            MQTT_LOG_I("%s:%d %s()..., mqtt clean session....", __FILE__,
                        __LINE__, __FUNCTION__);
             network_disconnect(c->mqtt_network);
             mqtt_clean_session(c);
@@ -1168,6 +1168,7 @@ static uint32_t mqtt_read_buf_malloc(mqtt_client_t* c, uint32_t size) {
                    __LINE__, __FUNCTION__);
         RETURN_ERROR(MQTT_MEM_NOT_ENOUGH_ERROR);
     }
+    memset(c->mqtt_read_buf,0,c->mqtt_read_buf_size);//清空申请的内存
     return c->mqtt_read_buf_size;
 }
 
@@ -1569,9 +1570,10 @@ int mqtt_list_subscribe_topic(mqtt_client_t* c) {
     if (NULL == c)
         RETURN_ERROR(MQTT_NULL_VALUE_ERROR);
 
-    if (mqtt_list_is_empty(&c->mqtt_msg_handler_list))
+    if (mqtt_list_is_empty(&c->mqtt_msg_handler_list)) {
         MQTT_LOG_I("%s:%d %s()... there are no subscribed topics...", __FILE__,
                    __LINE__, __FUNCTION__);
+    }
     
     LIST_FOR_EACH_SAFE(curr, next, &c->mqtt_msg_handler_list) {
         msg_handler = LIST_ENTRY(curr, message_handlers_t, list);
@@ -1629,9 +1631,9 @@ int mqtt_release_free(mqtt_client_t* c) {
        mqtt_clean_session(c);
     }
 
-    MQTT_LOG_E("%s:%d %s() 1", __FILE__,__LINE__, __FUNCTION__);
+    MQTT_LOG_I("%s:%d %s() 1", __FILE__,__LINE__, __FUNCTION__);
     mqtt_release(c);
-    MQTT_LOG_E("%s:%d %s() 2", __FILE__,__LINE__, __FUNCTION__);
+    MQTT_LOG_I("%s:%d %s() 2", __FILE__,__LINE__, __FUNCTION__);
     platform_memory_free(c);
     return 0;
 }

src/PikaObj.c

@@ -1552,7 +1552,7 @@ char* obj_toStr(PikaObj* self) {
     return obj_getStr(self, "__res");
 }
 
-void pks_eventLicener_registEvent(PikaEventListener* self,
+void pks_eventListener_registEvent(PikaEventListener* self,
                                   uint32_t eventId,
                                   PikaObj* eventHandleObj) {
     Args buffs = {0};
@@ -1564,7 +1564,7 @@ void pks_eventLicener_registEvent(PikaEventListener* self,
     strsDeinit(&buffs);
 }
 
-void pks_eventLicener_removeEvent(PikaEventListener* self, uint32_t eventId) {
+void pks_eventListener_removeEvent(PikaEventListener* self, uint32_t eventId) {
     Args buffs = {0};
     char* event_name =
         strsFormat(&buffs, PIKA_SPRINTF_BUFF_SIZE, "%ld", eventId);
@@ -1572,7 +1572,7 @@ void pks_eventLicener_removeEvent(PikaEventListener* self, uint32_t eventId) {
     strsDeinit(&buffs);
 }
 
-PikaObj* pks_eventLisener_getEventHandleObj(PikaEventListener* self,
+PikaObj* pks_eventListener_getEventHandleObj(PikaEventListener* self,
                                             uint32_t eventId) {
     Args buffs = {0};
     char* event_name =
@@ -1583,21 +1583,21 @@ PikaObj* pks_eventLisener_getEventHandleObj(PikaEventListener* self,
     return eventHandleObj;
 }
 
-void pks_eventLisener_init(PikaEventListener** p_self) {
+void pks_eventListener_init(PikaEventListener** p_self) {
     *p_self = newNormalObj(New_TinyObj);
 }
 
-void pks_eventLisener_deinit(PikaEventListener** p_self) {
+void pks_eventListener_deinit(PikaEventListener** p_self) {
     if (NULL != *p_self) {
         obj_deinit(*p_self);
         *p_self = NULL;
     }
 }
 
-Arg* __eventLisener_runEvent(PikaEventListener* lisener,
+Arg* __eventListener_runEvent(PikaEventListener* lisener,
                              uint32_t eventId,
                              int eventSignal) {
-    PikaObj* handler = pks_eventLisener_getEventHandleObj(lisener, eventId);
+    PikaObj* handler = pks_eventListener_getEventHandleObj(lisener, eventId);
     if (NULL == handler) {
         __platform_printf(
             "Error: can not find event handler by id: [0x%02x]\r\n", eventId);
@@ -1625,7 +1625,7 @@ Arg* __eventLisener_runEvent(PikaEventListener* lisener,
     return res;
 }
 
-void pks_eventLisener_sendSignal(PikaEventListener* self,
+void pks_eventListener_sendSignal(PikaEventListener* self,
                                  uint32_t eventId,
                                  int eventSignal) {
 #if !PIKA_EVENT_ENABLE
@@ -1648,7 +1648,7 @@ void pks_eventLisener_sendSignal(PikaEventListener* self,
 #endif
 }
 
-Arg* pks_eventLisener_sendSignalAwaitResult(PikaEventListener* self,
+Arg* pks_eventListener_sendSignalAwaitResult(PikaEventListener* self,
                                             uint32_t eventId,
                                             int eventSignal) {
     /*
@@ -1661,7 +1661,7 @@ Arg* pks_eventLisener_sendSignalAwaitResult(PikaEventListener* self,
 #else
     extern volatile VMSignal PikaVMSignal;
     int tail = PikaVMSignal.cq.tail;
-    pks_eventLisener_sendSignal(self, eventId, eventSignal);
+    pks_eventListener_sendSignal(self, eventId, eventSignal);
     while (1) {
         Arg* res = PikaVMSignal.cq.res[tail];
         __platform_thread_delay();

src/PikaObj.h

@@ -342,21 +342,21 @@ enum shellCTRL obj_runChar(PikaObj* self, char inputChar);
 
 typedef PikaObj PikaEventListener;
 
-void pks_eventLisener_sendSignal(PikaEventListener* self,
+void pks_eventListener_sendSignal(PikaEventListener* self,
                                  uint32_t eventId,
                                  int eventSignal);
 
-void pks_eventLicener_registEvent(PikaEventListener* self,
+void pks_eventListener_registEvent(PikaEventListener* self,
                                   uint32_t eventId,
                                   PikaObj* eventHandleObj);
 
-void pks_eventLicener_removeEvent(PikaEventListener* self, uint32_t eventId);
+void pks_eventListener_removeEvent(PikaEventListener* self, uint32_t eventId);
 
-PikaObj* pks_eventLisener_getEventHandleObj(PikaEventListener* self,
+PikaObj* pks_eventListener_getEventHandleObj(PikaEventListener* self,
                                             uint32_t eventId);
 
-void pks_eventLisener_init(PikaEventListener** p_self);
+void pks_eventListener_init(PikaEventListener** p_self);
-void pks_eventLisener_deinit(PikaEventListener** p_self);
+void pks_eventListener_deinit(PikaEventListener** p_self);
 PikaObj* methodArg_getDefContext(Arg* method_arg);
 PikaObj* Obj_linkLibraryFile(PikaObj* self, char* input_file_name);
 NewFun obj_getClass(PikaObj* obj);
@@ -499,10 +499,10 @@ void _obj_updateProxyFlag(PikaObj* self);
     _obj_updateProxyFlag((_self))
 
 Arg* _obj_getProp(PikaObj* obj, char* name);
-Arg* __eventLisener_runEvent(PikaEventListener* lisener,
+Arg* __eventListener_runEvent(PikaEventListener* lisener,
                              uint32_t eventId,
                              int eventSignal);
-Arg* pks_eventLisener_sendSignalAwaitResult(PikaEventListener* self,
+Arg* pks_eventListener_sendSignalAwaitResult(PikaEventListener* self,
                                             uint32_t eventId,
                                             int eventSignal);
 

src/PikaVM.c

@@ -136,7 +136,7 @@ void VMSignale_pickupEvent(void) {
     if (PIKA_RES_OK ==
         VMSignal_popEvent(&event_lisener, &event_id, &event_signal, &head)) {
         Arg* res =
-            __eventLisener_runEvent(event_lisener, event_id, event_signal);
+            __eventListener_runEvent(event_lisener, event_id, event_signal);
         PikaVMSignal.cq.res[head] = res;
     }
 #endif

test/event-test.cpp

@@ -14,9 +14,9 @@ TEST(event, gpio) {
 #define GPIO_PA8_EVENT_ID 0x08
 
     /* simulate run in the call back */
-    pks_eventLisener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
+    pks_eventListener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
                                 EVENT_SIGAL_IO_RISING_EDGE);
-    pks_eventLisener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
+    pks_eventListener_sendSignal(g_pika_device_event_listener, GPIO_PA8_EVENT_ID,
                                 EVENT_SIGAL_IO_FALLING_EDGE);
     /* assert */
 
@@ -32,9 +32,9 @@ TEST(event, gpio) {
     }
 
     /* simulate run in the call back */
-    Arg* res_123 = pks_eventLisener_sendSignalAwaitResult(
+    Arg* res_123 = pks_eventListener_sendSignalAwaitResult(
         g_pika_device_event_listener, GPIO_PA8_EVENT_ID, 123);
-    Arg* res_456 = pks_eventLisener_sendSignalAwaitResult(
+    Arg* res_456 = pks_eventListener_sendSignalAwaitResult(
         g_pika_device_event_listener, GPIO_PA8_EVENT_ID, 456);
     
     EXPECT_EQ(arg_getInt(res_123), 123);
@@ -50,16 +50,16 @@ TEST(event, remove_regist) {
     /* init */
     PikaObj* pikaMain = newRootObj("pikaMain", New_PikaMain);
     /* run */
-    pks_eventLisener_init(&g_pika_device_event_listener);
+    pks_eventListener_init(&g_pika_device_event_listener);
     PikaObj* testobj = newNormalObj(New_TinyObj);
-    pks_eventLicener_registEvent(g_pika_device_event_listener, 0, testobj);
+    pks_eventListener_registEvent(g_pika_device_event_listener, 0, testobj);
     EXPECT_EQ(testobj->refcnt, 2);
-    pks_eventLicener_removeEvent(g_pika_device_event_listener, 0);
+    pks_eventListener_removeEvent(g_pika_device_event_listener, 0);
     EXPECT_EQ(testobj->refcnt, 1);
     /* deinit */
     obj_deinit(pikaMain);
     obj_deinit(testobj);
-    pks_eventLisener_deinit(&g_pika_device_event_listener);
+    pks_eventListener_deinit(&g_pika_device_event_listener);
     EXPECT_EQ(pikaMemNow(), 0);
 }
 

test/python/mqtt/mqtt_connect.py

@@ -3,7 +3,7 @@ import mqtt
 client = mqtt.MQTT('broker.emqx.io',port=1883,clinetID='clientid',username='name_',password='passwd_')
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
 
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)

test/python/mqtt/mqtt_publish.py

@@ -7,15 +7,15 @@ client.setPort(1883)
 client.setClientID('123456dddecetdc')
 client.setUsername('test1')
 client.setPassword('aabbccdd')
-client.setVersion('3')
+client.setVersion('3.1')
-client.setKeepAlive('10')
+client.setKeepAlive(10)
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
 
-client.publish(0,'topic_pikapy', 'hello pikascript qos=0')
+client.publish('topic_pikapy', 'hello pikascript qos=0', 0)
-client.publish(1,'topic_pikapy', 'hello pikascript qos=1')
+client.publish('topic_pikapy', 'hello pikascript qos=1', 1)
-client.publish(2,'topic_pikapy', 'hello pikascript qos=2')
+client.publish('topic_pikapy', 'hello pikascript qos=2', 2)
 
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)

test/python/mqtt/mqtt_set_para.py

@@ -35,11 +35,11 @@ client.setPort(1883)
 client.setClientID('123456dddecetdc')
 client.setUsername('j6npr4w/mqtt-client-dev')
 client.setPassword('lcUhUs5VYLMSbrnB')
-client.setVersion('3')
+client.setVersion('3.1.1')
-client.setKeepAlive('10')
+client.setKeepAlive(10)
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
 
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)

test/python/mqtt/mqtt_subscribe.py

@@ -4,36 +4,54 @@ import PikaStdDevice
 client = mqtt.MQTT('broker.emqx.io', port=1883, clinetID='clientid', username='name_', password='passwd_')
 
 ret = client.connect()
-print("ret:%d" % ret)
+print("connect ret:%d" % ret)
+
 
 def callback0(signal):
-    print("py cb: %s:%s" % (client.recv_topic, client.recv_msg))
+    recv_msg = client.getMsg(signal)
+    recv_topic = client.getTopic(signal)
+    recv_qos = client.getQos(signal)
+    print("py0 cb: %s-qos:%d-->>%s" % (recv_topic, recv_qos, recv_msg))
 
 
-ret = client.subscribe('topic_pikapy_qos0', 0, callback0)
+def callback1(signal):
-print("ret:%d" % ret)
+    recv_msg = client.getMsg(signal)
-ret = client.subscribe('topic_pikapy_qos1', 1,0)
+    recv_topic = client.getTopic(signal)
-print("ret:%d" % ret)
+    recv_qos = client.getQos(signal)
-ret = client.subscribe('topic_pikapy_qos2', 2,0)
+    print("py1 cb: %s-qos:%d-->>%s" % (recv_topic, recv_qos, recv_msg))
-print("ret:%d" % ret)
 
 
+def callback2(signal):
+    recv_msg = client.getMsg(signal)
+    recv_topic = client.getTopic(signal)
+    recv_qos = client.getQos(signal)
+    print("py2 cb: %s-qos:%d-->>%s" % (recv_topic, recv_qos, recv_msg))
+
+
+ret = client.subscribe('topic_pikapy_qos0', callback0, 0)
+print("subscribe ret:%d" % ret)
+ret = client.subscribe('topic_pikapy_qos1', callback1, 1)
+print("subscribe ret:%d" % ret)
+ret = client.subscribe('topic_pikapy_qos2', callback2, 2)
+print("subscribe ret:%d" % ret)
+
 # sleep wait for recv data
 T = PikaStdDevice.Time()
 T.sleep_s(5)
 
-out = client.listSubscribrTopic()
+out = client.listSubscribeTopic()
-print('out',out)
+print('listSubscribeTopic out', out)
 
 # client.unsubscribe('topic_pikapy_qos0');
 # client.unsubscribe('topic_pikapy_qos1');
 # client.unsubscribe('topic_pikapy_qos2');
 # T.sleep_s(5)
-# client.listSubscribrTopic()
+# out2 = client.listSubscribeTopic()
+# print('listSubscribeTopic out2',out2)
 
 
 # ret = client.setWill(1,'topic_will',1,'lost mqtt connect')
 T.sleep_s(10)
 # exit()
 ret = client.disconnect()
-print("ret:%d" % ret)
+print("disconnect ret:%d" % ret)