dtu_simple/app/System/source/sys_mqtt.c

/*
 * @Author: jiejie
 * @Github: https://github.com/jiejieTop
 * @Date: 2019-12-11 21:53:07
 * @LastEditTime : 2022-06-15 23:03:30
 * @Description: the code belongs to jiejie, please keep the author information and source code according to the license.
 */
#include <stdio.h>
#include <stdlib.h>
#include "mqtt_config.h"
#include "mqtt_log.h"
#include "sys_mqtt.h"
#include "gateway_message.h"
#include "data_task.h"
#include "netconf.h"

int mqtt_connectFlag;
static void mqtt_publish_task(void *arg);

/*
 *接收并处理mqtt订阅消息
 */
static void topic1_handler(void *client, message_data_t *msg)
{
	(void)client;
	MQTT_LOG_I("topic: %s\nmessage:%s", msg->topic_name, (char *)msg->message->payload);
	// json 解析数据
	
	write_modbus_data((char*)msg->message->payload);
}
/*
 *初始化mqtt连接
 */
int8_t mqtt_init(mqtt_client_t *client, char *clientId, char *user_name, char *password, char *ip, char* port, uint16_t keepAlive)
{
	mqtt_set_client_id(client, clientId);
	mqtt_set_port(client, port);
	mqtt_set_host(client, ip);
	mqtt_set_user_name(client, user_name);
	mqtt_set_password(client, password);
	mqtt_set_clean_session(client, 1);
	mqtt_set_keep_alive_interval(client, keepAlive);
	return mqtt_connect(client);
}
// 发送数据队列句柄
QueueHandle_t xQueue1;
void mqtt_task_creat()
{
	mqtt_client_t *client = NULL; // 创建一个客户端
	mqtt_log_init();
	client = mqtt_lease();
	printf("\nwelcome to mqttclient test...\n");
	xQueue1 = xQueueCreate(10, sizeof(struct Pub_Queue *)); // 创建一个mqtt上传的队列
	xTaskCreate(mqtt_publish_task, "mqtt_publish_task", 1024, client, 4, NULL);
}

static void mqtt_publish_task(void *arg)
{
	while(dhcp_done!=1)
	{
		vTaskDelay(100);
	}
	int rc;
	mqtt_client_t *client = (mqtt_client_t *)arg;

	while (rc < 0)
	{
	  rc = mqtt_init(client, "client_id_002", NULL, NULL, "36.134.23.11", "1883", 120);
		LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"NET TEST");
		vTaskDelay(1000);	
	}
	mqtt_connectFlag = 1;
//	vPortFree(port);
	mqtt_subscribe(client, "sub_topic_task", QOS0, topic1_handler);
	mqtt_message_t msg;
	memset(&msg, 0, sizeof(msg));

	mqtt_list_subscribe_topic(client);
	struct Pub_Queue *pxMessage;
	while (1)
	{
		if (xQueue1 != 0)
		{
			if (xQueueReceive(xQueue1, &(pxMessage), (TickType_t)10))
			{
				// 处理队列内部数据并上传
				msg.payloadlen = pxMessage->pubLength;
				msg.qos = pxMessage->qos;
				msg.payload = (void *)pxMessage->message;
				mqtt_publish(client, pxMessage->pub_topic, &msg);
		
				
				memset(pxMessage->message,0, pxMessage->pubLength);
				memset(pxMessage->pub_topic, 0,strlen(pxMessage->pub_topic));
				memset(pxMessage, 0, sizeof(struct Pub_Queue));
				vPortFree(pxMessage->message);
				vPortFree(pxMessage->pub_topic);
				vPortFree(pxMessage);
				pxMessage->message = NULL;
				pxMessage->pub_topic = NULL;
				pxMessage = NULL;
				
			}
		}
		vTaskDelay(100);
	}
}
/*
 *  函数名：void mqtt_publish_data(uint8_t *payload,mqtt_qos_t qos,uint16_t pub_length,char *topic)
 *  输入参数：payload上传的数据包，qos以什么等级去发布数据，pub_length上传的数据包长度，topic上传的topic
 *  输出参数：无
 *  返回值：无
 *  函数作用：向队列中写入mqtt上传的消息
 *  TODO：队列满无法写入的情况处理
 */
void mqtt_publish_data(char *payload, mqtt_qos_t qos, uint16_t pub_length, char *topic)
{

	struct Pub_Queue *pxMessage = pvPortMalloc(sizeof(struct Pub_Queue));
	pxMessage->message = pvPortMalloc(pub_length + 1);
	memset(pxMessage->message, 0, (pub_length + 1));
	memcpy(pxMessage->message, payload, pub_length);

	pxMessage->pub_topic = pvPortMalloc(strlen(topic) + 1);
	memset(pxMessage->pub_topic, 0, (strlen(topic) + 1));
//	strcpy(pxMessage->pub_topic, topic);
	pxMessage->pub_topic = "/device/DTtest0003";
	pxMessage->pubLength = pub_length;
	pxMessage->qos = qos;
	xQueueSend(xQueue1, (void *)&pxMessage, (TickType_t)0);
	
}