DTU_4GLAN_HAL/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 "app_ethernet.h"
#include "tcp_server.h"
#include "main.h"
#include "usart.h"


int mqtt_connectFlag;
static void mqtt_publish_task(void const *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 解析数据
	if(ProtocolsModeFlag)
	{
		write_modbus_data((char*)msg->message->payload);
	}
	else if(TransparentModeFlag)
	{
		USART_485_Send(msg->message->payload, msg->message->payloadlen);
	}
}
/*
 *初始化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上传的队列
	osThreadDef(MQTT_task, mqtt_publish_task, osPriorityNormal, 0, configMINIMAL_STACK_SIZE * 8);
  osThreadCreate (osThread(MQTT_task), client);
}

static void mqtt_publish_task(void const *arg)
{	
	GATEWAY_PARAMS *get;
	get= get_gateway_config_params();	
	int rc;
	char port[6];
	mqtt_client_t *client;

__MQTT_START:
	rc = -1;
	mqtt_connectFlag = 0;
	client = (mqtt_client_t *)arg;	
	while (rc != 0)
	{
		vTaskDelay(500);
		sprintf(port, "%hu", get->port);
	  rc = mqtt_init(client, "client_id_002", (char*)&get->username,(char*)&get->passwd,(char*)&get->host,port, 120);				
//	  rc = mqtt_init(client, "client_id_002", "device","Dev&*(789","36.134.23.11","1883", 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");
					myfree(SRAMEX, pxMessage->pub_topic);
					myfree(SRAMEX, pxMessage->message);
					myfree(SRAMEX, pxMessage);
					mqtt_unsubscribe(client,(char*)&get->commandTopic);
					mqtt_disconnect(client);
					platform_thread_destroy(client->mqtt_thread);
					client->mqtt_thread = NULL;
					goto __MQTT_START;
				}
				LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"MQTT Publish Message");
				myfree(SRAMEX, pxMessage->pub_topic);
				myfree(SRAMEX, pxMessage->message);
				myfree(SRAMEX, pxMessage);
			}
		}
		vTaskDelay(500);
	}	
}
/*
 *  函数名：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：队列满无法写入的情况处理
 */
uint8_t size;
void mqtt_publish_data(char *payload, mqtt_qos_t qos, uint16_t pub_length, char *topic)
{
	size = my_mem_perused(SRAMEX);//9
	struct Pub_Queue *pxMessage = mymalloc(SRAMEX, sizeof(struct Pub_Queue));
	if(pxMessage == NULL)
		LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"queue message malloc fail");
	
	size = my_mem_perused(SRAMEX);
	pxMessage->message = mymalloc(SRAMEX, pub_length + 1);
	if(pxMessage == NULL)
		LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"queue message malloc fail");
	
	memset(pxMessage->message, 0, (pub_length + 1));
	memcpy(pxMessage->message, payload, pub_length);
	
	size = my_mem_perused(SRAMEX);
	pxMessage->pub_topic = mymalloc(SRAMEX,strlen(topic) + 1);
	if(pxMessage->pub_topic == NULL)
		LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"queue message malloc fail");
	
	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);
}