123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104 |
- #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);
- 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);
- }
- 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 *));
- xTaskCreate(mqtt_publish_task, "mqtt_publish_task", 1024, client, 4, NULL);
- }
- static void mqtt_publish_task(void *arg)
- {
- mqtt_client_t *client = (mqtt_client_t *)arg;
- while (1)
- {
- LOG_PRINT(LOG_INFO,"mqtt_start");
- 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)
- {
- 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);
- }
|