sys_mqtt.c 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. /*
  2. * @Author: jiejie
  3. * @Github: https://github.com/jiejieTop
  4. * @Date: 2019-12-11 21:53:07
  5. * @LastEditTime : 2022-06-15 23:03:30
  6. * @Description: the code belongs to jiejie, please keep the author information and source code according to the license.
  7. */
  8. #include <stdio.h>
  9. #include <stdlib.h>
  10. #include "mqtt_config.h"
  11. #include "mqtt_log.h"
  12. #include "sys_mqtt.h"
  13. #include "sys_http.h"
  14. #include "gateway_message.h"
  15. #include "data_task.h"
  16. #include "netconf.h"
  17. #include "tcp_server.h"
  18. int mqtt_connectFlag;
  19. static void mqtt_publish_task(void *arg);
  20. /*
  21. *接收并处理mqtt订阅消息
  22. */
  23. static void topic1_handler(void *client, message_data_t *msg)
  24. {
  25. (void)client;
  26. MQTT_LOG_I("topic: %s\nmessage:%s", msg->topic_name, (char *)msg->message->payload);
  27. LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"Write Modbus Data");
  28. // json 解析数据
  29. write_modbus_data((char*)msg->message->payload);
  30. }
  31. /*
  32. *初始化mqtt连接
  33. */
  34. int8_t mqtt_init(mqtt_client_t *client, char *clientId, char *user_name, char *password, char *ip, char* port, uint16_t keepAlive)
  35. {
  36. mqtt_set_client_id(client, clientId);
  37. mqtt_set_port(client, port);
  38. mqtt_set_host(client, ip);
  39. mqtt_set_user_name(client, user_name);
  40. mqtt_set_password(client, password);
  41. mqtt_set_clean_session(client, 1);
  42. mqtt_set_keep_alive_interval(client, keepAlive);
  43. return mqtt_connect(client);
  44. }
  45. // 发送数据队列句柄
  46. QueueHandle_t xQueue1;
  47. void mqtt_task_creat()
  48. {
  49. mqtt_client_t *client = NULL; // 创建一个客户端
  50. mqtt_log_init();
  51. client = mqtt_lease();
  52. printf("\nwelcome to mqttclient test...\n");
  53. xQueue1 = xQueueCreate(10, sizeof(struct Pub_Queue *)); // 创建一个mqtt上传的队列
  54. xTaskCreate(mqtt_publish_task, "mqtt_publish_task", 512, client, 4, NULL);
  55. }
  56. static void mqtt_publish_task(void *arg)
  57. {
  58. // 确保dhcp配置完成
  59. while(dhcp_done != 1)
  60. {
  61. vTaskDelay(100);
  62. }
  63. GATEWAY_PARAMS *get;
  64. get= get_gateway_config_params();
  65. int rc;
  66. int msgId = 0;
  67. char port[6];
  68. mqtt_client_t *client;
  69. __MQTT_START:
  70. rc = -1;
  71. client = (mqtt_client_t *)arg;
  72. while (rc < 0)
  73. {
  74. vTaskDelay(1000);
  75. sprintf(port, "%hu", get->port);
  76. rc = mqtt_init(client, "client_id_001", (char*)&get->username,(char*)&get->passwd,(char*)&get->host,port, 120);
  77. }
  78. LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"MQTT Connect success");
  79. mqtt_connectFlag = 1;
  80. if(mqtt_subscribe(client, (char*)&get->commandTopic, QOS0, topic1_handler) != 0)
  81. {
  82. LogPrint(LOG_WARN,__FILE__,__FUNCTION__,__LINE__,"MQTT Subscribe fail");
  83. }
  84. mqtt_message_t msg;
  85. memset(&msg, 0, sizeof(msg));
  86. mqtt_list_subscribe_topic(client);
  87. struct Pub_Queue *pxMessage;
  88. while (1)
  89. {
  90. if (xQueue1 != 0)
  91. {
  92. if (xQueueReceive(xQueue1, &(pxMessage), (TickType_t)10))
  93. {
  94. // 处理队列内部数据并上传
  95. msg.payloadlen = pxMessage->pubLength;
  96. msg.qos = pxMessage->qos;
  97. msg.payload = (void *)pxMessage->message;
  98. if(mqtt_publish(client, pxMessage->pub_topic, &msg) != 0)// 0:成功
  99. {
  100. LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"MQTT Publish Failed");
  101. mqtt_disconnect(client);// 关闭mqtt连接
  102. goto __MQTT_START;
  103. }
  104. LogPrint(LOG_INFO,__FILE__,__FUNCTION__,__LINE__,"MQTT Publish Message No:%d",ID++);
  105. vPortFree(pxMessage->pub_topic);
  106. vPortFree(pxMessage->message);
  107. vPortFree(pxMessage);
  108. }
  109. }
  110. vTaskDelay(1000);
  111. }
  112. }
  113. /*
  114. * 函数名:void mqtt_publish_data(uint8_t *payload,mqtt_qos_t qos,uint16_t pub_length,char *topic)
  115. * 输入参数:payload上传的数据包,qos以什么等级去发布数据,pub_length上传的数据包长度,topic上传的topic
  116. * 输出参数:无
  117. * 返回值:无
  118. * 函数作用:向队列中写入mqtt上传的消息
  119. * TODO:队列满无法写入的情况处理
  120. */
  121. void mqtt_publish_data(char *payload, mqtt_qos_t qos, uint16_t pub_length, char *topic)
  122. {
  123. struct Pub_Queue *pxMessage = pvPortMalloc(sizeof(struct Pub_Queue));
  124. pxMessage->message = pvPortMalloc(pub_length + 1);
  125. memset(pxMessage->message, 0, (pub_length + 1));
  126. memcpy(pxMessage->message, payload, pub_length);
  127. pxMessage->pub_topic = pvPortMalloc(strlen(topic) + 1);
  128. memset(pxMessage->pub_topic, 0, (strlen(topic) + 1));
  129. strcpy(pxMessage->pub_topic, topic);
  130. pxMessage->pubLength = pub_length;
  131. pxMessage->qos = qos;
  132. if(pxMessage->message == NULL)
  133. LogPrint(LOG_ERROR,__FILE__,__FUNCTION__,__LINE__,"queue message malloc fail");
  134. xQueueSend(xQueue1, (void *)&pxMessage, (TickType_t)0);
  135. }