1 /*
2 * 这个例程演示了用SDK配置MQTT参数并建立连接, 之后创建2个线程
3 *
4 * + 一个线程用于保活长连接
5 * + 一个线程用于接收消息, 并在有消息到达时进入默认的数据回调, 在连接状态变化时进入事件回调
6 *
7 * 接着演示了在MQTT连接上进行属性上报, 事件上报, 以及处理收到的属性设置, 服务调用, 取消这些代码段落的注释即可观察运行效果
8 *
9 * 需要用户关注或修改的部分, 已经用 TODO 在注释中标明
10 *
11 */
12
13 #include <stdio.h>
14 #include <string.h>
15 #include <unistd.h>
16 #include <aos/kernel.h>
17
18 #include "aiot_state_api.h"
19 #include "aiot_sysdep_api.h"
20 #include "aiot_mqtt_api.h"
21 #include "aiot_dm_api.h"
22 #include "flower_app.h"
23
24 /* 位于portfiles/aiot_port文件夹下的系统适配函数集合 */
25 extern aiot_sysdep_portfile_t g_aiot_sysdep_portfile;
26
27 /* 位于external/ali_ca_cert.c中的服务器证书 */
28 extern const char *ali_ca_cert;
29
30
31 static uint8_t g_mqtt_process_thread_running = 0;
32 static uint8_t g_mqtt_recv_thread_running = 0;
33 uint8_t mqtt_status = 0;
34 void *g_dm_handle = NULL;
35
36 /* TODO: 如果要关闭日志, 就把这个函数实现为空, 如果要减少日志, 可根据code选择不打印
37 *
38 * 例如: [1577589489.033][LK-0317] mqtt_basic_demo&a13FN5TplKq
39 *
40 * 上面这条日志的code就是0317(十六进制), code值的定义见core/aiot_state_api.h
41 *
42 */
43
44 /* 日志回调函数, SDK的日志会从这里输出 */
demo_state_logcb(int32_t code,char * message)45 int32_t demo_state_logcb(int32_t code, char *message)
46 {
47 printf("%s", message);
48 return 0;
49 }
50
51 /* MQTT事件回调函数, 当网络连接/重连/断开时被触发, 事件定义见core/aiot_mqtt_api.h */
demo_mqtt_event_handler(void * handle,const aiot_mqtt_event_t * event,void * userdata)52 void demo_mqtt_event_handler(void *handle, const aiot_mqtt_event_t *event, void *userdata)
53 {
54 switch (event->type) {
55 /* SDK因为用户调用了aiot_mqtt_connect()接口, 与mqtt服务器建立连接已成功 */
56 case AIOT_MQTTEVT_CONNECT: {
57 printf("AIOT_MQTTEVT_CONNECT\n");
58 mqtt_status = 1;
59 }
60 break;
61
62 /* SDK因为网络状况被动断连后, 自动发起重连已成功 */
63 case AIOT_MQTTEVT_RECONNECT: {
64 printf("AIOT_MQTTEVT_RECONNECT\n");
65 mqtt_status = 1;
66 }
67 break;
68
69 /* SDK因为网络的状况而被动断开了连接, network是底层读写失败, heartbeat是没有按预期得到服务端心跳应答 */
70 case AIOT_MQTTEVT_DISCONNECT: {
71 char *cause = (event->data.disconnect == AIOT_MQTTDISCONNEVT_NETWORK_DISCONNECT) ? ("network disconnect") :
72 ("heartbeat disconnect");
73 printf("AIOT_MQTTEVT_DISCONNECT: %s\n", cause);
74 mqtt_status = 0;
75 }
76 break;
77
78 default: {
79
80 }
81 }
82 }
83
84 /* 执行aiot_mqtt_process的线程, 包含心跳发送和QoS1消息重发 */
demo_mqtt_process_thread(void * args)85 void *demo_mqtt_process_thread(void *args)
86 {
87 int32_t res = STATE_SUCCESS;
88
89 while (g_mqtt_process_thread_running) {
90 res = aiot_mqtt_process(args);
91 if (res == STATE_USER_INPUT_EXEC_DISABLED) {
92 break;
93 }
94 aos_msleep(1000);
95 }
96 return NULL;
97 }
98
99 /* 执行aiot_mqtt_recv的线程, 包含网络自动重连和从服务器收取MQTT消息 */
demo_mqtt_recv_thread(void * args)100 void *demo_mqtt_recv_thread(void *args)
101 {
102 int32_t res = STATE_SUCCESS;
103
104 while (g_mqtt_recv_thread_running) {
105 res = aiot_mqtt_recv(args);
106 if (res < STATE_SUCCESS) {
107 if (res == STATE_USER_INPUT_EXEC_DISABLED) {
108 break;
109 }
110 aos_msleep(1000);
111 }
112 }
113 return NULL;
114 }
115
116 /* 用户数据接收处理回调函数 */
demo_dm_recv_handler(void * dm_handle,const aiot_dm_recv_t * recv,void * userdata)117 static void demo_dm_recv_handler(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata)
118 {
119 printf("demo_dm_recv_handler, type = %d\r\n", recv->type);
120
121 switch (recv->type) {
122
123 /* 属性上报, 事件上报, 获取期望属性值或者删除期望属性值的应答 */
124 case AIOT_DMRECV_GENERIC_REPLY: {
125 printf("msg_id = %d, code = %d, data = %.*s, message = %.*s\r\n",
126 recv->data.generic_reply.msg_id,
127 recv->data.generic_reply.code,
128 recv->data.generic_reply.data_len,
129 recv->data.generic_reply.data,
130 recv->data.generic_reply.message_len,
131 recv->data.generic_reply.message);
132 }
133 break;
134
135 /* 属性设置 */
136 case AIOT_DMRECV_PROPERTY_SET: {
137 printf("msg_id = %ld, params = %.*s\r\n",
138 (unsigned long)recv->data.property_set.msg_id,
139 recv->data.property_set.params_len,
140 recv->data.property_set.params);
141
142 /* TODO: 以下代码演示如何对来自云平台的属性设置指令进行应答, 用户可取消注释查看演示效果 */
143 /*
144 {
145 aiot_dm_msg_t msg;
146
147 memset(&msg, 0, sizeof(aiot_dm_msg_t));
148 msg.type = AIOT_DMMSG_PROPERTY_SET_REPLY;
149 msg.data.property_set_reply.msg_id = recv->data.property_set.msg_id;
150 msg.data.property_set_reply.code = 200;
151 msg.data.property_set_reply.data = "{}";
152 int32_t res = aiot_dm_send(dm_handle, &msg);
153 if (res < 0) {
154 printf("aiot_dm_send failed\r\n");
155 }
156 }
157 */
158 }
159 break;
160
161 /* 异步服务调用 */
162 case AIOT_DMRECV_ASYNC_SERVICE_INVOKE: {
163 printf("msg_id = %ld, service_id = %s, params = %.*s\r\n",
164 (unsigned long)recv->data.async_service_invoke.msg_id,
165 recv->data.async_service_invoke.service_id,
166 recv->data.async_service_invoke.params_len,
167 recv->data.async_service_invoke.params);
168
169 /* TODO: 以下代码演示如何对来自云平台的异步服务调用进行应答, 用户可取消注释查看演示效果
170 *
171 * 注意: 如果用户在回调函数外进行应答, 需要自行保存msg_id, 因为回调函数入参在退出回调函数后将被SDK销毁, 不可以再访问到
172 */
173
174 /*
175 {
176 aiot_dm_msg_t msg;
177
178 memset(&msg, 0, sizeof(aiot_dm_msg_t));
179 msg.type = AIOT_DMMSG_ASYNC_SERVICE_REPLY;
180 msg.data.async_service_reply.msg_id = recv->data.async_service_invoke.msg_id;
181 msg.data.async_service_reply.code = 200;
182 msg.data.async_service_reply.service_id = "ToggleLightSwitch";
183 msg.data.async_service_reply.data = "{\"dataA\": 20}";
184 int32_t res = aiot_dm_send(dm_handle, &msg);
185 if (res < 0) {
186 printf("aiot_dm_send failed\r\n");
187 }
188 }
189 */
190 }
191 break;
192
193 /* 同步服务调用 */
194 case AIOT_DMRECV_SYNC_SERVICE_INVOKE: {
195 printf("msg_id = %ld, rrpc_id = %s, service_id = %s, params = %.*s\r\n",
196 (unsigned long)recv->data.sync_service_invoke.msg_id,
197 recv->data.sync_service_invoke.rrpc_id,
198 recv->data.sync_service_invoke.service_id,
199 recv->data.sync_service_invoke.params_len,
200 recv->data.sync_service_invoke.params);
201
202 /* TODO: 以下代码演示如何对来自云平台的同步服务调用进行应答, 用户可取消注释查看演示效果
203 *
204 * 注意: 如果用户在回调函数外进行应答, 需要自行保存msg_id和rrpc_id字符串, 因为回调函数入参在退出回调函数后将被SDK销毁, 不可以再访问到
205 */
206
207 /*
208 {
209 aiot_dm_msg_t msg;
210
211 memset(&msg, 0, sizeof(aiot_dm_msg_t));
212 msg.type = AIOT_DMMSG_SYNC_SERVICE_REPLY;
213 msg.data.sync_service_reply.rrpc_id = recv->data.sync_service_invoke.rrpc_id;
214 msg.data.sync_service_reply.msg_id = recv->data.sync_service_invoke.msg_id;
215 msg.data.sync_service_reply.code = 200;
216 msg.data.sync_service_reply.service_id = "SetLightSwitchTimer";
217 msg.data.sync_service_reply.data = "{}";
218 int32_t res = aiot_dm_send(dm_handle, &msg);
219 if (res < 0) {
220 printf("aiot_dm_send failed\r\n");
221 }
222 }
223 */
224 }
225 break;
226
227 /* 下行二进制数据 */
228 case AIOT_DMRECV_RAW_DATA: {
229 printf("raw data len = %d\r\n", recv->data.raw_data.data_len);
230 /* TODO: 以下代码演示如何发送二进制格式数据, 若使用需要有相应的数据透传脚本部署在云端 */
231 /*
232 {
233 aiot_dm_msg_t msg;
234 uint8_t raw_data[] = {0x01, 0x02};
235
236 memset(&msg, 0, sizeof(aiot_dm_msg_t));
237 msg.type = AIOT_DMMSG_RAW_DATA;
238 msg.data.raw_data.data = raw_data;
239 msg.data.raw_data.data_len = sizeof(raw_data);
240 aiot_dm_send(dm_handle, &msg);
241 }
242 */
243 }
244 break;
245
246 /* 二进制格式的同步服务调用, 比单纯的二进制数据消息多了个rrpc_id */
247 case AIOT_DMRECV_RAW_SYNC_SERVICE_INVOKE: {
248 printf("raw sync service rrpc_id = %s, data_len = %d\r\n",
249 recv->data.raw_service_invoke.rrpc_id,
250 recv->data.raw_service_invoke.data_len);
251 }
252 break;
253
254 default:
255 break;
256 }
257 }
258
259 /* 属性上报函数演示 */
demo_send_property_post(void * dm_handle,char * params)260 int32_t demo_send_property_post(void *dm_handle, char *params)
261 {
262 aiot_dm_msg_t msg;
263
264 memset(&msg, 0, sizeof(aiot_dm_msg_t));
265 msg.type = AIOT_DMMSG_PROPERTY_POST;
266 msg.data.property_post.params = params;
267
268 return aiot_dm_send(dm_handle, &msg);
269 }
270
271 /* 事件上报函数演示 */
demo_send_event_post(void * dm_handle,char * event_id,char * params)272 int32_t demo_send_event_post(void *dm_handle, char *event_id, char *params)
273 {
274 aiot_dm_msg_t msg;
275
276 memset(&msg, 0, sizeof(aiot_dm_msg_t));
277 msg.type = AIOT_DMMSG_EVENT_POST;
278 msg.data.event_post.event_id = event_id;
279 msg.data.event_post.params = params;
280
281 return aiot_dm_send(dm_handle, &msg);
282 }
283
284 /* 演示了获取属性LightSwitch的期望值, 用户可将此函数加入到main函数中运行演示 */
demo_send_get_desred_requset(void * dm_handle)285 int32_t demo_send_get_desred_requset(void *dm_handle)
286 {
287 aiot_dm_msg_t msg;
288
289 memset(&msg, 0, sizeof(aiot_dm_msg_t));
290 msg.type = AIOT_DMMSG_GET_DESIRED;
291 msg.data.get_desired.params = "[\"LightSwitch\"]";
292
293 return aiot_dm_send(dm_handle, &msg);
294 }
295
296 /* 演示了删除属性LightSwitch的期望值, 用户可将此函数加入到main函数中运行演示 */
demo_send_delete_desred_requset(void * dm_handle)297 int32_t demo_send_delete_desred_requset(void *dm_handle)
298 {
299 aiot_dm_msg_t msg;
300
301 memset(&msg, 0, sizeof(aiot_dm_msg_t));
302 msg.type = AIOT_DMMSG_DELETE_DESIRED;
303 msg.data.get_desired.params = "{\"LightSwitch\":{}}";
304
305 return aiot_dm_send(dm_handle, &msg);
306 }
307
308
demo_main(int argc,char * argv[])309 int demo_main(int argc, char *argv[])
310 {
311 int32_t res = STATE_SUCCESS;
312 void *dm_handle = NULL;
313 void *mqtt_handle = NULL;
314 char *url = "iot-as-mqtt.cn-shanghai.aliyuncs.com"; /* 阿里云平台上海站点的域名后缀 */
315 char host[100] = {0}; /* 用这个数组拼接设备连接的云平台站点全地址, 规则是 ${productKey}.iot-as-mqtt.cn-shanghai.aliyuncs.com */
316 uint16_t port = 443; /* 无论设备是否使用TLS连接阿里云平台, 目的端口都是443 */
317 aiot_sysdep_network_cred_t cred; /* 安全凭据结构体, 如果要用TLS, 这个结构体中配置CA证书等参数 */
318
319 /* TODO: 替换为自己设备的三元组 */
320 char *product_key = "产品key";
321 char *device_name = "设备名";
322 char *device_secret = "设备密钥";
323
324 /* 配置SDK的底层依赖 */
325 aiot_sysdep_set_portfile(&g_aiot_sysdep_portfile);
326 /* 配置SDK的日志输出 */
327 aiot_state_set_logcb(demo_state_logcb);
328
329 /* 创建SDK的安全凭据, 用于建立TLS连接 */
330 memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
331 cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA; /* 使用RSA证书校验MQTT服务端 */
332 cred.max_tls_fragment = 16384; /* 最大的分片长度为16K, 其它可选值还有4K, 2K, 1K, 0.5K */
333 cred.sni_enabled = 1; /* TLS建连时, 支持Server Name Indicator */
334 cred.x509_server_cert = ali_ca_cert; /* 用来验证MQTT服务端的RSA根证书 */
335 cred.x509_server_cert_len = strlen(ali_ca_cert); /* 用来验证MQTT服务端的RSA根证书长度 */
336
337 /* 创建1个MQTT客户端实例并内部初始化默认参数 */
338 mqtt_handle = aiot_mqtt_init();
339 if (mqtt_handle == NULL) {
340 printf("aiot_mqtt_init failed\n");
341 return -1;
342 }
343
344 snprintf(host, 100, "%s.%s", product_key, url);
345 /* 配置MQTT服务器地址 */
346 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_HOST, (void *)host);
347 /* 配置MQTT服务器端口 */
348 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PORT, (void *)&port);
349 /* 配置设备productKey */
350 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PRODUCT_KEY, (void *)product_key);
351 /* 配置设备deviceName */
352 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_NAME, (void *)device_name);
353 /* 配置设备deviceSecret */
354 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_SECRET, (void *)device_secret);
355 /* 配置网络连接的安全凭据, 上面已经创建好了 */
356 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_NETWORK_CRED, (void *)&cred);
357 /* 配置MQTT事件回调函数 */
358 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_EVENT_HANDLER, (void *)demo_mqtt_event_handler);
359
360 /* 创建DATA-MODEL实例 */
361 dm_handle = aiot_dm_init();
362 g_dm_handle = dm_handle;
363 if (dm_handle == NULL) {
364 printf("aiot_dm_init failed");
365 return -1;
366 }
367 /* 配置MQTT实例句柄 */
368 aiot_dm_setopt(dm_handle, AIOT_DMOPT_MQTT_HANDLE, mqtt_handle);
369 /* 配置消息接收处理回调函数 */
370 aiot_dm_setopt(dm_handle, AIOT_DMOPT_RECV_HANDLER, (void *)demo_dm_recv_handler);
371
372 /* 与服务器建立MQTT连接 */
373 res = aiot_mqtt_connect(mqtt_handle);
374 if (res < STATE_SUCCESS) {
375 /* 尝试建立连接失败, 销毁MQTT实例, 回收资源 */
376 aiot_mqtt_deinit(&mqtt_handle);
377 printf("aiot_mqtt_connect failed: -0x%04X\n", -res);
378 return -1;
379 }
380
381 /* 创建一个单独的线程, 专用于执行aiot_mqtt_process, 它会自动发送心跳保活, 以及重发QoS1的未应答报文 */
382 g_mqtt_process_thread_running = 1;
383 res = aos_task_new("demo_mqtt_process", demo_mqtt_process_thread, mqtt_handle, 4096);
384 //res = pthread_create(&g_mqtt_process_thread, NULL, demo_mqtt_process_thread, mqtt_handle);
385 if (res != 0) {
386 printf("create demo_mqtt_process_thread failed: %d\n", res);
387 return -1;
388 }
389
390 /* 创建一个单独的线程用于执行aiot_mqtt_recv, 它会循环收取服务器下发的MQTT消息, 并在断线时自动重连 */
391 g_mqtt_recv_thread_running = 1;
392 res = aos_task_new("demo_mqtt_process", demo_mqtt_recv_thread, mqtt_handle, 4096);
393 //res = pthread_create(&g_mqtt_recv_thread, NULL, demo_mqtt_recv_thread, mqtt_handle);
394 if (res != 0) {
395 printf("create demo_mqtt_recv_thread failed: %d\n", res);
396 return -1;
397 }
398
399 /* 主循环进入休眠 */
400 while (1) {
401 report_2_cloud(dm_handle);
402 aos_msleep(3000);
403 }
404
405 /* 断开MQTT连接, 一般不会运行到这里 */
406 res = aiot_mqtt_disconnect(mqtt_handle);
407 if (res < STATE_SUCCESS) {
408 aiot_mqtt_deinit(&mqtt_handle);
409 printf("aiot_mqtt_disconnect failed: -0x%04X\n", -res);
410 return -1;
411 }
412
413 /* 销毁DATA-MODEL实例, 一般不会运行到这里 */
414 res = aiot_dm_deinit(&dm_handle);
415 if (res < STATE_SUCCESS) {
416 printf("aiot_dm_deinit failed: -0x%04X\n", -res);
417 return -1;
418 }
419
420 /* 销毁MQTT实例, 一般不会运行到这里 */
421 res = aiot_mqtt_deinit(&mqtt_handle);
422 if (res < STATE_SUCCESS) {
423 printf("aiot_mqtt_deinit failed: -0x%04X\n", -res);
424 return -1;
425 }
426
427 g_mqtt_process_thread_running = 0;
428 g_mqtt_recv_thread_running = 0;
429
430 return 0;
431 }
432
433