业务场景
1、子设备与子设备、网关与子设备之间通过业务协议进行通信（ZigBee、wifi、蓝牙等）
2、网关设备连接物联网平台，创建物理通道，并代理子设备上线创建逻辑通道
3、子设备通过逻辑通道与物联网平台进行通信，但是实际不建立任何物理上的连接。
原理介绍
1、子设备上报物模型，首先得满足网关-子设备架构，先要创建网关和子设备产品
关于网关产品和子设备的产品创建这里就不过多赘述。物模型上报，首先肯定得有物模型的定义吧。
2、子设备和网关要建立一种关联关系，建立网关与子设备的topo结构。(这样网关和平台才知道，谁是这个网关的子设备)
（1）可以直接在控制台上进行操作
（2）可以通过云端API进行操作
（3）可以通过SDK的接口进行操作
C-SDK 4.x 接口
components/subdev/aiot_subdev_api.h
3、网关设备创建与物联网平台的链路，再代理子设备进行上线(注意：这些操作都是在网关上完成)
网关设备如同普通直连设备一样，与物联网平台建立连接,即网关-子设备架构中的物理通道
代理子设备上线，帮助子设备与物联网平台建立逻辑通道
components/subdev/aiot_subdev_api.h
4、子设备通过网关与物联网平台的物理链路上报消息到物联网平台（注意：这些操作都是在网关上完成，子设备只需要提供三元组信息）
*不管是MQTT API还是物模型API，他们的底层原理都是一样的，网关设备和物联网平台进行直连，存在唯一的一条物理通道，而子设备通过网关代理上线，使用逻辑通道与平台进行通信，复用的是网关的物理通道
*
官网说明：
可以看到这里有两种方式，分别是基于MQTT API的接口
和基于物模型API的接口。
那么aiot_mqtt_pub和aiot_dm_send又是如何体现
物理通道和逻辑通道呢？
(1) MQTT API aiot_mqtt_pub()接口方式
core/aiot_mqtt_api.h
直接调用aiot_mqtt_pub传入带子设备信息的Topic
(2) 物模型API aiot_dm_send()接口方式
在aiot_dm_send()的两个参数中handle使用网关代理上线后的那个handle，msg携带子设备的信息，完成逻辑通道的消息传输。
components/data-model/aiot_dm_api.h
这个接口里面优先使用的是msg里面携带的product key和devicename来组装topic
这个msg结果如下aiot_dm_msg_t
操作步骤
1、创建网关产品和子设备的产品&&设备，并建立Topo结构，这里使用控制台方式方便快捷（此篇文章主要介绍子设备的物模型上报）
子设备产品定义一个测试用的物模型属性
2、打开C-SDK 4.x Demos
demos/subdev_basic_demo.c
(1) 修改网关设备三元组信息以及url
在demo_mqtt_start()函数中
(2)修改子设备的三元组信息
全局变量aiot_subdev_dev_t g_subdev[]
(3)添加demo_subdev_send_property_post接口，上报子设备物模型数据
增加头文件包含
编写demo_subdev_send_property_post接口
(4) 在main函数中，子设备代理上线成功之后，添加子设备上报物模型逻辑，调用demo_subdev_send_property_post接口
(5)在main函数中，在 物模型API上报子设备的物模型逻辑之后，再添加一个通过MQTT API接口上报物模型的逻辑
(6)在C-SDK 4.x主目录
make
cd output
./subdev-basic-demo
我们可以看到三个子设备的物模型数据都上报成功
温度分别是0.9 、1.9、 2.9
音量分别是88、89、90
控制台中查看物模型数据：
子设备1
子设备2
子设备3
修改后的Demo代码源码：demos/subdev_basic_demo.c
/*
* 这个例程适用于`Linux`这类支持pthread的POSIX设备, 它演示了......
*
* 需要用户关注或修改的部分, 已经用 TODO 在注释中标明
*
*/
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include "aiot_state_api.h"
#include "aiot_sysdep_api.h"
#include "aiot_mqtt_api.h"
#include "aiot_subdev_api.h"
#include "aiot_dm_api.h"
/* 位于portfiles/aiot_port文件夹下的系统适配函数集合 */
extern aiot_sysdep_portfile_t g_aiot_sysdep_portfile;
/* 位于external/ali_ca_cert.c中的服务器证书 */
extern const char *ali_ca_cert;
static pthread_t g_mqtt_process_thread;
static pthread_t g_mqtt_recv_thread;
static uint8_t g_mqtt_process_thread_running = 0;
static uint8_t g_mqtt_recv_thread_running = 0;
//TODO：修改子设备三元组信息
aiot_subdev_dev_t g_subdev[] = {
{
"a1H****0mgG",
"IoT******bDemo1",
"e98c465******6dcb35cfa75",
"prGp****OjLS7GI"
},
{
"a1H*****mgG",
"IoTDe*****Demo2",
"92d719*****67f65bfe4e4d895",
"prG*****jLS7GI"
},
{
"a1H9wUf0mgG",
"IoTDeviceSubDemo3",
"b2d20a9bb7dbe34e97a4c56f47aed6ee",
"prG*****LS7GI"
}
/*
{
"a13FN5TplKq",
"subdev_basic_demo_01",
"768XBgQwgOakz3K4uhOiLeeh9xjJQx6h",
"y7GSILD480lBSsP8"
},
{
"a13FN5TplKq",
"subdev_basic_demo_02",
"iwTZrbjbgNVChfuJkihjE5asekoyKoYv",
"y7GSILD480lBSsP8"
},
{
"a13FN5TplKq",
"subdev_basic_demo_03",
"fdutq35iKMYdcWWBuIINY26hsNhgFXWE",
"y7GSILD480lBSsP8"
},
{
"a13FN5TplKq",
"subdev_basic_demo_04",
"HCKv50YqgwdKhy5cE0Vz4aydmK2ojPvr",
"y7GSILD480lBSsP8"
}
*/
};
/* TODO: 如果要关闭日志, 就把这个函数实现为空, 如果要减少日志, 可根据code选择不打印
*
* 例如: [1577589489.033][LK-0317] subdev_basic_demo&a13FN5TplKq
*
* 上面这条日志的code就是0317(十六进制), code值的定义见core/aiot_state_api.h
*
*/
/* 日志回调函数, SDK的日志会从这里输出 */
static int32_t demo_state_logcb(int32_t code, char *message)
{
printf("%s", message);
return 0;
}
/* MQTT事件回调函数, 当网络连接/重连/断开时被触发, 事件定义见core/aiot_mqtt_api.h */
void demo_mqtt_event_handler(void *handle, const aiot_mqtt_event_t *event, void *userdata)
{
switch (event->type) {
/* SDK因为用户调用了aiot_mqtt_connect()接口, 与mqtt服务器建立连接已成功 */
case AIOT_MQTTEVT_CONNECT: {
printf("AIOT_MQTTEVT_CONNECT\n");
/* TODO: 处理SDK建连成功, 不可以在这里调用耗时较长的阻塞函数 */
}
break;
/* SDK因为网络状况被动断连后, 自动发起重连已成功 */
case AIOT_MQTTEVT_RECONNECT: {
printf("AIOT_MQTTEVT_RECONNECT\n");
/* TODO: 处理SDK重连成功, 不可以在这里调用耗时较长的阻塞函数 */
}
break;
/* SDK因为网络的状况而被动断开了连接, network是底层读写失败, heartbeat是没有按预期得到服务端心跳应答 */
case AIOT_MQTTEVT_DISCONNECT: {
char *cause = (event->data.disconnect == AIOT_MQTTDISCONNEVT_NETWORK_DISCONNECT) ? ("network disconnect") :
("heartbeat disconnect");
printf("AIOT_MQTTEVT_DISCONNECT: %s\n", cause);
/* TODO: 处理SDK被动断连, 不可以在这里调用耗时较长的阻塞函数 */
}
break;
default: {
}
}
}
/* MQTT默认消息处理回调, 当SDK从服务器收到MQTT消息时, 且无对应用户回调处理时被调用 */
void demo_mqtt_default_recv_handler(void *handle, const aiot_mqtt_recv_t *packet, void *userdata)
{
switch (packet->type) {
case AIOT_MQTTRECV_HEARTBEAT_RESPONSE: {
printf("heartbeat response\n");
/* TODO: 处理服务器对心跳的回应, 一般不处理 */
}
break;
case AIOT_MQTTRECV_SUB_ACK: {
printf("suback, res: -0x%04X, packet id: %d, max qos: %d\n",
-packet->data.sub_ack.res, packet->data.sub_ack.packet_id, packet->data.sub_ack.max_qos);
/* TODO: 处理服务器对订阅请求的回应, 一般不处理 */
}
break;
case AIOT_MQTTRECV_PUB: {
printf("pub, qos: %d, topic: %.*s\n", packet->data.pub.qos, packet->data.pub.topic_len, packet->data.pub.topic);
printf("pub, payload: %.*s\n", packet->data.pub.payload_len, packet->data.pub.payload);
/* TODO: 处理服务器下发的业务报文 */
}
break;
case AIOT_MQTTRECV_PUB_ACK: {
printf("puback, packet id: %d\n", packet->data.pub_ack.packet_id);
/* TODO: 处理服务器对QoS1上报消息的回应, 一般不处理 */
}
break;
default: {
}
}
}
/* 执行aiot_mqtt_process的线程, 包含心跳发送和QoS1消息重发 */
void *demo_mqtt_process_thread(void *args)
{
int32_t res = STATE_SUCCESS;
while (g_mqtt_process_thread_running) {
res = aiot_mqtt_process(args);
if (res == STATE_USER_INPUT_EXEC_DISABLED) {
break;
}
sleep(1);
}
return NULL;
}
/* 执行aiot_mqtt_recv的线程, 包含网络自动重连和从服务器收取MQTT消息 */
void *demo_mqtt_recv_thread(void *args)
{
int32_t res = STATE_SUCCESS;
while (g_mqtt_recv_thread_running) {
res = aiot_mqtt_recv(args);
if (res < STATE_SUCCESS) {
if (res == STATE_USER_INPUT_EXEC_DISABLED) {
break;
}
sleep(1);
}
}
return NULL;
}
int32_t demo_mqtt_start(void **handle)
{
int32_t res = STATE_SUCCESS;
void *mqtt_handle = NULL;
char *url = "iot-as-mqtt.cn-shanghai.aliyuncs.com"; /* 阿里云平台上海站点的域名后缀 */
char host[100] = {0}; /* 用这个数组拼接设备连接的云平台站点全地址, 规则是 ${productKey}.iot-as-mqtt.cn-shanghai.aliyuncs.com */
uint16_t port = 443; /* 无论设备是否使用TLS连接阿里云平台, 目的端口都是443 */
aiot_sysdep_network_cred_t cred; /* 安全凭据结构体, 如果要用TLS, 这个结构体中配置CA证书等参数 */
/* TODO: 替换为自己设备的三元组 */
char *product_key = "a1L****pp";
char *device_name = "Io*****Demo1";
char *device_secret = "68b10******a4b052a197538";
/* 创建SDK的安全凭据, 用于建立TLS连接 */
memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA; /* 使用RSA证书校验MQTT服务端 */
cred.max_tls_fragment = 16384; /* 最大的分片长度为16K, 其它可选值还有4K, 2K, 1K, 0.5K */
cred.sni_enabled = 1; /* TLS建连时, 支持Server Name Indicator */
cred.x509_server_cert = ali_ca_cert; /* 用来验证MQTT服务端的RSA根证书 */
cred.x509_server_cert_len = strlen(ali_ca_cert); /* 用来验证MQTT服务端的RSA根证书长度 */
/* 创建1个MQTT客户端实例并内部初始化默认参数 */
mqtt_handle = aiot_mqtt_init();
if (mqtt_handle == NULL) {
printf("aiot_mqtt_init failed\n");
return -1;
}
/* TODO: 如果以下代码不被注释, 则例程会用TCP而不是TLS连接云平台 */
/*
{
memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
cred.option = AIOT_SYSDEP_NETWORK_CRED_NONE;
}
*/
snprintf(host, 100, "%s.%s", product_key, url);
/* 配置MQTT服务器地址 */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_HOST, (void *)host);
/* 配置MQTT服务器端口 */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PORT, (void *)&port);
/* 配置设备productKey */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PRODUCT_KEY, (void *)product_key);
/* 配置设备deviceName */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_NAME, (void *)device_name);
/* 配置设备deviceSecret */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_SECRET, (void *)device_secret);
/* 配置网络连接的安全凭据, 上面已经创建好了 */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_NETWORK_CRED, (void *)&cred);
/* 配置MQTT默认消息接收回调函数 */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_RECV_HANDLER, (void *)demo_mqtt_default_recv_handler);
/* 配置MQTT事件回调函数 */
aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_EVENT_HANDLER, (void *)demo_mqtt_event_handler);
/* 与服务器建立MQTT连接 */
res = aiot_mqtt_connect(mqtt_handle);
if (res < STATE_SUCCESS) {
/* 尝试建立连接失败, 销毁MQTT实例, 回收资源 */
aiot_mqtt_deinit(&mqtt_handle);
printf("aiot_mqtt_connect failed: -0x%04X\n", -res);
return -1;
}
/* 创建一个单独的线程, 专用于执行aiot_mqtt_process, 它会自动发送心跳保活, 以及重发QoS1的未应答报文 */
g_mqtt_process_thread_running = 1;
res = pthread_create(&g_mqtt_process_thread, NULL, demo_mqtt_process_thread, mqtt_handle);
if (res < 0) {
printf("pthread_create demo_mqtt_process_thread failed: %d\n", res);
aiot_mqtt_deinit(&mqtt_handle);
return -1;
}
/* 创建一个单独的线程用于执行aiot_mqtt_recv, 它会循环收取服务器下发的MQTT消息, 并在断线时自动重连 */
g_mqtt_recv_thread_running = 1;
res = pthread_create(&g_mqtt_recv_thread, NULL, demo_mqtt_recv_thread, mqtt_handle);
if (res < 0) {
printf("pthread_create demo_mqtt_recv_thread failed: %d\n", res);
g_mqtt_process_thread_running = 0;
pthread_join(g_mqtt_process_thread, NULL);
aiot_mqtt_deinit(&mqtt_handle);
return -1;
}
*handle = mqtt_handle;
return 0;
}
int32_t demo_mqtt_stop(void **handle)
{
int32_t res = STATE_SUCCESS;
void *mqtt_handle = NULL;
mqtt_handle = *handle;
g_mqtt_process_thread_running = 0;
g_mqtt_recv_thread_running = 0;
pthread_join(g_mqtt_process_thread, NULL);
pthread_join(g_mqtt_recv_thread, NULL);
/* 断开MQTT连接 */
res = aiot_mqtt_disconnect(mqtt_handle);
if (res < STATE_SUCCESS) {
aiot_mqtt_deinit(&mqtt_handle);
printf("aiot_mqtt_disconnect failed: -0x%04X\n", -res);
return -1;
}
/* 销毁MQTT实例 */
res = aiot_mqtt_deinit(&mqtt_handle);
if (res < STATE_SUCCESS) {
printf("aiot_mqtt_deinit failed: -0x%04X\n", -res);
return -1;
}
return 0;
}
void demo_subdev_recv_handler(void *handle, const aiot_subdev_recv_t *packet, void *user_data)
{
switch (packet->type) {
case AIOT_SUBDEVRECV_TOPO_ADD_REPLY:
case AIOT_SUBDEVRECV_TOPO_DELETE_REPLY:
case AIOT_SUBDEVRECV_TOPO_GET_REPLY:
case AIOT_SUBDEVRECV_BATCH_LOGIN_REPLY:
case AIOT_SUBDEVRECV_BATCH_LOGOUT_REPLY:
case AIOT_SUBDEVRECV_SUB_REGISTER_REPLY:
case AIOT_SUBDEVRECV_PRODUCT_REGISTER_REPLY: {
printf("msgid : %d\n", packet->data.generic_reply.msg_id);
printf("code : %d\n", packet->data.generic_reply.code);
printf("product key : %s\n", packet->data.generic_reply.product_key);
printf("device name : %s\n", packet->data.generic_reply.device_name);
printf("message : %s\n", (packet->data.generic_reply.message == NULL)?("NULL"):(packet->data.generic_reply.message));
printf("data : %s\n", packet->data.generic_reply.data);
}
break;
case AIOT_SUBDEVRECV_TOPO_CHANGE_NOTIFY: {
printf("msgid : %d\n", packet->data.generic_notify.msg_id);
printf("product key : %s\n", packet->data.generic_notify.product_key);
printf("device name : %s\n", packet->data.generic_notify.device_name);
printf("params : %s\n", packet->data.generic_notify.params);
}
break;
default: {
}
}
}
/* 子设备物模型属性上报函数演示 */
int32_t demo_subdev_send_property_post(void *dm_handle, char *params,
char *subdev_product_key,char *subdev_device_name)
{
aiot_dm_msg_t msg;
memset(&msg, 0, sizeof(aiot_dm_msg_t));
msg.type = AIOT_DMMSG_PROPERTY_POST;
msg.data.property_post.params = params;
msg.product_key = subdev_product_key;
msg.device_name = subdev_device_name;
return aiot_dm_send(dm_handle, &msg);
}
int main(int argc, char *argv[])
{
int32_t res = STATE_SUCCESS;
void *mqtt_handle = NULL, *subdev_handle = NULL;
/* 配置SDK的底层依赖 */
aiot_sysdep_set_portfile(&g_aiot_sysdep_portfile);
/* 配置SDK的日志输出 */
aiot_state_set_logcb(demo_state_logcb);
res = demo_mqtt_start(&mqtt_handle);
if (res < 0) {
printf("demo_mqtt_start failed\n");
return -1;
}
subdev_handle = aiot_subdev_init();
if (subdev_handle == NULL) {
printf("aiot_subdev_init failed\n");
demo_mqtt_stop(&mqtt_handle);
return -1;
}
aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_MQTT_HANDLE, mqtt_handle);
aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_RECV_HANDLER, demo_subdev_recv_handler);
res = aiot_subdev_send_topo_add(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t));
if (res < STATE_SUCCESS) {
printf("aiot_subdev_send_topo_add failed, res: -0x%04X\n", -res);
aiot_subdev_deinit(&subdev_handle);
demo_mqtt_stop(&mqtt_handle);
return -1;
}
sleep(2);
// aiot_subdev_send_topo_delete(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t));
// if (res < STATE_SUCCESS) {
// printf("aiot_subdev_send_topo_delete failed, res: -0x%04X\n", -res);
// aiot_subdev_deinit(&subdev_handle);
// demo_mqtt_stop(&mqtt_handle);
// return -1;
// }
// sleep(2);
// aiot_subdev_send_topo_get(subdev_handle);
// if (res < STATE_SUCCESS) {
// printf("aiot_subdev_send_topo_get failed, res: -0x%04X\n", -res);
// aiot_subdev_deinit(&subdev_handle);
// demo_mqtt_stop(&mqtt_handle);
// return -1;
// }
// sleep(2);
// aiot_subdev_send_sub_register(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t));
// if (res < STATE_SUCCESS) {
// printf("aiot_subdev_send_sub_register failed, res: -0x%04X\n", -res);
// aiot_subdev_deinit(&subdev_handle);
// demo_mqtt_stop(&mqtt_handle);
// return -1;
// }
// sleep(2);
// aiot_subdev_send_product_register(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t));
// if (res < STATE_SUCCESS) {
// printf("aiot_subdev_send_product_register failed, res: -0x%04X\n", -res);
// aiot_subdev_deinit(&subdev_handle);
// demo_mqtt_stop(&mqtt_handle);
// return -1;
// }
// sleep(2);
aiot_subdev_send_batch_login(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t));
if (res < STATE_SUCCESS) {
printf("aiot_subdev_send_batch_login failed, res: -0x%04X\n", -res);
aiot_subdev_deinit(&subdev_handle);
demo_mqtt_stop(&mqtt_handle);
return -1;
}
sleep(2);
// aiot_subdev_send_batch_logout(subdev_handle, g_subdev, sizeof(g_subdev)/sizeof(aiot_subdev_dev_t));
// if (res < STATE_SUCCESS) {
// printf("aiot_subdev_send_batch_logout failed, res: -0x%04X\n", -res);
// aiot_subdev_deinit(&subdev_handle);
// demo_mqtt_stop(&mqtt_handle);
// return -1;
// }
/* 子设备上报物模型属性 基于物模型API接口*/
for(int i = 0; i < sizeof(g_subdev)/sizeof(aiot_subdev_dev_t) ; i++) {
/* ***********************************************************************************
* 替换为子设备的ProductKey 和 子设备的deviceName */
char *product_key = g_subdev[i].product_key;
char *device_name = g_subdev[i].device_name;
static float cnt = 0.9;
char property_payload[30] = {0};
snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": %f}", cnt++);
//HAL_Snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": %d,\"time\":1603248304}}", cnt++);
//snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": 77.77,\"time\":1603248304000}}");
demo_subdev_send_property_post(subdev_handle,property_payload,product_key,device_name);
//demo_subdev_send_property_post(mqtt_handle,property_payload,product_key,device_name);
/*********************************************************************************************/
sleep(1);
}
/* 子设备上报物模型属性 基于MQTT API接口*/
for(int i = 0; i < sizeof(g_subdev)/sizeof(aiot_subdev_dev_t) ; i++) {
/* ***********************************************************************************
* 替换为子设备的ProductKey 和 子设备的deviceName */
char *product_key2 = g_subdev[i].product_key;
char *device_name2 = g_subdev[i].device_name;
static int cnt2 = 88;
char property_payload2[30] = {0};
snprintf(property_payload2, sizeof(property_payload2), "{\"params\":{\"Volume\": %d}}", cnt2++);
//HAL_Snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": %d,\"time\":1603248304}}", cnt++);
//snprintf(property_payload, sizeof(property_payload), "{\"Temperature\": {\"value\": 77.77,\"time\":1603248304000}}");
char pub_topic[128] = {0};
snprintf(pub_topic,sizeof(pub_topic),"/sys/%s/%s/thing/event/property/post",product_key2,device_name2);
res = aiot_mqtt_pub(mqtt_handle, pub_topic, (uint8_t *)property_payload2, (uint32_t)strlen(property_payload2), 0);
//res = aiot_mqtt_pub(subdev_handle, pub_topic, (uint8_t *)property_payload2, (uint32_t)strlen(property_payload2), 0);
if (res < 0) {
printf("aiot_mqtt_sub failed, res: -0x%04X\n", -res);
return -1;
}
/*********************************************************************************************/
sleep(1);
}
res = aiot_subdev_deinit(&subdev_handle);
if (res < STATE_SUCCESS) {
printf("aiot_subdev_deinit failed: -0x%04X\n", res);
}
res = demo_mqtt_stop(&mqtt_handle);
if (res < 0) {
printf("demo_start_stop failed\n");
return -1;
}
return 0;
}