devbox
小桥流水78
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

ESP32 基于4G模块透传 模式 实现MQTT通信_esp32 4g上网

作者：小桥流水78 | 2024-08-07 03:36:26

赞

踩

esp32 4g上网

这里选择的是SIM7600CE 和EC20 4G通信模块，工作在透传模式

  1. 创建GPRS 拨号的通信接口源文件 GprsUtils.c ,实现4G模块 拨号及连接服务器功能

  1. #include <string.h>
  2. #include <stdint.h>
  3. #include "freertos/FreeRTOS.h"
  4. #include "freertos/task.h"
  5. #include "GprsUtils.h"
  6. #include "UartUtils.h"
  7. #include "driver/gpio.h"
  8.     
  9. static uint8_t find_string(const char * recvBuff,uint16_t recvBuffLen,const char* p)
  10. { 
  11.     if(recvBuffLen>0&&strstr(recvBuff,p)!=NULL)
  12.     {
  13.         return 1;
  14.     }
  15.     else
  16.     {
  17.         return 0;
  18.     }
  19. }
  20. static void delay_ms(int waitTimeMs){
  21.     vTaskDelay(waitTimeMs/portTICK_PERIOD_MS);
  22.     
  23. }
  24.  
  25. void gprs_send_string(char* s){
  26.     uartSendBytes(GPRSPORT,(uint8_t *)s,strlen(s));
  27. }
  28.  
  29. void gprs_recv_ack(uint8_t *buf,int *len,int wait_time){
  30.     uartRecvBytes(GPRSPORT, buf, len,MAX_UART_RECV_BUFF_SIZE,wait_time);
  31. }
  32.  
  33.  
  34. uint8_t gprs_send_cmd(char *b,char *a,uint8_t times,uint16_t wait_time)         
  35. {
  36.     uint8_t i;
  37.     i = 0;
  38.     
  39.     while(i < times)                    
  40.     {
  41.         char gprsBuf[MAX_UART_RECV_BUFF_SIZE]={0};
  42.         int gprsBufLen=0;
  43.         gprs_send_string(b);
  44.         gprs_send_string("\r\n");             // 回车换行
  45.         gprs_recv_ack((uint8_t *)gprsBuf,&gprsBufLen,wait_time);
  46.         if(find_string(gprsBuf,gprsBufLen,a))
  47.             return 1;
  48.         i++;
  49.     }
  50.     return 0;
  51. }
  52.  
  53.  
  54. /******************************************************************************
  55. * 函数名称:str_delim
  56. * 描    述:字符串分割函数
  57. * 输入参数: num   0---取出分隔字符串（delim）前面的字符串  1---取出分隔字符串（delim）后面的字符串
  58. *           temp  要分割的字符串
  59. *           delim 分隔符字符串
  60. * 输    出: 
  61. * 返    回: 
  62. * 说    明:          
  63. *******************************************************************************/
  64. char *str_delim(uint8_t num,char *temp,char *delim)
  65. {
  66.     int i;
  67.     char *str[2]={0};
  68.     char *tok=temp;
  69.     char *restr;
  70.     for(i=0;i<2;i++)
  71.     {
  72.         tok=strtok(tok,delim);
  73.         str[i]=tok;
  74.         tok = NULL;
  75.     }
  76.     restr=str[num];
  77.   return restr;
  78. }
  79.  
  80.  
  81.  
  82. uint8_t sim7600CheckStatus(void);
  83. uint8_t sim7600SelectNet(void);
  84. uint8_t sim7600NetConfig(void);
  85. uint8_t sim7600TcpConnect(const char *serverIp,int port);
  86. uint8_t sim7600ConnectServer(const char *serverIp,int port);
  87. void     sim7600ExitDataMode(void);
  88. uint8_t sim7600CloseTcpConnect(void);
  89. uint8_t sim7600CloseNet(void);
  90.  
  91. /******************************************************************************
  92. * 函数名称: check_status
  93. * 描    述: 核心板基本状态测试
  94. * 输入参数: 无
  95. * 输    出: 无
  96. * 返    回: 0 --- 出错  1 --- 正确
  97. * 说    明: 核心板开机后，先判断AT命令是否正常、能否读到卡、能否注册网络。确认无误后
  98. *           再进行其他功能测试
  99. *******************************************************************************/
  100. uint8_t sim7600CheckStatus(void)
  101. {
  102.      sim7600ExitDataMode();                                               // 退出透传模式
  103.     // 同步波特率       
  104.     if(!gprs_send_cmd("AT","OK",5,200)) 
  105.     {
  106.         printf("handshake failed\r\n");
  107.         return 0;
  108.     }
  109.     // 关闭网络，避免后面因为网络已经开启，导致打开网络失败
  110.     gprsCloseNet();
  111.     // 取消回显             
  112.     if(!gprs_send_cmd("ATE0","OK",1,200)){
  113.         printf("echo cancelled failed\r\n");
  114.         return 0;
  115.     }
  116.     
  117.     // 查询核心板能否读到SIM卡            
  118.     if(!gprs_send_cmd("AT+CPIN?","+CPIN: READY",2,500)){
  119.        printf("no sim card detected\r\n");
  120.        return 0;
  121.     }
  122.     
  123.     if(!gprs_send_cmd("AT+CSQ","+CSQ",2,100))
  124.     {                
  125.         printf("serching CSQ failed\r\n");
  126.         return 0;
  127.     }
  128.     if(!gprs_send_cmd("AT+COPS?","OK",2,100))
  129.     {                
  130.         printf("get Operator failed\r\n");
  131.         return 0;
  132.     }
  133.     
  134.     // 选择网络
  135.     if(!sim7600SelectNet())                                              // 第一次选择好网路后，可直接去掉该函数
  136.       return 0;    
  137.     return 1;
  138. }
  139.  
  140. /******************************************************************************
  141. * 函数名称: select_net
  142. * 描    述: 核心板选择网络
  143. * 输入参数: 
  144. * 输    出: 
  145. * 返    回: 0 --- 出错  1 --- 正确
  146. * 说    明: SIM卡首次插入核心板测试时，需要选择网络，选择好网络后，如果不换卡，则不需要再执行该函数
  147. *******************************************************************************/
  148.  
  149. uint8_t sim7600SelectNet(void)
  150. {
  151.     
  152.     // 设置APN
  153.     if(!gprs_send_cmd("AT+CGSOCKCONT=1,\"IP\",\"CMNET\"","OK",1,500)){
  154.         printf("set APN failed\r\n");
  155.         return 0;
  156.     }
  157.     
  158.     // 选择网络，可在常量声明中修改
  159.     if(!gprs_send_cmd("AT+CNMP=38","OK",1,200)){
  160.         printf("select network failed\r\n");
  161.         return 0;
  162.     }    
  163.     //查询核心板所处网络
  164.     if(!gprs_send_cmd("AT+CNMP?","+CNMP: 38",3,500)){
  165.         printf("register network failed\r\n");
  166.         return 0;
  167.     }
  168.  
  169.  
  170.     
  171.     // 查询核心板是否注册成功             
  172.     if(!gprs_send_cmd("AT+CREG?","+CREG: 0,1",5,1000)){
  173.          printf("register network failed\r\n");
  174.          return 0;
  175.     }
  176.     return 1;
  177. }
  178.  
  179. /******************************************************************************
  180. * 函数名称: net_config
  181. * 描    述: 核心板进行连接前的网络配置
  182. * 输入参数: 无
  183. * 输    出: 无
  184. * 返    回: 0 --- 出错  1 --- 正确
  185. * 说    明: 判断核心板是否可以进行通信连接          
  186. *******************************************************************************/
  187. uint8_t sim7600NetConfig(void)
  188. {
  189.     // 激活启动场景
  190.     if(!gprs_send_cmd("AT+CSOCKSETPN=1","OK",1,200) ){
  191.          printf("activate mobile scene failed\r\n");
  192.          return 0;
  193.     }    
  194.  
  195.   // 设置为非透传模式
  196.     if(!gprs_send_cmd("AT+CIPMODE?","+CIPMODE: 1",1,200)){   // 这里不能直接设置非透传模式，避免因为重复设置导致ERROR
  197.     
  198.          if(!gprs_send_cmd("AT+CIPMODE=1","OK",1,200)){
  199.              printf("Transparent transmission mode failed\r\n");
  200.              return 0;
  201.          }
  202.              
  203.     }
  204.  
  205.     // 打开网络
  206.     if(!gprs_send_cmd("AT+NETOPEN","+NETOPEN: 0",1,5000)){  // 不能重复操作，否则会ERROR
  207.         printf("open network failed\r\n");
  208.         return 0;
  209.     }
  210.  
  211.     // 获取本地IP地址
  212.     if(gprs_send_cmd("AT+IPADDR","ERROR",1,2000))
  213.     {
  214.          printf("get local Ipaddress failed\r\n"); 
  215.          return 0;
  216.     }
  217.  
  218.     /************
  219.     else
  220.     {
  221.        printf("11.获取本地IP成功\r\n");
  222.        printf("%s",str_delim(1,gprsBuf,":"));            // 将获取到的本地IP地址打印到串口调试助手
  223.     }
  224.     ************/
  225.     return 1;
  226. }
  227.  
  228. /******************************************************************************
  229. * 函数名称: tcp_connect
  230. * 描    述: 核心板与云服务器进行TCP透传模式通信
  231. * 输入参数: 无
  232. * 输    出: 无
  233. * 返    回: 0 --- 出错  1 --- 正确
  234. * 说    明: 无       
  235. *******************************************************************************/
  236. uint8_t sim7600TcpConnect(const char *serverIp,int port)
  237. {
  238.     char serverIpCmd[512]={0};
  239.     sprintf(serverIpCmd,"AT+CIPOPEN=0,\"TCP\",\"%s\",%d\r\n",serverIp,port);    // TCP服务器IP地址，可自行修改
  240.     // 建立TCP连接
  241.     if(!gprs_send_cmd((char*)serverIpCmd,"CONNECT",2,3000)){
  242.          printf("set Tcp Transparent transmission mode failed \r\n");
  243.          return 0;    
  244.     }
  245.     return 1;
  246. }
  247.  
  248.  
  249. uint8_t sim7600ConnectServer(const char *serverIp,int port){
  250.     if(!sim7600CheckStatus())                                   // 判断核心板状态是否正常
  251.     {
  252.        printf("*** check_status failed ***\r\n");
  253.        return 0;
  254.     }else
  255.        printf("*** check_status ok ***\r\n");
  256.     if(!sim7600NetConfig()){                                     // 判断通信连接是否正常
  257.         printf("*** net_config failed  ***\r\n");
  258.         return 0; 
  259.     }else
  260.         printf("*** net_config OK  ***\r\n");
  261.     
  262.     return sim7600TcpConnect(serverIp,port);
  263. }
  264.  
  265.  
  266. /********************************************************************************
  267. * 函数名称:  exit_data_mode 
  268. * 描    述:  退出数据模式
  269. * 输入参数:  无
  270. * 输    出:  无
  271. * 返    回:  无
  272. * 说    明:  发送“+++”，确保核心板退出透传数据模式，前后1s延时
  273. *******************************************************************************/
  274. void sim7600ExitDataMode(void)
  275. {
  276.   char gprsBuf[MAX_UART_RECV_BUFF_SIZE]={0};
  277.   int gprsBufLen=0;
  278.   delay_ms(1000);
  279.   gprs_send_string("+++");
  280.   gprs_recv_ack((uint8_t *)gprsBuf,&gprsBufLen,1000);
  281. }
  282.  
  283. /********************************************************************************
  284. * 函数名称:  close_tcp_connect 
  285. * 描    述:  关闭TCP连接
  286. * 输入参数:  无
  287. * 输    出:  无
  288. * 返    回:  无
  289. * 说    明: 关闭TCP连接
  290. *******************************************************************************/
  291. uint8_t sim7600CloseTcpConnect(void)
  292. {
  293.     // 退出透传数据模式
  294.     sim7600ExitDataMode();
  295.     delay_ms(1500);
  296.      // 关闭TCP连接    
  297.     if(!gprs_send_cmd("AT+CIPCLOSE=0","OK",2,2000)){       // 关闭连接耗时较长（+CIPCLOSE: 0,0），可根据实际情况调整
  298.         printf("close tcp port failed\n");
  299.         return 0;
  300.     }
  301.     return 1;
  302. }
  303.  
  304.  
  305.  
  306. uint8_t sim7600CloseNet(void){
  307.     // 关闭网络
  308.     sim7600CloseTcpConnect();
  309.     if(!gprs_send_cmd("AT+NETCLOSE","+NETCLOSE:",5,3000))
  310.     {
  311.         printf("close network failed\r\n");    
  312.         return 0;        
  313.     }    
  314.     return 1;
  315. }
  316.  
  317.  
  318.  
  319.  
  320. /EC20  Start/
  321.  
  322.  
  323. uint8_t ec20ConnectServer(const char *serverIp,int port);
  324. uint8_t ec20CloseNet(void);
  325.  
  326. uint8_t ec20ConnectServer(const char *serverIp,int port){
  327.     printf("ec20ConnectServer:%s %d\n",serverIp,port);
  328.     if(!gprs_send_cmd("AT","OK",2,1000)){
  329.         printf("handshake failed\r\n");
  330.         return 0;
  331.     }
  332.     // 查询核心板能否读到SIM卡            
  333.     if(!gprs_send_cmd("AT+CPIN?","+CPIN: READY",2,1000)){
  334.        printf("no sim card detected\r\n");
  335.        return 0;
  336.     }
  337.     if(!gprs_send_cmd("AT+CSQ","+CSQ",2,1000))
  338.     {                
  339.         printf("serching CSQ failed\r\n");
  340.         return 0;
  341.     }
  342.     // 查询核心板是否注册成功             
  343.     if(!gprs_send_cmd("AT+CREG?","+CREG: 0,1",5,1000)){
  344.          printf("register GSM network failed\r\n");
  345.          return 0;
  346.     }
  347.     if(!gprs_send_cmd("AT+CGREG?","OK",5,1000)){
  348.          printf("register GPRS network failed\r\n");
  349.          return 0;
  350.     }
  351.     // 设置APN
  352.     if(!gprs_send_cmd("AT+QICSGP=1,1,\"CMNET\"","OK",2,1000)){
  353.         printf("set APN failed\r\n");
  354.         return 0;
  355.     }
  356.  
  357.     if(!gprs_send_cmd("AT+QIDEACT=1","OK",2,1000)){
  358.         printf("Deactivates a PDP context\r\n");
  359.         return 0;
  360.     }
  361.     if(!gprs_send_cmd("AT+QIACT=1","OK",2,1000)){
  362.         printf("activates a PDP context\r\n");
  363.         return 0;
  364.     }
  365.  
  366.     char serverIpCmd[512]={0};
  367.     sprintf(serverIpCmd,"AT+QIOPEN=1,0,\"TCP\",\"%s\",%d,0,2\r\n",serverIp,port);    // TCP服务器IP地址，可自行修改
  368.     // 建立TCP连接
  369.     if(!gprs_send_cmd((char*)serverIpCmd,"CONNECT",2,3000)){
  370.          printf("set Tcp Transparent transmission mode failed \r\n");
  371.          return 0;    
  372.     }
  373.     return 1;
  374.  
  375.  
  376. }
  377.  
  378. uint8_t ec20CloseNet(void){
  379.  
  380.     sim7600ExitDataMode();
  381.     delay_ms(2000);
  382.      // 关闭TCP连接    
  383.     if(!gprs_send_cmd("AT+QICLOSE=0","OK",2,2000)){       // 关闭连接耗时较长（+CIPCLOSE: 0,0），可根据实际情况调整
  384.         printf("close tcp port failed\n");
  385.         return 0;
  386.     }
  387.     return 1;
  388. }
  389.  
  390. /EC20  End/
  391.  
  392.  
  393.  
  394.  
  395. //#define SIM7600CE_4G        1
  396. #define EC20_4G                1
  397.  
  398.  
  399.  
  400.  
  401.  
  402. void gprsInit(int uartPort,uint32_t bound){
  403.     uart_init(uartPort,bound);
  404.     gprsPower(1);
  405.     gprsReset(0);
  406.     delay_ms(2000);
  407. }
  408.  
  409. uint8_t gprsTryConnectServer(const char *serverIp,int port){
  410.     #if SIM7600CE_4G
  411.         return sim7600ConnectServer(serverIp,port);
  412.     #else
  413.         return ec20ConnectServer(serverIp,port);
  414.     #endif
  415.     
  416. }
  417.  
  418. uint8_t gprsCloseNet(void)
  419. {
  420.     #if SIM7600CE_4G
  421.         return sim7600CloseNet();
  422.     #else
  423.         return ec20CloseNet();
  424.     #endif
  425. }
  426.  
  427. int gprsTcpWrite(uint8_t *data,int len){
  428.     return uartSendBytes(GPRSPORT,data,len);
  429. }
  430.  
  431. int gprsTcpRead(uint8_t *recvbuf,int maxRecvLen,int waitTime){
  432.     int readLen=0;
  433.     uartRecvBytes(GPRSPORT, recvbuf, &readLen,maxRecvLen,waitTime);
  434.     return readLen;
  435. }
  436. int gprsTcpAvailable(int waitTime){
  437.     return uartHasDataReceived(GPRSPORT, waitTime);
  438. }
  439.  
  440. void gprsReboot(void){
  441.     gpio_set_level(SIM_RST_Pin,1);
  442.     delay_ms(2000);
  443.     gpio_set_level(SIM_RST_Pin, 0);
  444.     delay_ms(20000);
  445. }
  446.  
  447. void gprsPower(uint8_t isOn){
  448.     gpio_set_level(SIM_PEN_Pin,isOn);
  449. }
  450. void gprsReset(uint8_t isOn){
  451.     gpio_set_level(SIM_RST_Pin, isOn);
  452. }
  453. void gprsRepowerOn(void) {
  454.     gpio_set_level(SIM_PEN_Pin,1);
  455.     delay_ms(4000);
  456. }
  457. void gprsRepowerOff(void) {
  458.     gpio_set_level(SIM_PEN_Pin,0);
  459.     delay_ms(2000);
  460. }

  1. 移植MQTT 通信协议

从https://github.com/eclipse/paho.mqtt.embedded-c/tree/master/MQTTPacket 下载MQTT通信源码包, 将src目录下所有文件拷贝到自己的工程目录下，然后移植samples目录下的transport.c

  1. #include "transport.h"
  2. #include "lwip/opt.h"
  3. #include "lwip/arch.h"
  4. #include "lwip/api.h"
  5. #include "lwip/inet.h"
  6. #include "lwip/sockets.h"
  7. #include "string.h"
  8. #include "../GprsUtils/GprsUtils.h"
  9.  
  10. static int mysock;
  11.  
  12. /************************************************************************
  13. ** 函数名称: transport_sendPacketBuffer                                    
  14. ** 函数功能: 以TCP方式发送数据
  15. ** 入口参数: unsigned char* buf：数据缓冲区
  16. **           int buflen：数据长度
  17. ** 出口参数: <0发送数据失败                            
  18. ************************************************************************/
  19. int transport_sendPacketBuffer( uint8_t* buf, int buflen)
  20. {
  21.     return gprsTcpWrite(buf,buflen);
  22. }
  23.  
  24. /************************************************************************
  25. ** 函数名称: transport_getdata                                    
  26. ** 函数功能: 以阻塞的方式接收TCP数据
  27. ** 入口参数: unsigned char* buf：数据缓冲区
  28. **           int count：数据长度
  29. ** 出口参数: <=0接收数据失败                                    
  30. ************************************************************************/
  31. int transport_getdata(uint8_t* buf, int count)
  32. {
  33.     return gprsTcpRead(buf,count,500);
  34. }
  35.  
  36.  
  37. /************************************************************************
  38. ** 函数名称: transport_open                                    
  39. ** 函数功能: 打开一个接口，并且和服务器 建立连接
  40. ** 入口参数: char* servip:服务器域名
  41. **           int   port:端口号
  42. ** 出口参数: <0打开连接失败                                        
  43. ************************************************************************/
  44. int transport_open(const char *servip, int port)
  45. {
  46.     return gprsTryConnectServer(servip,port);
  47. }
  48.  
  49.  
  50. /************************************************************************
  51. ** 函数名称: transport_close                                    
  52. ** 函数功能: 关闭套接字
  53. ** 入口参数: unsigned char* buf：数据缓冲区
  54. **           int buflen：数据长度
  55. ** 出口参数: <0发送数据失败                            
  56. ************************************************************************/
  57. int transport_close(void)
  58. {
  59.     return gprsCloseNet();
  60. }

  1. 创建MQTT通信接口源文件，包含连接MQTT服务器、订阅、发布,接收

  1. #include <string.h>
  2. #include "freertos/FreeRTOS.h"
  3. #include "freertos/task.h"
  4. #include "MqttUtils.h"
  5. #include "../GprsUtils/GprsUtils.h"
  6.  
  7.  
  8.  
  9.  
  10.  
  11. /************************************************************************
  12. ** 函数名称: MQTTClientInit                                
  13. ** 函数功能: 初始化客户端并登录服务器
  14. ** 入口参数: int sock:网络描述符
  15. ** 出口参数: >=0:发送成功 <0:发送失败
  16. ** 备    注: 
  17. ************************************************************************/
  18. static int MQTTClientInit(const char *username,const char *password,int keepalive){
  19.         MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer;
  20.         uint8_t buf[100];
  21.         int buflen = sizeof(buf);
  22.         int len;
  23.        uint8_t sessionPresent,connack_rc;
  24.     
  25.     
  26.          //创建MQTT客户端连接参数
  27.       connectData.willFlag = 0;
  28.       //MQTT版本
  29.         connectData.MQTTVersion = 4;
  30.         //客户端ID--必须唯一
  31.         connectData.clientID.cstring = "MIMITOTO";
  32.         //保活间隔
  33.         connectData.keepAliveInterval = keepalive;
  34.         if(username&&password){
  35.             //用户名
  36.             connectData.username.cstring = username;
  37.             //用户密码
  38.             connectData.password.cstring = password;
  39.         }
  40.         //清除会话
  41.         connectData.cleansession = 1;
  42.         
  43.       //串行化连接消息
  44.         len = MQTTSerialize_connect(buf, buflen, &connectData);
  45.         //发送TCP数据
  46.     if(transport_sendPacketBuffer(buf, len) < 0)
  47.         return -1;
  48.       if(gprsTcpAvailable(2000)<0)
  49.           return -2;
  50.     if(MQTTPacket_read(buf, buflen, transport_getdata) != CONNACK)
  51.         return -3;    
  52.         //拆解连接回应包
  53.     if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0)
  54.         return -4;
  55.         
  56.     if(sessionPresent == 1)
  57.         return 1;//不需要重新订阅--服务器已经记住了客户端的状态
  58.     else 
  59.         return 0;//需要重新订阅
  60. }
  61.  
  62.  
  63. int mqttConnnectServer(const char *servip, int port,const char *username,const char *password,int keepalive){
  64.     int sessionPresent=-1;
  65.     if(!transport_open(servip,port))
  66.         return -1;
  67.     if((sessionPresent=MQTTClientInit(username,password,keepalive))<0){
  68.         transport_close();
  69.         return -1;
  70.     }
  71.     return sessionPresent;
  72. }
  73.  
  74.  
  75. /************************************************************************
  76. ** 函数名称: ReadPacketTimeout                    
  77. ** 函数功能: 阻塞读取MQTT数据
  78. ** 入口参数: int sock:网络描述符
  79. **           uint8_t *buf:数据缓存区
  80. **           int buflen:缓冲区大小
  81. **           uint32_t timeout:超时时间--0-表示直接查询，没有数据立即返回
  82. ** 出口参数: -1：错误,其他--包类型
  83. ** 备    注: 
  84. ************************************************************************/
  85. int ReadPacketTimeout(uint8_t *buf,int buflen,uint32_t timeout)
  86. {
  87.       if(timeout != 0)
  88.       {
  89.         if(gprsTcpAvailable(timeout)<0)
  90.             return -1;            
  91.       }
  92.         //读取TCP/IP事件
  93.         return MQTTPacket_read(buf, buflen, transport_getdata);
  94. }
  95.  
  96. /************************************************************************
  97. ** 函数名称: MQTTSubscribe                                
  98. ** 函数功能: 订阅消息
  99. ** 入口参数: int sock：套接字
  100. **           char *topic：主题
  101. **           enum QoS pos：消息质量
  102. ** 出口参数: >=0:发送成功 <0:发送失败
  103. ** 备    注: 
  104. ************************************************************************/
  105. int MQTTSubscribe(char *topic,enum QoS pos){
  106.       static uint32_t PacketID = 0;
  107.       uint16_t packetidbk = 0;
  108.       int conutbk = 0;
  109.         uint8_t buf[100];
  110.         int buflen = sizeof(buf);
  111.       MQTTString topicString = MQTTString_initializer;  
  112.         int len;
  113.       int req_qos,qosbk;
  114.     
  115.       //复制主题
  116.     topicString.cstring = (char *)topic;
  117.         //订阅质量
  118.     req_qos = pos;
  119.     
  120.       //串行化订阅消息
  121.     len = MQTTSerialize_subscribe(buf, buflen, 0, PacketID++, 1, &topicString, &req_qos);
  122.     //发送TCP数据
  123.     if(transport_sendPacketBuffer(buf, len) < 0)
  124.         return -1;
  125.     if(gprsTcpAvailable(2000) <0 ){
  126.         return -2;
  127.     }
  128.     //等待订阅返回--未收到订阅返回
  129.     if(MQTTPacket_read(buf, buflen, transport_getdata) != SUBACK)
  130.         return -4;    
  131.         
  132.         //拆订阅回应包
  133.     if(MQTTDeserialize_suback(&packetidbk,1, &conutbk, &qosbk, buf, buflen) != 1)
  134.         return -5;
  135.         
  136.         //检测返回数据的正确性
  137.     if((qosbk == 0x80)||(packetidbk != (PacketID-1)))
  138.         return -6;
  139.     //订阅成功
  140.         return 0;
  141. }
  142.  
  143.  
  144.  
  145.  
  146. /************************************************************************
  147. ** 函数名称: GetNextPackID                        
  148. ** 函数功能: 产生下一个数据包ID
  149. ** 入口参数: 无
  150. ** 出口参数: uint16_t packetid:产生的ID
  151. ** 备    注: 
  152. ************************************************************************/
  153. static uint16_t GetNextPackID(void)
  154. {
  155.      static uint16_t pubpacketid = 0;
  156.      return pubpacketid++;
  157. }
  158.  
  159. /************************************************************************
  160. ** 函数名称: WaitForPacket                    
  161. ** 函数功能: 等待特定的数据包
  162. ** 入口参数: int sock:网络描述符
  163. **           uint8_t packettype:包类型
  164. **           uint8_t times:等待次数
  165. ** 出口参数: >=0:等到了特定的包 <0:没有等到特定的包
  166. ** 备    注: 
  167. ************************************************************************/
  168. static int WaitForPacket(uint8_t packettype,uint8_t times)
  169. {
  170.       int type;
  171.         uint8_t buf[MSG_MAX_LEN];
  172.       uint8_t n = 0;
  173.         int buflen = sizeof(buf);
  174.         do
  175.         {
  176.                 //读取数据包
  177.             type = ReadPacketTimeout(buf,buflen,2000);
  178.             if(type != -1)
  179.                 mqtt_pktype_ctl(type,buf,buflen);
  180.             n++;
  181.         }while((type != packettype)&&(n < times));
  182.         //收到期望的包
  183.         if(type == packettype)
  184.              return 0;
  185.         else 
  186.              return -1;        
  187. }
  188.  
  189.  
  190. int MQTTMsgPublish(char *topic, char qos, char retained,uint8_t* msg,uint32_t msg_len){
  191.         uint8_t buf[MSG_MAX_LEN];
  192.         int buflen = sizeof(buf),len;
  193.         MQTTString topicString = MQTTString_initializer;
  194.       uint16_t packid = 0,packetidbk=0;
  195.     
  196.         //填充主题
  197.       topicString.cstring = (char *)topic;
  198.       //填充数据包ID
  199.       if((qos == QOS1)||(qos == QOS2))
  200.         { 
  201.             packid = GetNextPackID();
  202.         }
  203.         else
  204.         {
  205.               qos = QOS0;
  206.               retained = 0;
  207.               packid = 0;
  208.         }
  209.         
  210.         //推送消息
  211.         len = MQTTSerialize_publish(buf, buflen, 0, qos, retained, packid, topicString, (unsigned char*)msg, msg_len);
  212.         if(len <= 0)
  213.                 return -1;
  214.         if(transport_sendPacketBuffer(buf, len) < 0)    
  215.                 return -2;    
  216.         
  217.         //质量等级0，不需要返回
  218.         if(qos == QOS0)
  219.         {
  220.                 return 0;
  221.         }
  222.         
  223.         //等级1
  224.         if(qos == QOS1)
  225.         {
  226.                 //等待PUBACK
  227.               if(WaitForPacket(PUBACK,5) < 0)
  228.                      return -3;
  229.                 return 1;
  230.               
  231.         }
  232.         //等级2
  233.         if(qos == QOS2)    
  234.         {
  235.               //等待PUBREC
  236.               if(WaitForPacket(PUBREC,5) < 0)
  237.                      return -3;
  238.               //发送PUBREL
  239.               len = MQTTSerialize_pubrel(buf, buflen,0, packetidbk);
  240.                 if(len == 0)
  241.                     return -4;
  242.                 if(transport_sendPacketBuffer(buf, len) < 0)    
  243.                     return -6;            
  244.               //等待PUBCOMP
  245.               if(WaitForPacket(PUBREC,5) < 0)
  246.                      return -7;
  247.                 return 2;
  248.         }
  249.         //等级错误
  250.         return -8;
  251. }
  252.  
  253.  
  254. int my_mqtt_send_pingreq(){
  255.     int len;
  256.     uint8_t buf[200];
  257.     int buflen = sizeof(buf);     
  258.     len = MQTTSerialize_pingreq(buf, buflen);
  259.     transport_sendPacketBuffer(buf, len);
  260.     if(gprsTcpAvailable(5000)<0)
  261.             return -1;
  262.     if(MQTTPacket_read(buf, buflen, transport_getdata) != PINGRESP)
  263.             return -2;
  264.     return 0;
  265. }
  266. int mqttHasDataIn(int waitTime){
  267.     return gprsTcpAvailable(waitTime);
  268. }
  269.  
  270.  
  271.  
  272.  
  273. 消息处理/
  274.  
  275.  
  276. /************************************************************************
  277. ** 函数名称: deliverMessage                        
  278. ** 函数功能: 接受服务器发来的消息
  279. ** 入口参数: MQTTMessage *msg:MQTT消息结构体
  280. **           MQTT_USER_MSG *mqtt_user_msg:用户接受结构体
  281. **           MQTTString  *TopicName:主题
  282. ** 出口参数: 无
  283. ** 备    注: 
  284. ************************************************************************/
  285. static void deliverMessage(MQTTString  *TopicName,MQTTMessage *msg,MQTT_USER_MSG *mqtt_user_msg)
  286. {
  287.         //消息质量
  288.         mqtt_user_msg->msgqos = msg->qos;
  289.         //保存消息
  290.         memcpy(mqtt_user_msg->msg,msg->payload,msg->payloadlen);
  291.         mqtt_user_msg->msg[msg->payloadlen] = '\0';
  292.         //保存消息长度
  293.         mqtt_user_msg->msglenth = msg->payloadlen;
  294.         //消息主题
  295.         memcpy((char *)mqtt_user_msg->topic,TopicName->lenstring.data,TopicName->lenstring.len);
  296.         mqtt_user_msg->topic[TopicName->lenstring.len] = 0;
  297.         //消息ID
  298.         mqtt_user_msg->packetid = msg->id;
  299.         //标明消息合法
  300.         mqtt_user_msg->valid = 1;        
  301. }
  302.  
  303. /************************************************************************
  304. ** 函数名称: UserMsgCtl                        
  305. ** 函数功能: 用户消息处理函数
  306. ** 入口参数: MQTT_USER_MSG  *msg：消息结构体指针
  307. ** 出口参数: 无
  308. ** 备    注: 
  309. ************************************************************************/
  310. void UserMsgCtl(MQTT_USER_MSG  *msg)
  311. {
  312.       
  313.  
  314.     //这里处理数据只是打印，用户可以在这里添加自己的处理方式
  315.       printf("MQTT>>****收到客户端自己订阅的消息！！****\n");
  316.         //返回后处理消息
  317.       switch(msg->msgqos)
  318.         {
  319.             case 0:
  320.                     printf("MQTT>>消息质量：QoS0\n");
  321.                     break;
  322.             case 1:
  323.                     printf("MQTT>>消息质量：QoS1\n");
  324.                     break;
  325.             case 2:
  326.                     printf("MQTT>>消息质量：QoS2\n");
  327.                     break;
  328.             default:
  329.                     printf("MQTT>>错误的消息质量\n");
  330.                     break;
  331.         }
  332.       printf("MQTT>>消息主题：%s\n",msg->topic);    
  333.       printf("MQTT>>消息类容：%s\n",msg->msg);    
  334.       printf("MQTT>>消息长度：%d\n",msg->msglenth);
  335.       //处理完后销毁数据
  336.       msg->valid  = 0;
  337.  
  338.     char MqttSendbuf[512]={0};
  339.     static int timeCount=0;
  340.     sprintf(MqttSendbuf,"Hello server %d",timeCount++);
  341.     MQTTMsgPublish("mqttSendTopic",QOS0, 0,(uint8_t *)MqttSendbuf,strlen(MqttSendbuf));
  342. }
  343.  
  344.  
  345.  
  346. /************************************************************************
  347. ** 函数名称: mqtt_pktype_ctl                        
  348. ** 函数功能: 根据包类型进行处理
  349. ** 入口参数: uint8_t packtype:包类型
  350. ** 出口参数: 无
  351. ** 备    注: 
  352. ************************************************************************/
  353. void mqtt_pktype_ctl(uint8_t packtype,uint8_t *buf,uint32_t buflen)
  354. {
  355.         MQTTMessage msg;
  356.         int rc;
  357.         MQTTString receivedTopic;
  358.         uint32_t len;
  359.         MQTT_USER_MSG  mqtt_user_msg;
  360.         switch(packtype)
  361.         {
  362.             case PUBLISH:
  363.                                         //拆析PUBLISH消息
  364.                                         if(MQTTDeserialize_publish(&msg.dup,(int*)&msg.qos, &msg.retained, &msg.id, &receivedTopic,(unsigned char **)&msg.payload, &msg.payloadlen, buf, buflen) != 1)
  365.                                                 return;    
  366.                                         //接受消息
  367.                                         deliverMessage(&receivedTopic,&msg,&mqtt_user_msg);
  368.                                         
  369.                                         //消息质量不同，处理不同
  370.                                         if(msg.qos == QOS0)
  371.                                         {
  372.                                              //QOS0-不需要ACK
  373.                                              //直接处理数据
  374.                                              UserMsgCtl(&mqtt_user_msg);
  375.                                              return;
  376.                                         }
  377.                                         //发送PUBACK消息
  378.                                         if(msg.qos == QOS1)
  379.                                         {
  380.                                                 len =MQTTSerialize_puback(buf,buflen,mqtt_user_msg.packetid);
  381.                                                 if(len == 0)
  382.                                                     return;
  383.                                                 //发送返回
  384.                                                 if(transport_sendPacketBuffer(buf,len)<0)
  385.                                                      return;    
  386.                                                 //返回后处理消息
  387.                                                 UserMsgCtl(&mqtt_user_msg); 
  388.                                                 return;                                                
  389.                                         }
  390.                 
  391.                                         //对于质量2,只需要发送PUBREC就可以了
  392.                                         if(msg.qos == QOS2)
  393.                                         {
  394.                                              len = MQTTSerialize_ack(buf, buflen, PUBREC, 0, mqtt_user_msg.packetid);                            
  395.                                              if(len == 0)
  396.                                                  return;
  397.                                              //发送返回
  398.                                              transport_sendPacketBuffer(buf,len);    
  399.                                         }        
  400.                             break;
  401.             case  PUBREL:                           
  402.                                         //解析包数据，必须包ID相同才可以
  403.                                         rc = MQTTDeserialize_ack(&msg.type,&msg.dup, &msg.id, buf,buflen);
  404.                                         if((rc != 1)||(msg.type != PUBREL)||(msg.id != mqtt_user_msg.packetid))
  405.                                             return ;
  406.                                         //收到PUBREL，需要处理并抛弃数据
  407.                                         if(mqtt_user_msg.valid == 1)
  408.                                         {
  409.                                              //返回后处理消息
  410.                                              UserMsgCtl(&mqtt_user_msg);
  411.                                         }      
  412.                                         //串行化PUBCMP消息
  413.                                         len = MQTTSerialize_pubcomp(buf,buflen,msg.id);                           
  414.                                         if(len == 0)
  415.                                             return;                                    
  416.                                         //发送返回--PUBCOMP
  417.                                         transport_sendPacketBuffer(buf,len);                                        
  418.                         break;
  419.             case   PUBACK://等级1客户端推送数据后，服务器返回
  420.                             break;
  421.             case   PUBREC://等级2客户端推送数据后，服务器返回
  422.                             break;
  423.             case   PUBCOMP://等级2客户端推送PUBREL后，服务器返回
  424.                             break;
  425.             default:
  426.                            break;
  427.         }
  428. }

  1. 创建MQTT 任务

  1. #include <stdint.h>
  2. #include "freertos/FreeRTOS.h"
  3. #include "freertos/task.h"
  4. #include "string.h"
  5. #include "freertos/queue.h"
  6. #include "malloc.h"
  7.  
  8. #include "taskMqttClient.h"
  9. #include "../MQTT/transport.h"
  10. #include "../MQTT/MQTTPacket.h"
  11. #include "../MqttUtils/MqttUtils.h"
  12. #include "../GprsUtils/GprsUtils.h"
  13.  
  14.  
  15. #define HOST_NAME "183.220.123.35"
  16. #define HOST_PORT 17788
  17. #define USER_NAME     "jdtf1"
  18. #define USER_PASSWD  "jdtf1"
  19.  
  20.  
  21.  
  22. #define MQTT_RECV_TOPIC         "mqttRecvTopic"
  23. #define MQTT_SEND_TOPIC            "mqttSendTopic"
  24.  
  25.  
  26.  
  27.  
  28.  
  29.  
  30.  
  31.  
  32. #define MQTT_RECV_TASK_STACKSIZE                    (configMINIMAL_STACK_SIZE*8)    
  33. #define MQTT_RECV_TASK_PRIO                            4
  34. static TaskHandle_t                                 MqttRecvTask_Handler;
  35. static void mqttRecvThread(void *pvParameters);
  36.  
  37. #define MQTT_SEND_TASK_STACKSIZE                    (configMINIMAL_STACK_SIZE*8)    
  38. #define MQTT_SEND_TASK_PRIO                            5
  39. static TaskHandle_t                                 MqttSendTask_Handler;
  40. static void mqttSendThread(void *pvParameters);
  41.  
  42.  
  43. QueueHandle_t MQTT_Data_Queue =NULL;
  44. #define  MQTT_QUEUE_LEN    10   /* 队列的长度，最大可包含多少个消息 */
  45. #define  MQTT_QUEUE_SIZE   sizeof(MQTT_MSG_ITEM )   /* 队列中每个消息大小（字节） */
  46. void mqtt_thread_init(void){
  47.     /* 创建Test_Queue */
  48.     MQTT_Data_Queue = xQueueCreate((UBaseType_t ) MQTT_QUEUE_LEN,(UBaseType_t ) MQTT_QUEUE_SIZE);/* 消息的大小 */
  49.     if(NULL != MQTT_Data_Queue)
  50.         printf("The MQTT_Data_Queue was created successfully!\r\n");
  51.     xTaskCreate(mqttRecvThread, (const char* ) "MQTTRecvTask", MQTT_RECV_TASK_STACKSIZE, NULL, MQTT_RECV_TASK_PRIO, (TaskHandle_t*)&MqttRecvTask_Handler);
  52.     xTaskCreate(mqttSendThread, (const char* ) "MQTTSendTask", MQTT_SEND_TASK_STACKSIZE, NULL, MQTT_SEND_TASK_PRIO, (TaskHandle_t*)&MqttSendTask_Handler);
  53. }
  54.  
  55.  
  56. static void mqttSendThread(void *pvParameters){
  57.     /* 定义一个创建信息返回值，默认为pdTRUE */
  58.     BaseType_t xReturn = pdTRUE;
  59.     /* 定义一个接收消息的变量 */
  60.     MQTT_MSG_ITEM recvMqttMsg;
  61.     while(1){
  62.       xReturn = xQueueReceive(MQTT_Data_Queue, &recvMqttMsg,portMAX_DELAY); /* 等待时间 3000ms */
  63.       if(xReturn == pdTRUE){
  64.           printf("mqttSendThread: msg=%s len=%d\r\n",recvMqttMsg.msgStr,recvMqttMsg.msgLength);
  65.           MQTTMsgPublish((char*)MQTT_SEND_TOPIC,QOS0, 0,(uint8_t *)recvMqttMsg.msgStr,recvMqttMsg.msgLength);
  66.           free(recvMqttMsg.msgStr);
  67.       }
  68.     }
  69. }
  70.  
  71.  
  72. #define MAX_TRY_CONNECT_TIMES    15
  73. #define MAX_PING_FAILED_TIMES    3
  74. static void mqttRecvThread(void *pvParameters){
  75.       
  76.         uint32_t curtick=0;
  77.         uint8_t no_mqtt_msg_exchange = 1;
  78.         int sessionPresent = 0;
  79.         uint8_t buf[MSG_MAX_LEN];
  80.         int buflen = sizeof(buf),type;    
  81.         int isMqttServerConnected=0;
  82.         int tryConnectServerFailedTime=0;
  83.         int tryPingServerFFailedTimes=0;
  84.         gprsInit(GPRSPORT,115200);
  85.         gprsReboot();
  86.         //获取当前滴答，作为心跳包起始时间
  87.         curtick = xTaskGetTickCount();
  88.         //无线循环
  89.         printf("MQTT>>4.开始循环接收订阅的消息...\n");
  90.         while(1)
  91.         {
  92.  
  93.                 if(!isMqttServerConnected){
  94.                     printf("MQTT>>1.开始创建网络连接...\n");
  95.                     if((sessionPresent=mqttConnnectServer(HOST_NAME,HOST_PORT,USER_NAME,USER_PASSWD,KEEPLIVE_TIME))<0){
  96.                         vTaskDelay(3000/portTICK_RATE_MS);
  97.                         if(tryConnectServerFailedTime++>MAX_TRY_CONNECT_TIMES){
  98.                             tryConnectServerFailedTime=0;
  99.                             gprsReboot();
  100.                             printf("Try max %d time to connect server:%s port:%d failed,then reboot 4G model\n",MAX_TRY_CONNECT_TIMES,HOST_NAME,HOST_PORT);
  101.                         }
  102.                         continue;
  103.                     }
  104.                     printf("连接MQTT 服务器成功 ...\n");
  105.                     isMqttServerConnected=1;
  106.                     tryConnectServerFailedTime=0;
  107.                     if(sessionPresent!=1){    //订阅消息
  108.                         if(MQTTSubscribe((char*)MQTT_RECV_TOPIC,QOS0) < 0)
  109.                         {
  110.                                 //重连服务器
  111.                                 printf("MQTT>>客户端订阅消息失败...\n");
  112.                                 transport_close();
  113.                                 isMqttServerConnected=0;
  114.                                 continue;
  115.                         }else
  116.                                 printf("订阅主题 %s 成功 \n",MQTT_RECV_TOPIC);
  117.                     }
  118.                     curtick = xTaskGetTickCount();    
  119.                 }
  120.                 //表明无数据交换
  121.                 no_mqtt_msg_exchange = 1;
  122.                 //判断MQTT服务器是否有数据
  123.                 if((type=ReadPacketTimeout(buf,buflen,1000))!=-1)
  124.                 {    
  125.                     mqtt_pktype_ctl(type,buf,buflen);
  126.                     //表明有数据交换
  127.                     no_mqtt_msg_exchange = 0;
  128.                     //获取当前滴答，作为心跳包起始时间
  129.                     curtick = xTaskGetTickCount();
  130.                         
  131.                 }
  132.                 
  133.             //这里主要目的是定时向服务器发送PING保活命令
  134.              if((xTaskGetTickCount() - curtick) >(pdMS_TO_TICKS(KEEPLIVE_TIME)))
  135.             {
  136.                 curtick = xTaskGetTickCount();
  137.                 //判断是否有数据交换
  138.                 if(no_mqtt_msg_exchange == 0)
  139.                 {
  140.                     //如果有数据交换，这次就不需要发送PING消息
  141.                     continue;
  142.                 }
  143.                     
  144.                 if(my_mqtt_send_pingreq() < 0){
  145.                          //重连服务器
  146.                          printf("MQTT>>发送ping失败....\n");
  147.                          if(tryPingServerFFailedTimes++>=MAX_PING_FAILED_TIMES){
  148.                             tryPingServerFFailedTimes=0;
  149.                              transport_close();
  150.                              isMqttServerConnected=0;
  151.                          }
  152.                          continue;     
  153.                 }
  154.                 tryPingServerFFailedTimes=0;
  155.                 printf("MQTT>>发送ping作为心跳成功....\n");
  156.                 //表明有数据交换
  157.                 no_mqtt_msg_exchange = 0;
  158.           }                      
  159.         }
  160. }

工程源码地址:https://download.csdn.net/download/du2005023029/87505817

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小桥流水78/article/detail/940767
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号