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"


static void mqtt_publish_task(void *arg);
/*
*接收并处理mqtt订阅消息
*/
static void topic1_handler(void* client, message_data_t* msg)
{
    (void) client;
    MQTT_LOG_I("-----------------------------------------------------------------------------------");
    MQTT_LOG_I("%s:%d %s()...\ntopic: %s\nmessage:%s", __FILE__, __LINE__, __FUNCTION__, msg->topic_name, (char*)msg->message->payload);
    MQTT_LOG_I("-----------------------------------------------------------------------------------");
}
/*
*初始化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, 2, NULL);
}

static void mqtt_publish_task(void *arg)
{
    mqtt_client_t *client = (mqtt_client_t *)arg;
		while(1)
		{
			  int rc=mqtt_init(client,"client_id_001",NULL,NULL,"36.134.23.11","1883",1000);
				vTaskDelay(100);
		}
		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);
						
						vPortFree(pxMessage->message);
						vPortFree(pxMessage->pub_topic);
						vPortFree(pxMessage);
					}
				}
				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(uint8_t *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);

	xQueueSend( xQueue1, (void *)&pxMessage, ( TickType_t ) 0 );
}