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 "sys_http.h"
#include "gateway_message.h"
#include "data_task.h"
#include "netconf.h"
#include "tcp_server.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);
	LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"Write Modbus Data");
	// 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", 512, client, 4, NULL);
}

static void mqtt_publish_task(void *arg)
{
		// 确保dhcp配置完成
	while(dhcp_done != 1)
	{
		vTaskDelay(100);
	}

	GATEWAY_PARAMS *get;
	get= get_gateway_config_params();	
	int rc;
	int msgId = 0;
	char port[6];  
	mqtt_client_t *client;
	
__MQTT_START:
	rc = -1;
	client = (mqtt_client_t *)arg;
	while (rc < 0)
	{
		vTaskDelay(1000);
		sprintf(port, "%hu", get->port);
	  rc = mqtt_init(client, "client_id_001", (char*)&get->username,(char*)&get->passwd,(char*)&get->host,port, 120);				
	}
	LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"MQTT Connect success");
	mqtt_connectFlag = 1;
	if(mqtt_subscribe(client, (char*)&get->commandTopic, QOS0, topic1_handler) != 0)
	{
		LogPrint(LOG_WARN,__FILE__,__FUNCTION__,__LINE__,"MQTT Subscribe fail");
	}
	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;
				if(mqtt_publish(client, pxMessage->pub_topic, &msg) != 0)// 0：成功
				{
					LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"MQTT Publish Failed");
					mqtt_disconnect(client);// 关闭mqtt连接
					goto __MQTT_START;
				}
				LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"MQTT Publish Message No:%d",ID++);
				vPortFree(pxMessage->pub_topic);
				vPortFree(pxMessage->message);
				vPortFree(pxMessage);
			}
		}
		vTaskDelay(1000);
	}
}
/*
 *  函数名：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->pubLength = pub_length;
	pxMessage->qos = qos;
	
	if(pxMessage->message == NULL)
		LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"queue message malloc fail");
	
	xQueueSend(xQueue1, (void *)&pxMessage, (TickType_t)0);
}