OneNET连接流程

87 阅读 0 评论 58 点赞

我是靠谱客的博主清秀皮带，这篇文章主要介绍OneNET连接流程，现在分享给大家，希望可以做个参考。

OneNET-中国移动物联网开放平台是由中国移动打造的PaaS物联网开放平台。平台能够帮助开发者轻松实现设备接入与设备连接,提供综合性的物联网解决方案,实现物联网设备的数据获取,数据存储,数据展现。本文就介绍如何将传统的设备连接到OneNET。
采用MQTT+JSON格式对接OneNET平台上传温度度的数据，当然我们也可以在平台下发消息来控制设备。

1.操作和实战

注册OneNET：https://open.iot.10086.cn/，注册OneNET账号。注册过程比较简单这里就不再累述。登录之后。打开产品服务 MQTT物联网套件点击立刻使用，如下图所示：
在这里插入图片描述

进入MQTT物联网套件，在该页面点击添加产品，如下图所示：
在这里插入图片描述

从上图中，技术参数根据自己的硬件选择，如果我们的联网硬件为无线产品，则选择wifi;如果我们的联网硬件为有线产品，则选择移动蜂窝网络。协议为MQTTS，操作系统为无，而网络运营商可以全选，最后点击完成就可以创建产品，如下图所示：
在这里插入图片描述

点击上图的产品，进入相关界面，然后点击设备列表创建设备，如下图所示：
在这里插入图片描述

添加完成之后，我们得到一个设备，，如下图所示：
在这里插入图片描述

点击上图中的数据流，这些数据流是由用户添加的，例如显示温湿度等等，如下图所示：
在这里插入图片描述

最后我们得到三元组内容，如下源码所示：

复制代码

1
2
3
4
5
6
/* 用户需要根据设备信息完善以下宏定义中的三元组内容 */
#define CLIENTID "MQTT"   /* 设备名 */
#define USERNAME "366007" /* 产品ID */
/* 该密码需要onenet提供的token软件计算得出 */
#define PASSWORD "version=2018-10-31&res=products%2F366007%2Fdevices%2FMQTT&et=1672735919&method=md5&sign=qI0pgDJnICGoPdhNi%2BHtfg%3D%3D"

上述的参数是由OneNET来提供，MQTT是刚刚我们创建的设备名称，“366007”是刚刚我们创建的产品ID，而密钥是从产品概述获取，如下图所示：

上述是OneNET平台构建MQTT服务器完成，下面我们讲解代码部分。
在这里插入图片描述

2. MQTT源码包和cjSON源码包下载

MQTT官方网址（http://mqtt.org/）下载MQTT代码包；
cjSON官方网址（https://sourceforge.net/projects/cjson/）下载cjSON代码包；
然后把这些源码移植到我们工程当中，移植步骤如下所示：
(1)添加一下源码到工程当中
在这里插入图片描述

(2)在工程添加一下分组：
在这里插入图片描述

(3)添加这些源码头文件路径：

3. 连接

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
/* 用户需要根据设备信息完善以下宏定义中的三元组内容 */
#define CLIENTID "MQTT"   /* 设备名 */
#define USERNAME "366007" /* 产品ID */
/* 该密码需要onenet提供的token软件计算得出 */
#define PASSWORD "version=2018-10-31&res=products%2F366007%2Fdevices%2FMQTT&et=1672735919&method=md5&sign=qI0pgDJnICGoPdhNi%2BHtfg%3D%3D"

/* 以下参数的宏定义固定，不需要修改，只修改上方的参数即可 */
#define HOST_NAME     "open.iot.10086.cn"  /*onenet域名 */
#define HOST_PORT     1883
#define DEVICE_SUBSCRIBE   "$sys/"USERNAME"/"CLIENTID"/dp/post/json/+"/* 订阅 */
#define DEVICE_PUBLISH     "$sys/"USERNAME"/"CLIENTID"/dp/post/json"/* 发布 */

复制代码

1
2
3
订阅和发布的写法根据OneNET的要求而定义。
(2) 在onenet.c文件定义四个函数，分别连接、获取、打开和关闭，如下源码所示：

复制代码

/*
* @brief  通过TCP方式发送数据到TCP服务器
*/
int lwip_transport_send_packet_buffer(int sock, unsigned char *buf, 
int buflen)
{
    int rc = 0;
    rc = write(sock, buf, buflen);
    return rc;
}

/*
* @brief  阻塞方式接受TCP服务器发送的数据
/
int lwip_transport_getdata(uint8_t* buf, int32_t count)
{
    int rc = recv(mysock, buf, count, 0);
    return rc;
}

/*
* @brief  打开一个网络接口，其实就是和服务器建立一个 TCP 连接。
/
int lwip_transport_open(char* addr, int port)
{
    int* sock = &mysock;
    struct hostent *server;
    struct sockaddr_in serv_addr;
    int timeout = 1000;

*sock = socket(AF_INET, SOCK_STREAM, 0);
    if(*sock < 0)
      printf("[ERROR] Create socket failedn");
    
    server = gethostbyname(addr);              /* 对阿里云服务器地址解析 */
    if(server == NULL)
      printf("[ERROR] Get host ip failedn");
    
    memset(&serv_addr,0,sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(port);          /* 设置端口号 */
    memcpy(&serv_addr.sin_addr.s_addr,server->h_addr,server->h_length);
    
    if(connect(*sock,(struct sockaddr *)&serv_addr,sizeof(serv_addr)) < 0)
    {
      printf("[ERROR] connect failedn");
          return -1;
    }
    
setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO,
            (char*)&timeout,sizeof(timeout));

return mysock;
}

/*
 @brief  关闭连接
/
int lwip_transport_close(int sock)
{
    int rc;

rc = shutdown(sock, SHUT_WR);
    rc = recv(sock, NULL, (size_t)0, 0);
    rc = close(sock);

return rc;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
/*
* @brief  通过TCP方式发送数据到TCP服务器
*/
int lwip_transport_send_packet_buffer(int sock, unsigned char *buf, 
int buflen)
{
    int rc = 0;
    rc = write(sock, buf, buflen);
    return rc;
}

/*
* @brief  阻塞方式接受TCP服务器发送的数据
/
int lwip_transport_getdata(uint8_t* buf, int32_t count)
{
    int rc = recv(mysock, buf, count, 0);
    return rc;
}

/*
* @brief  打开一个网络接口，其实就是和服务器建立一个 TCP 连接。
/
int lwip_transport_open(char* addr, int port)
{
    int* sock = &mysock;
    struct hostent *server;
    struct sockaddr_in serv_addr;
    int timeout = 1000;

    *sock = socket(AF_INET, SOCK_STREAM, 0);
    if(*sock < 0)
      printf("[ERROR] Create socket failedn");
    
    server = gethostbyname(addr);              /* 对阿里云服务器地址解析 */
    if(server == NULL)
      printf("[ERROR] Get host ip failedn");
    
    memset(&serv_addr,0,sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(port);          /* 设置端口号 */
    memcpy(&serv_addr.sin_addr.s_addr,server->h_addr,server->h_length);
    
    if(connect(*sock,(struct sockaddr *)&serv_addr,sizeof(serv_addr)) < 0)
    {
      printf("[ERROR] connect failedn");
          return -1;
    }
    
setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO,
            (char*)&timeout,sizeof(timeout));

    return mysock;
}

/*
 @brief  关闭连接
/
int lwip_transport_close(int sock)
{
    int rc;

    rc = shutdown(sock, SHUT_WR);
    rc = recv(sock, NULL, (size_t)0, 0);
    rc = close(sock);

    return rc;
}

复制代码

1
2
3
这些函数是由用户来实现，例如连接函数，根据自己的硬件编写连接。上述的源码小编以LWIP为例。
(3) 订阅和发布操作

复制代码

void lwip_demo(void)
{
    mysock = lwip_transport_open((char *)HOST_NAME, HOST_PORT);
    data.clientID.cstring = CLIENTID;
    data.username.cstring = USERNAME;
    data.password.cstring = PASSWORD;
    unsigned char payload_out[200];
    int payload_out_len = 0;
    uint32_t sendtick = xTaskGetTickCount();

while (1)
{
/* 客户端的发布操作 */
        if ((xTaskGetTickCount() - sendtick) >= (send_duration * 100)) 
        {
            sendtick = xTaskGetTickCount();/* sendtick重新计数 */
            taskENTER_CRITICAL();/* 进入临界区 */
            temp = 30 + rand() % 10 + 1; /* 温度的数据 */
            humid = 54.8 + rand() % 10 + 1; /* 湿度的数据 */
            sprintf((char *)payload_out, "{"id": 123,"dp":
                     { "temperatrue": [{"v": %0.1f,}],"power":
                     [{"v": %0.1f,}]}}", temp, humid);
            payload_out_len = strlen((char *)payload_out);
            topicString.cstring = DEVICE_PUBLISH;   /* 属性上报 发布 */
            len = MQTTSerialize_publish((unsigned char *)buf, 
buflen, 0, req_qos, 
retained, msgid, 
topicString, 
payload_out, 
payload_out_len);
            rc = lwip_transport_send_packet_buffer(mysock, 
(unsigned char *)buf, len);
            taskEXIT_CRITICAL();                    /* 退出临界区 */
            if (rc == len)
                printf("send PUBLISH Successfullyrn");
            else
                printf("send PUBLISH failedrn");

printf("send temp(%0.1f)&humid(%0.1f) !rn", temp, humid);
            recv_server_flag = 1;
        }
        
        vTaskDelay(100);
        
        switch (msgtypes)
        {

case CONNECT:  /* 客户端发送服务器的连接操作 */
                printf("进入CONNECT状态rn");
                len = MQTTSerialize_connect((unsigned char *)buf, 
buflen, &data);         /* 获取数据组长发送连接信息  */
                rc = lwip_transport_send_packet_buffer(mysock, 
(unsigned char *)buf, len); /* 发送返回发送数组长度 */

if (rc == len)
                    printf("发送连接成功rn");
                else
                    printf("发送连接失败rn");

printf("MQTT连接服务器!rn");
                printf("退出CONNECT状态rnrn");
                msgtypes = 0;
                recv_server_flag = 1;
                break;

case CONNACK:  /* 服务器发送客户端确认连接请求 */
                if (MQTTDeserialize_connack(&sessionPresent, 
&connack_rc, (unsigned char *)buf, buflen) != 1 
|| connack_rc != 0)  /* 收到回执 */
                {
/* 回执不一致，连接失败 */
                    printf("Unable to connect, return code %drn", connack_rc);   
                }
                else
                {
                    printf("MQTT is concet OK!rn");   /* 连接成功 */
                }
                msgtypes = SUBSCRIBE;   /* 连接成功 执行 订阅 操作 */
                break;

case SUBSCRIBE:  /* 客户端发送到服务器的订阅操作 */
                topicString.cstring = DEVICE_SUBSCRIBE;
                len = MQTTSerialize_subscribe((unsigned char *)buf, 
buflen, 0, msgid, 1, &topicString, &req_qos);
                rc = lwip_transport_send_packet_buffer(mysock, 
(unsigned char *)buf, len);

if (rc == len)
                    printf("send SUBSCRIBE Successfullyrn");
                else
                {
                    printf("send SUBSCRIBE failedrn");
                    t++;

if (t >= 10)
                    {
                        t = 0;
                        msgtypes = CONNECT;
                    }
                    else
                        msgtypes = SUBSCRIBE;

break;
                }
                msgtypes = 0;
                recv_server_flag = 1;
                break;

case SUBACK:  /* 服务器发送到客户端的订阅确认 */
/* 有回执  QoS */
                rc = MQTTDeserialize_suback(&submsgid, 1, &subcount,
                                                  &granted_qos, (unsigned char *)buf,
                                                   buflen);  
                printf("granted qos is %drn", granted_qos);    /* 打印 Qos */
                msgtypes = 0;
                break;

case PUBLISH:  /* 服务器的发布操作 */
                /* 服务器的发布数据，自行编写 */
                break;

case PUBACK:   /* 发布成功 */
                msgtypes = 0;
                break;

default:
                break;

}
        
        if (recv_server_flag == 1)
        {
            memset(buf, 0, buflen); 
/* 轮询，读MQTT返回数据，*/
            rc = MQTTPacket_read((unsigned char *)buf, buflen,
                                      lwip_transport_getdata); 
            if (rc > 0) /* 如果有数据，进入相应状态。*/
            {
                msgtypes = rc;
            }
            recv_server_flag = 0;
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
void lwip_demo(void)
{
    mysock = lwip_transport_open((char *)HOST_NAME, HOST_PORT);
    data.clientID.cstring = CLIENTID;
    data.username.cstring = USERNAME;
    data.password.cstring = PASSWORD;
    unsigned char payload_out[200];
    int payload_out_len = 0;
    uint32_t sendtick = xTaskGetTickCount();

    while (1)
{
/* 客户端的发布操作 */
        if ((xTaskGetTickCount() - sendtick) >= (send_duration * 100)) 
        {
            sendtick = xTaskGetTickCount();/* sendtick重新计数 */
            taskENTER_CRITICAL();/* 进入临界区 */
            temp = 30 + rand() % 10 + 1; /* 温度的数据 */
            humid = 54.8 + rand() % 10 + 1; /* 湿度的数据 */
            sprintf((char *)payload_out, "{"id": 123,"dp":
                     { "temperatrue": [{"v": %0.1f,}],"power":
                     [{"v": %0.1f,}]}}", temp, humid);
            payload_out_len = strlen((char *)payload_out);
            topicString.cstring = DEVICE_PUBLISH;   /* 属性上报 发布 */
            len = MQTTSerialize_publish((unsigned char *)buf, 
buflen, 0, req_qos, 
retained, msgid, 
topicString, 
payload_out, 
payload_out_len);
            rc = lwip_transport_send_packet_buffer(mysock, 
(unsigned char *)buf, len);
            taskEXIT_CRITICAL();                    /* 退出临界区 */
            if (rc == len)
                printf("send PUBLISH Successfullyrn");
            else
                printf("send PUBLISH failedrn");

            printf("send temp(%0.1f)&humid(%0.1f) !rn", temp, humid);
            recv_server_flag = 1;
        }
        
        vTaskDelay(100);
        
        switch (msgtypes)
        {

            case CONNECT:  /* 客户端发送服务器的连接操作 */
                printf("进入CONNECT状态rn");
                len = MQTTSerialize_connect((unsigned char *)buf, 
buflen, &data);         /* 获取数据组长发送连接信息  */
                rc = lwip_transport_send_packet_buffer(mysock, 
(unsigned char *)buf, len); /* 发送返回发送数组长度 */

                if (rc == len)
                    printf("发送连接成功rn");
                else
                    printf("发送连接失败rn");

                printf("MQTT连接服务器!rn");
                printf("退出CONNECT状态rnrn");
                msgtypes = 0;
                recv_server_flag = 1;
                break;

            case CONNACK:  /* 服务器发送客户端确认连接请求 */
                if (MQTTDeserialize_connack(&sessionPresent, 
&connack_rc, (unsigned char *)buf, buflen) != 1 
|| connack_rc != 0)  /* 收到回执 */
                {
/* 回执不一致，连接失败 */
                    printf("Unable to connect, return code %drn", connack_rc);   
                }
                else
                {
                    printf("MQTT is concet OK!rn");   /* 连接成功 */
                }
                msgtypes = SUBSCRIBE;   /* 连接成功 执行 订阅 操作 */
                break;

            case SUBSCRIBE:  /* 客户端发送到服务器的订阅操作 */
                topicString.cstring = DEVICE_SUBSCRIBE;
                len = MQTTSerialize_subscribe((unsigned char *)buf, 
buflen, 0, msgid, 1, &topicString, &req_qos);
                rc = lwip_transport_send_packet_buffer(mysock, 
(unsigned char *)buf, len);

                if (rc == len)
                    printf("send SUBSCRIBE Successfullyrn");
                else
                {
                    printf("send SUBSCRIBE failedrn");
                    t++;

                    if (t >= 10)
                    {
                        t = 0;
                        msgtypes = CONNECT;
                    }
                    else
                        msgtypes = SUBSCRIBE;

                    break;
                }
                msgtypes = 0;
                recv_server_flag = 1;
                break;

            case SUBACK:  /* 服务器发送到客户端的订阅确认 */
/* 有回执  QoS */
                rc = MQTTDeserialize_suback(&submsgid, 1, &subcount,
                                                  &granted_qos, (unsigned char *)buf,
                                                   buflen);  
                printf("granted qos is %drn", granted_qos);    /* 打印 Qos */
                msgtypes = 0;
                break;

            case PUBLISH:  /* 服务器的发布操作 */
                /* 服务器的发布数据，自行编写 */
                break;

            case PUBACK:   /* 发布成功 */
                msgtypes = 0;
                break;

            default:
                break;

        }
        
        if (recv_server_flag == 1)
        {
            memset(buf, 0, buflen); 
/* 轮询，读MQTT返回数据，*/
            rc = MQTTPacket_read((unsigned char *)buf, buflen,
                                      lwip_transport_getdata); 
            if (rc > 0) /* 如果有数据，进入相应状态。*/
            {
                msgtypes = rc;
            }
            recv_server_flag = 0;
        }
    }
}