【建筑施工监测与安防系统】十一、Kaluga MQTT连接OneNET

sonicfirr

【建筑施工监测与安防系统】十一、Kaluga MQTT连接OneNET [复制链接]

 

        参考ESP32 IDF的案例“mqtt/tcp”（..\esp-idf-v4.4\examples\protocols\mqtt\tcp）引入MQTT Client功能，使得Kaluga板可以连接OneNET（本人采用MQTT旧版，非加密）。

1、官方mqtt/tcp案例说明

        IDF中有关联网及协议应用的案例大多依靠另一个联网案例实现Wi-Fi接入，这个案例就是“..\esp-idf-v4.4\examples\common_components\protocol_examples_common”。所以在移植MQTT案例到自己的项目时，要注意自己做好Wi-Fi接入。

 

图11-1 mqtt/tcp案例的主要函数

 

2、mqtt移植到自己项目

        这里不介绍如何实现Wi-Fi接入了，不清楚的朋友可以去看本人的第七篇帖子。将mqtt/tcp的源码拷贝到自己的项目中，当然相关函数都要进行修改。首先是原案例需要导入的头文件：

 

#include <stdio.h>
#include <stdint.h>
#include <stddef.h>
#include <string.h>
#include "esp_wifi.h"
#include "esp_system.h"
#include "nvs_flash.h"
#include "esp_event.h"
#include "esp_netif.h"
#include "protocol_examples_common.h"

#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/semphr.h"
#include "freertos/queue.h"

#include "lwip/sockets.h"
#include "lwip/dns.h"
#include "lwip/netdb.h"

#include "esp_log.h"
#include "mqtt_client.h"

 

        然后就是回调函数，下面代码本人加了注释的两个地方，是在下觉得需要关注的点，一个是断联后原案例只是给个控制台输出，这里加上重连函数调用。另一个是接收到OneNET发送命令的处理分支，后续可以在这里加入命令解析和处理代码。

 

// 原案例封装的一个输出错误信息的函数，在事件回调中被调用
void log_error_if_nonzero(const char *message, int error_code) {
    if(error_code != 0) {
        ESP_LOGE(TAG, "Last error %s: 0x%x", message, error_code);
    }
}

void mqtt_event_handler(void *handler_args, 
    esp_event_base_t base, int32_t event_id, void *event_data) {
    ESP_LOGD(TAG, "Event dispatched from event loop base=%s, event_id=%d", base, event_id);
    esp_mqtt_event_handle_t event = event_data;
    esp_mqtt_client_handle_t client = event->client;
    int msg_id;
    switch ((esp_mqtt_event_id_t)event_id) {
    case MQTT_EVENT_CONNECTED:
        ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");
        msg_id = esp_mqtt_client_publish(client, "/topic/qos1", "data_3", 0, 1, 0);
        ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);

        msg_id = esp_mqtt_client_subscribe(client, "/topic/qos0", 0);
        ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);

        msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);
        ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);

        msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");
        ESP_LOGI(TAG, "sent unsubscribe successful, msg_id=%d", msg_id);
        break;
    case MQTT_EVENT_DISCONNECTED:
        ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");
        esp_mqtt_client_reconnect(client); //by author. reconnect
        break;

    case MQTT_EVENT_SUBSCRIBED:
        ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);
        msg_id = esp_mqtt_client_publish(client, "/topic/qos0", "data", 0, 0, 0);
        ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);
        break;
    case MQTT_EVENT_UNSUBSCRIBED:
        ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);
        break;
    case MQTT_EVENT_PUBLISHED:
        ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);
        break;
    case MQTT_EVENT_DATA:
        ESP_LOGI(TAG, "MQTT_EVENT_DATA"); //by author. OneNET order
        printf("TOPIC=%.*s\r\n", event->topic_len, event->topic);
        printf("DATA=%.*s\r\n", event->data_len, event->data);
        break;
    case MQTT_EVENT_ERROR:
        ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
        if (event->error_handle->error_type == MQTT_ERROR_TYPE_TCP_TRANSPORT) {
            log_error_if_nonzero("reported from esp-tls", event->error_handle->esp_tls_last_esp_err);
            log_error_if_nonzero("reported from tls stack", event->error_handle->esp_tls_stack_err);
            log_error_if_nonzero("captured as transport's socket errno",  event->error_handle->esp_transport_sock_errno);
            ESP_LOGI(TAG, "Last errno string (%s)", strerror(event->error_handle->esp_transport_sock_errno));
        }
        break;
    default:
        ESP_LOGI(TAG, "Other event id:%d", event->event_id);
        break;
    }
}

 

        然后是原案例的接入函数mqtt_app_start()这里做了修改——原例也没法直接用，缺少必要的配置（如：client_id，username等）。MQTT相关操作（连接、发布等）都需要一个“句柄”（esp_mqtt_client_handle_t类型），原案例是定义在函数中的局部量，因为事件回调可以得到这个句柄变量的传参所以无所谓。但是要单独编写发布数据包函数，就需要将句柄定义为全局量了。

 

esp_mqtt_client_handle_t client;
void aita_StartMqtt(void) {
    esp_mqtt_client_config_t mqtt_cfg = {
        .host = "183.230.40.39", // OneNET MQTT服务器IP
        .port = 6002,             // OneNET MQTT服务器端口
        .client_id = "deviceid",// 设备ID，请替换成自己的
        .username = "userid",    // 用户ID，请替换成自己的
        .password = "authorization", // 鉴权信息，请替换成自己的
    };

    client = esp_mqtt_client_init(&mqtt_cfg);
    /* The last argument may be used to pass data to the event handler, in this example mqtt_event_handler */
    esp_mqtt_client_register_event(client, ESP_EVENT_ANY_ID, mqtt_event_handler, NULL);
    esp_mqtt_client_start(client);
}

 

         再编写一个发送函数，注意data也就是发布数据包的载荷（payload）需要按照OneNET平台的格式要求就行封装，不清楚的朋友可以查看本人的ESP32 C3测评帖子。

 

void aita_PubMqtt(const char *data, int len) {
    esp_mqtt_client_publish(client, "$dp", data, len, 0, 0);
}

 

        最后，结合本人的上篇帖子，在定时器回调中调用aita_PubMqtt()，暂时还是功能测试，就将定时器的启用时间发送到OneNET平台上。

 

static void periodic_timer_callback(void* arg) {
    int64_t time_since_boot = esp_timer_get_time();
    char dat[80];
    sprintf(dat, "{\"tick\":%d}", (int)time_since_boot);
    char buf[83];
    int len = strlen(dat);
    buf[0] = 0x03;
    buf[1] = len>>8;
    buf[2] = len&0xff;
    memcpy(buf+3, dat, len);
    buf[3+len] = 0;
    aita_PubMqtt(buf, 3+len);
    // ESP_LOGI(TAG, "periodic timer called, time since boot: %lld us", 
    //                 time_since_boot);
    // aita_InitSNTP();
}

 

图11-2 OneNET收到数据的展示