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 "cJSON.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
34 extern int audio_install_codec_driver();
35 extern void sound_example_loopback_entry(int argc, char **argv);
36 extern void cmd_tts_handler(char *buf, int len, int argc, char **argv);
37 extern void cmd_play_handler(char *buf, int len, int argc, char **argv);
38
39 /* TODO: 如果要关闭日志, 就把这个函数实现为空, 如果要减少日志, 可根据code选择不打印
40 *
41 * 例如: [1577589489.033][LK-0317] mqtt_basic_demo&a13FN5TplKq
42 *
43 * 上面这条日志的code就是0317(十六进制), code值的定义见core/aiot_state_api.h
44 *
45 */
46
47 /* 日志回调函数, SDK的日志会从这里输出 */
demo_state_logcb(int32_t code,char * message)48 int32_t demo_state_logcb(int32_t code, char *message)
49 {
50 printf("%s", message);
51 return 0;
52 }
53
54 /* MQTT事件回调函数, 当网络连接/重连/断开时被触发, 事件定义见core/aiot_mqtt_api.h */
demo_mqtt_event_handler(void * handle,const aiot_mqtt_event_t * event,void * userdata)55 void demo_mqtt_event_handler(void *handle, const aiot_mqtt_event_t *event, void *userdata)
56 {
57 switch (event->type) {
58 /* SDK因为用户调用了aiot_mqtt_connect()接口, 与mqtt服务器建立连接已成功 */
59 case AIOT_MQTTEVT_CONNECT: {
60 printf("AIOT_MQTTEVT_CONNECT\n");
61 }
62 break;
63
64 /* SDK因为网络状况被动断连后, 自动发起重连已成功 */
65 case AIOT_MQTTEVT_RECONNECT: {
66 printf("AIOT_MQTTEVT_RECONNECT\n");
67 }
68 break;
69
70 /* SDK因为网络的状况而被动断开了连接, network是底层读写失败, heartbeat是没有按预期得到服务端心跳应答 */
71 case AIOT_MQTTEVT_DISCONNECT: {
72 char *cause = (event->data.disconnect == AIOT_MQTTDISCONNEVT_NETWORK_DISCONNECT) ? ("network disconnect") :
73 ("heartbeat disconnect");
74 printf("AIOT_MQTTEVT_DISCONNECT: %s\n", cause);
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 /* ------ 云端钉一体智能语音播放器指令解析 ------ */
143 cJSON *root = NULL, *item_payload = NULL;
144 static int pause_state = 0, power_state = 0, recorder_state = 0;
145 char *param_array[10];
146
147 /* Parse Root */
148 root = cJSON_Parse(recv->data.property_set.params);
149 if (root == NULL || !cJSON_IsObject(root)) {
150 printf("JSON Parse Error\r\n");
151 return -1;
152 }
153
154 /* Payload: PowerSwitch */
155 item_payload = cJSON_GetObjectItem(root, "PowerSwitch");
156 if (item_payload != NULL && cJSON_IsNumber(item_payload)) {
157 if(1 == item_payload->valueint) {
158 printf("PowerSwitch = %d \r\n", item_payload->valueint);
159 if(0 == power_state) {
160 printf("start load sound driver... \r\n");
161 power_state = 1;
162 audio_install_codec_driver();
163 } else {
164 printf("sound driver already loaded. \r\n");
165 }
166 } else {
167 printf("uninstall sound driver, TBD \r\n");
168 }
169 }
170
171 /* Payload: startRecord */
172 item_payload = cJSON_GetObjectItem(root, "startRecord");
173 if (item_payload != NULL && cJSON_IsNumber(item_payload)) {
174 if((1 == item_payload->valueint) && (0 == recorder_state)) {
175 printf("start recorder. \r\n");
176 recorder_state = 1;
177 param_array[0] = "sound_loopback";
178 param_array[1] = "start";
179 sound_example_loopback_entry(2, param_array);
180 } else {
181 printf("stop recorder. \r\n");
182 recorder_state = 0;
183 param_array[0] = "sound_loopback";
184 param_array[1] = "stop";
185 sound_example_loopback_entry(2, param_array);
186 }
187 }
188
189 /* Payload: TTSText */
190 item_payload = cJSON_GetObjectItem(root, "TTSText");
191 if (item_payload != NULL && cJSON_IsString(item_payload)) {
192 printf("tts %s /data/tts.mp3\r\n", item_payload->valuestring);
193 param_array[0] = "tts";
194 param_array[1] = item_payload->valuestring;
195 param_array[2] = "/data/tts.mp3";
196 cmd_tts_handler(NULL, 0, 3, param_array);
197 }
198
199 /* Payload: PlayURL */
200 item_payload = cJSON_GetObjectItem(root, "PlayURL");
201 if (item_payload != NULL && cJSON_IsString(item_payload)) {
202 printf("play %s\r\n", item_payload->valuestring);
203 param_array[0] = "play";
204 param_array[1] = item_payload->valuestring;
205 cmd_play_handler(NULL, 0, 2, param_array);
206 }
207
208 /* Payload: PlayType */
209 item_payload = cJSON_GetObjectItem(root, "PlayType");
210 if (item_payload != NULL && cJSON_IsNumber(item_payload)) {
211 printf("PlayType = %d \r\n", item_payload->valueint);
212 switch(item_payload->valueint) {
213 case 0: // 停止
214 printf("play stop \r\n");
215 param_array[0] = "play";
216 param_array[1] = "stop";
217 cmd_play_handler(NULL, 0, 2, param_array);
218 break;
219 case 1: // 恢复
220 if(pause_state) {
221 printf("play resume \r\n");
222 param_array[0] = "play";
223 param_array[1] = "resume";
224 cmd_play_handler(NULL, 0, 2, param_array);
225 }
226 pause_state = 0;
227 break;
228 case 2: //暂停
229 printf("play pause \r\n");
230 param_array[0] = "play";
231 param_array[1] = "pause";
232 pause_state = 1;
233 cmd_play_handler(NULL, 0, 2, param_array);
234 break;
235 }
236 }
237
238 /* Payload: SetVolume */
239 item_payload = cJSON_GetObjectItem(root, "SetVolume");
240 if (item_payload != NULL && cJSON_IsNumber(item_payload)) {
241 printf("set volume = %d \r\n", item_payload->valueint);
242 aos_set_master_volume(item_payload->valueint);
243 }
244 }
245 break;
246
247 /* 异步服务调用 */
248 case AIOT_DMRECV_ASYNC_SERVICE_INVOKE: {
249 printf("msg_id = %ld, service_id = %s, params = %.*s\r\n",
250 (unsigned long)recv->data.async_service_invoke.msg_id,
251 recv->data.async_service_invoke.service_id,
252 recv->data.async_service_invoke.params_len,
253 recv->data.async_service_invoke.params);
254
255 /* TODO: 以下代码演示如何对来自云平台的异步服务调用进行应答, 用户可取消注释查看演示效果
256 *
257 * 注意: 如果用户在回调函数外进行应答, 需要自行保存msg_id, 因为回调函数入参在退出回调函数后将被SDK销毁, 不可以再访问到
258 */
259
260 /*
261 {
262 aiot_dm_msg_t msg;
263
264 memset(&msg, 0, sizeof(aiot_dm_msg_t));
265 msg.type = AIOT_DMMSG_ASYNC_SERVICE_REPLY;
266 msg.data.async_service_reply.msg_id = recv->data.async_service_invoke.msg_id;
267 msg.data.async_service_reply.code = 200;
268 msg.data.async_service_reply.service_id = "ToggleLightSwitch";
269 msg.data.async_service_reply.data = "{\"dataA\": 20}";
270 int32_t res = aiot_dm_send(dm_handle, &msg);
271 if (res < 0) {
272 printf("aiot_dm_send failed\r\n");
273 }
274 }
275 */
276 }
277 break;
278
279 /* 同步服务调用 */
280 case AIOT_DMRECV_SYNC_SERVICE_INVOKE: {
281 printf("msg_id = %ld, rrpc_id = %s, service_id = %s, params = %.*s\r\n",
282 (unsigned long)recv->data.sync_service_invoke.msg_id,
283 recv->data.sync_service_invoke.rrpc_id,
284 recv->data.sync_service_invoke.service_id,
285 recv->data.sync_service_invoke.params_len,
286 recv->data.sync_service_invoke.params);
287
288 /* TODO: 以下代码演示如何对来自云平台的同步服务调用进行应答, 用户可取消注释查看演示效果
289 *
290 * 注意: 如果用户在回调函数外进行应答, 需要自行保存msg_id和rrpc_id字符串, 因为回调函数入参在退出回调函数后将被SDK销毁, 不可以再访问到
291 */
292
293 /*
294 {
295 aiot_dm_msg_t msg;
296
297 memset(&msg, 0, sizeof(aiot_dm_msg_t));
298 msg.type = AIOT_DMMSG_SYNC_SERVICE_REPLY;
299 msg.data.sync_service_reply.rrpc_id = recv->data.sync_service_invoke.rrpc_id;
300 msg.data.sync_service_reply.msg_id = recv->data.sync_service_invoke.msg_id;
301 msg.data.sync_service_reply.code = 200;
302 msg.data.sync_service_reply.service_id = "SetLightSwitchTimer";
303 msg.data.sync_service_reply.data = "{}";
304 int32_t res = aiot_dm_send(dm_handle, &msg);
305 if (res < 0) {
306 printf("aiot_dm_send failed\r\n");
307 }
308 }
309 */
310 }
311 break;
312
313 /* 下行二进制数据 */
314 case AIOT_DMRECV_RAW_DATA: {
315 printf("raw data len = %d\r\n", recv->data.raw_data.data_len);
316 /* TODO: 以下代码演示如何发送二进制格式数据, 若使用需要有相应的数据透传脚本部署在云端 */
317 /*
318 {
319 aiot_dm_msg_t msg;
320 uint8_t raw_data[] = {0x01, 0x02};
321
322 memset(&msg, 0, sizeof(aiot_dm_msg_t));
323 msg.type = AIOT_DMMSG_RAW_DATA;
324 msg.data.raw_data.data = raw_data;
325 msg.data.raw_data.data_len = sizeof(raw_data);
326 aiot_dm_send(dm_handle, &msg);
327 }
328 */
329 }
330 break;
331
332 /* 二进制格式的同步服务调用, 比单纯的二进制数据消息多了个rrpc_id */
333 case AIOT_DMRECV_RAW_SYNC_SERVICE_INVOKE: {
334 printf("raw sync service rrpc_id = %s, data_len = %d\r\n",
335 recv->data.raw_service_invoke.rrpc_id,
336 recv->data.raw_service_invoke.data_len);
337 }
338 break;
339
340 default:
341 break;
342 }
343 }
344
345 /* 属性上报函数演示 */
demo_send_property_post(void * dm_handle,char * params)346 int32_t demo_send_property_post(void *dm_handle, char *params)
347 {
348 aiot_dm_msg_t msg;
349
350 memset(&msg, 0, sizeof(aiot_dm_msg_t));
351 msg.type = AIOT_DMMSG_PROPERTY_POST;
352 msg.data.property_post.params = params;
353
354 return aiot_dm_send(dm_handle, &msg);
355 }
356
357 /* 事件上报函数演示 */
demo_send_event_post(void * dm_handle,char * event_id,char * params)358 int32_t demo_send_event_post(void *dm_handle, char *event_id, char *params)
359 {
360 aiot_dm_msg_t msg;
361
362 memset(&msg, 0, sizeof(aiot_dm_msg_t));
363 msg.type = AIOT_DMMSG_EVENT_POST;
364 msg.data.event_post.event_id = event_id;
365 msg.data.event_post.params = params;
366
367 return aiot_dm_send(dm_handle, &msg);
368 }
369
370 /* 演示了获取属性LightSwitch的期望值, 用户可将此函数加入到main函数中运行演示 */
demo_send_get_desred_requset(void * dm_handle)371 int32_t demo_send_get_desred_requset(void *dm_handle)
372 {
373 aiot_dm_msg_t msg;
374
375 memset(&msg, 0, sizeof(aiot_dm_msg_t));
376 msg.type = AIOT_DMMSG_GET_DESIRED;
377 msg.data.get_desired.params = "[\"LightSwitch\"]";
378
379 return aiot_dm_send(dm_handle, &msg);
380 }
381
382 /* 演示了删除属性LightSwitch的期望值, 用户可将此函数加入到main函数中运行演示 */
demo_send_delete_desred_requset(void * dm_handle)383 int32_t demo_send_delete_desred_requset(void *dm_handle)
384 {
385 aiot_dm_msg_t msg;
386
387 memset(&msg, 0, sizeof(aiot_dm_msg_t));
388 msg.type = AIOT_DMMSG_DELETE_DESIRED;
389 msg.data.get_desired.params = "{\"LightSwitch\":{}}";
390
391 return aiot_dm_send(dm_handle, &msg);
392 }
393
394
demo_main(int argc,char * argv[])395 int demo_main(int argc, char *argv[])
396 {
397 int32_t res = STATE_SUCCESS;
398 void *dm_handle = NULL;
399 void *mqtt_handle = NULL;
400 char *url = "iot-as-mqtt.cn-shanghai.aliyuncs.com"; /* 阿里云平台上海站点的域名后缀 */
401 char host[100] = {0}; /* 用这个数组拼接设备连接的云平台站点全地址, 规则是 ${productKey}.iot-as-mqtt.cn-shanghai.aliyuncs.com */
402 uint16_t port = 443; /* 无论设备是否使用TLS连接阿里云平台, 目的端口都是443 */
403 aiot_sysdep_network_cred_t cred; /* 安全凭据结构体, 如果要用TLS, 这个结构体中配置CA证书等参数 */
404
405 /* TODO: 替换为自己设备的三元组 */
406 char *product_key = "";
407 char *device_name = "";
408 char *device_secret = "";
409
410 /* 配置SDK的底层依赖 */
411 aiot_sysdep_set_portfile(&g_aiot_sysdep_portfile);
412 /* 配置SDK的日志输出 */
413 aiot_state_set_logcb(demo_state_logcb);
414
415 /* 创建SDK的安全凭据, 用于建立TLS连接 */
416 memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
417 cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA; /* 使用RSA证书校验MQTT服务端 */
418 cred.max_tls_fragment = 16384; /* 最大的分片长度为16K, 其它可选值还有4K, 2K, 1K, 0.5K */
419 cred.sni_enabled = 1; /* TLS建连时, 支持Server Name Indicator */
420 cred.x509_server_cert = ali_ca_cert; /* 用来验证MQTT服务端的RSA根证书 */
421 cred.x509_server_cert_len = strlen(ali_ca_cert); /* 用来验证MQTT服务端的RSA根证书长度 */
422
423 /* 创建1个MQTT客户端实例并内部初始化默认参数 */
424 mqtt_handle = aiot_mqtt_init();
425 if (mqtt_handle == NULL) {
426 printf("aiot_mqtt_init failed\n");
427 return -1;
428 }
429
430 snprintf(host, 100, "%s.%s", product_key, url);
431 /* 配置MQTT服务器地址 */
432 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_HOST, (void *)host);
433 /* 配置MQTT服务器端口 */
434 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PORT, (void *)&port);
435 /* 配置设备productKey */
436 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PRODUCT_KEY, (void *)product_key);
437 /* 配置设备deviceName */
438 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_NAME, (void *)device_name);
439 /* 配置设备deviceSecret */
440 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_SECRET, (void *)device_secret);
441 /* 配置网络连接的安全凭据, 上面已经创建好了 */
442 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_NETWORK_CRED, (void *)&cred);
443 /* 配置MQTT事件回调函数 */
444 aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_EVENT_HANDLER, (void *)demo_mqtt_event_handler);
445
446 /* 创建DATA-MODEL实例 */
447 dm_handle = aiot_dm_init();
448 if (dm_handle == NULL) {
449 printf("aiot_dm_init failed");
450 return -1;
451 }
452 /* 配置MQTT实例句柄 */
453 aiot_dm_setopt(dm_handle, AIOT_DMOPT_MQTT_HANDLE, mqtt_handle);
454 /* 配置消息接收处理回调函数 */
455 aiot_dm_setopt(dm_handle, AIOT_DMOPT_RECV_HANDLER, (void *)demo_dm_recv_handler);
456
457 /* 与服务器建立MQTT连接 */
458 res = aiot_mqtt_connect(mqtt_handle);
459 if (res < STATE_SUCCESS) {
460 /* 尝试建立连接失败, 销毁MQTT实例, 回收资源 */
461 aiot_mqtt_deinit(&mqtt_handle);
462 printf("aiot_mqtt_connect failed: -0x%04X\n", -res);
463 return -1;
464 }
465
466 /* 创建一个单独的线程, 专用于执行aiot_mqtt_process, 它会自动发送心跳保活, 以及重发QoS1的未应答报文 */
467 g_mqtt_process_thread_running = 1;
468 res = aos_task_new("demo_mqtt_process", demo_mqtt_process_thread, mqtt_handle, 4096);
469 // res = pthread_create(&g_mqtt_process_thread, NULL, demo_mqtt_process_thread, mqtt_handle);
470 if (res != 0) {
471 printf("create demo_mqtt_process_thread failed: %d\n", res);
472 return -1;
473 }
474
475 /* 创建一个单独的线程用于执行aiot_mqtt_recv, 它会循环收取服务器下发的MQTT消息, 并在断线时自动重连 */
476 g_mqtt_recv_thread_running = 1;
477 res = aos_task_new("demo_mqtt_process", demo_mqtt_recv_thread, mqtt_handle, 4096);
478 // res = pthread_create(&g_mqtt_recv_thread, NULL, demo_mqtt_recv_thread, mqtt_handle);
479 if (res != 0) {
480 printf("create demo_mqtt_recv_thread failed: %d\n", res);
481 return -1;
482 }
483
484 /* 主循环进入休眠 */
485 while (1) {
486 /* TODO: 以下代码演示了简单的属性上报和事件上报, 用户可取消注释观察演示效果 */
487 demo_send_property_post(dm_handle, "{\"LightSwitch\": 0}");
488 demo_send_event_post(dm_handle, "Error", "{\"ErrorCode\": 0}");
489
490 aos_msleep(10000);
491 }
492
493 /* 断开MQTT连接, 一般不会运行到这里 */
494 res = aiot_mqtt_disconnect(mqtt_handle);
495 if (res < STATE_SUCCESS) {
496 aiot_mqtt_deinit(&mqtt_handle);
497 printf("aiot_mqtt_disconnect failed: -0x%04X\n", -res);
498 return -1;
499 }
500
501 /* 销毁DATA-MODEL实例, 一般不会运行到这里 */
502 res = aiot_dm_deinit(&dm_handle);
503 if (res < STATE_SUCCESS) {
504 printf("aiot_dm_deinit failed: -0x%04X\n", -res);
505 return -1;
506 }
507
508 /* 销毁MQTT实例, 一般不会运行到这里 */
509 res = aiot_mqtt_deinit(&mqtt_handle);
510 if (res < STATE_SUCCESS) {
511 printf("aiot_mqtt_deinit failed: -0x%04X\n", -res);
512 return -1;
513 }
514
515 g_mqtt_process_thread_running = 0;
516 g_mqtt_recv_thread_running = 0;
517
518 return 0;
519 }
520
521