add:mqtt特殊应用回调

components/network/libemqtt/luat_lib_mqtt.c

@@ -332,6 +332,7 @@ static int l_mqtt_create(lua_State *L) {
 		LLOGE("out of memory when malloc mqtt_ctrl");
 		return 0;
 	}
+	mqtt_ctrl->app_cb = NULL;
 
 	ret = luat_mqtt_init(mqtt_ctrl, adapter_index);
 	if (ret) {

components/network/libemqtt/luat_mqtt.h

@@ -45,6 +45,7 @@ typedef struct{
 	uint32_t rxbuff_size; 		/**< mqtt_packet_buffer的长度*/
 	uint8_t *mqtt_packet_buffer;/**< 接收BUFF*/
 	void* mqtt_cb;			/**< mqtt 回调函数*/
+	void *app_cb;				/**< mqtt 特殊应用回调，数据收发不再会调用到lua层*/
 	int8_t error_state;    		/**< mqtt 错误状态*/
 	uint16_t remote_port; 		/**< 远程端口号*/
 	uint32_t keepalive;   		/**< 心跳时长 单位s*/

components/network/libemqtt/luat_mqtt_client.c

@@ -27,12 +27,23 @@ int32_t luatos_mqtt_callback(lua_State *L, void* ptr);
 
 int l_luat_mqtt_msg_cb(luat_mqtt_ctrl_t * ptr, int arg1, int arg2) {
 #ifdef __LUATOS__
+	luat_mqtt_ctrl_t *mqtt_ctrl =(luat_mqtt_ctrl_t *)ptr;
 	rtos_msg_t msg = {
 		.handler = luatos_mqtt_callback,
 		.ptr = ptr,
 		.arg1 = arg1,
 		.arg2 = arg2
 	};
+
+	if (mqtt_ctrl->app_cb)
+	{
+		luat_mqtt_cb_t mqtt_cb = mqtt_ctrl->app_cb;
+		mqtt_cb(mqtt_ctrl, arg1);
+		if (MQTT_MSG_PUBLISH == arg1)
+		{
+			return 0;
+		}
+	}
 	luat_msgbus_put(&msg, 0);
 #else
 	luat_mqtt_ctrl_t *mqtt_ctrl =(luat_mqtt_ctrl_t *)ptr;
@@ -320,13 +331,20 @@ static int luat_mqtt_msg_cb(luat_mqtt_ctrl_t *mqtt_ctrl) {
 			LLOGD("MQTT_MSG_PUBLISH");
 			qos = MQTTParseMessageQos(mqtt_ctrl->mqtt_packet_buffer);
 #ifdef __LUATOS__
-			const uint8_t* ptr;
-			uint16_t topic_len = mqtt_parse_pub_topic_ptr(mqtt_ctrl->mqtt_packet_buffer, &ptr);
-			uint32_t payload_len = mqtt_parse_pub_msg_ptr(mqtt_ctrl->mqtt_packet_buffer, &ptr);
-			luat_mqtt_msg_t *mqtt_msg = (luat_mqtt_msg_t *)luat_heap_malloc(sizeof(luat_mqtt_msg_t)+topic_len+payload_len);
-			mqtt_msg->topic_len = mqtt_parse_pub_topic(mqtt_ctrl->mqtt_packet_buffer, mqtt_msg->data);
-            mqtt_msg->payload_len = mqtt_parse_publish_msg(mqtt_ctrl->mqtt_packet_buffer, mqtt_msg->data+topic_len);
-			l_luat_mqtt_msg_cb(mqtt_ctrl, MQTT_MSG_PUBLISH, (int)mqtt_msg);
+			if (!mqtt_ctrl->app_cb)
+			{
+				const uint8_t* ptr;
+				uint16_t topic_len = mqtt_parse_pub_topic_ptr(mqtt_ctrl->mqtt_packet_buffer, &ptr);
+				uint32_t payload_len = mqtt_parse_pub_msg_ptr(mqtt_ctrl->mqtt_packet_buffer, &ptr);
+				luat_mqtt_msg_t *mqtt_msg = (luat_mqtt_msg_t *)luat_heap_malloc(sizeof(luat_mqtt_msg_t)+topic_len+payload_len);
+				mqtt_msg->topic_len = mqtt_parse_pub_topic(mqtt_ctrl->mqtt_packet_buffer, mqtt_msg->data);
+	            mqtt_msg->payload_len = mqtt_parse_publish_msg(mqtt_ctrl->mqtt_packet_buffer, mqtt_msg->data+topic_len);
+				l_luat_mqtt_msg_cb(mqtt_ctrl, MQTT_MSG_PUBLISH, (int)mqtt_msg);
+			}
+			else
+			{
+				l_luat_mqtt_msg_cb(mqtt_ctrl, MQTT_MSG_PUBLISH, 0);
+			}
 #else
 			l_luat_mqtt_msg_cb(mqtt_ctrl, MQTT_MSG_PUBLISH, 0);
 #endif