RyanMqtt移植指南

冬至子 2023-10-09 924

电子说

1.3w人已加入

描述

测试环境：stm32F401RCT6、RT-Thread版本: v4.1.0、RT-Thread Studio版本: 2.2.6、网络硬件使用ec800m移植at_socket使用sal框架。

1、移植介绍
RyanMqtt 库希望应用程序为以下接口提供实现：

system 接口
RyanMqtt 需要 RTOS 支持，必须实现如下接口才可以保证 mqtt 客户端的正常运行

STM32F401

network 接口
RyanMqtt 依赖于底层传输接口 API，必须实现该接口 API 才能在网络上发送和接收数据包

MQTT 协议要求基础传输层能够提供有序的、可靠的、双向传输（从客户端到服务端和从服务端到客户端）的字节流

STM32F401

time 接口
RyanMqtt 依靠函数生成毫秒时间戳，用于计算持续时间和超时，内部已经做了数值溢出处理

STM32F401

2、开始移植
得益于RT-Thread驱动应用层分离的思想和SAL框架，platform/rtthread的适配层可以适应任何RT-Thread代码，所以我们就不拿RT-Thread来移植了。

使用FreeRTOS内核来移植，使用CMSIS-RTOS V2兼容层。

system 接口
系统接口，需要移植RTOS的接口。为方便管理类型使用平台结构体，修改platformSystem.h里面的结构体
就是线程和互斥锁

typedef struct
{
    osThreadId_t thread;
} platformThread_t;
typedef struct
{
    osMutexId_t mutex;
} platformMutex_t;

再来实现platformSystem.c里面的函数定义
注意里面的 platformPrint 函数，由于FreeRTOS没有官方的打印接口。记得修改为你的打印接口

#include "platformSystem.h"
/**

@brief 申请内存
@param size
@return void*
/
void platformMemoryMalloc(size_t size)
{
return pvPortMalloc(size);
}
/
@brief 释放内存
@param ptr
/
void platformMemoryFree(void ptr)
{
vPortFree(ptr);
}
/
@brief ms延时
@param ms
/
void platformDelay(uint32_t ms)
{
osDelay(ms);
}
/ *
@brief 打印字符串函数,可通过串口打印出去
@param str
@param strLen
/
void platformPrint(char str, uint16_t strLen)
{
}
/
@brief 初始化并运行线程
@param userData
@param platformThread
@param name
@param entry
@param param
@param stackSize
@param priority
@return RyanMqttError_e
*/
RyanMqttError_e platformThreadInit(void *userData,
platformThread_t *platformThread,
const char *name,
void (*entry)(void ),
void const param,
uint32_t stackSize,
uint32_t priority)
{
const osThreadAttr_t myTask02_attributes = {
.name = name,
.stack_size = stackSize,
.priority = (osPriority_t)priority,
};
platformThread->thread = osThreadNew(entry, param, &myTask02_attributes);
if (NULL == platformThread->thread)
return RyanMqttNoRescourceError;
return RyanMqttSuccessError;
}
/
@brief 销毁自身线程
@param userData
@param platformThread
@return RyanMqttError_e
*/
RyanMqttError_e platformThreadDestroy(void userData, platformThread_t platformThread)
{
osThreadTerminate(platformThread->thread);
return RyanMqttSuccessError;
}
/
@brief 开启线程
@param userData
@param platformThread
@return RyanMqttError_e
*/
RyanMqttError_e platformThreadStart(void userData, platformThread_t platformThread)
{
osThreadResume(platformThread->thread);
return RyanMqttSuccessError;
}
/
@brief 挂起线程
@param userData
@param platformThread
@return RyanMqttError_e
*/
RyanMqttError_e platformThreadStop(void userData, platformThread_t platformThread)
{
osThreadSuspend(platformThread->thread);
return RyanMqttSuccessError;
}
/
@brief 互斥锁初始化
@param userData
@param platformMutex
@return RyanMqttError_e
*/
RyanMqttError_e platformMutexInit(void userData, platformMutex_t platformMutex)
{
const osMutexAttr_t myMutex01_attributes = {
.name = "mqttMutex"};
platformMutex->mutex = osMutexNew(&myMutex01_attributes);
return RyanMqttSuccessError;
}
/
@brief 销毁互斥锁
@param userData
@param platformMutex
@return RyanMqttError_e
*/
RyanMqttError_e platformMutexDestroy(void userData, platformMutex_t platformMutex)
{
osMutexDelete(platformMutex->mutex);
return RyanMqttSuccessError;
}
/
@brief 阻塞获取互斥锁
@param userData
@param platformMutex
@return RyanMqttError_e
*/
RyanMqttError_e platformMutexLock(void userData, platformMutex_t platformMutex)
{
osMutexAcquire(platformMutex->mutex, osWaitForever);
return RyanMqttSuccessError;
}
/
@brief 释放互斥锁
@param userData
@param platformMutex
@return RyanMqttError_e
*/
RyanMqttError_e platformMutexUnLock(void userData, platformMutex_t platformMutex)
{
osMutexRelease(platformMutex->mutex);
return RyanMqttSuccessError;
}
/
@brief 进入临界区 / 关中断

/
void platformCriticalEnter(void)
{
osKernelLock();
}
/ *

@brief 退出临界区 / 开中断

*/
void platformCriticalExit(void)
{
osKernelUnlock();
}
time 接口
time接口，只需要提供一个ms时间戳就行，直接修改函数，这里使用FreeRTOS的心跳。

uint32_t platformUptimeMs(void)
{
if (1000 == osKernelGetTickFreq())
return (uint32_t)osKernelGetTickCount();
else
{
uint32_t tick = 0;
tick = osKernelGetTickCount() * 1000;
return (uint32_t)((tick + osKernelGetTickCount() - 1) / osKernelGetTickCount());
}
}
network 接口
MQTT 协议要求基础传输层能够提供有序的、可靠的、双向传输（从客户端到服务端和从服务端到客户端）的字节流
由于FreeRTOS没有规定标准的网络层，你可以选择 FreeRTOS-Plus-TCP / FreeRTOS-Cellular-Interface/ lwip / W5500等网络方法，几乎RT-Thread支持的你也可以在FreeRTOS仓库找到。

这里以lwip为例，使用socket接口来实现，网络阻塞发送和接收使用 SO_SNDTIMEO 和 SO_RCVTIMEO 来实现，你也可以选择select / poll / epoll等方式。

修改 platformNetwork_t 结构体以支持 socket

typedef struct
{
    int socket;
} platformNetwork_t;

接着实现platformNetwork.c里面的函数

#define rlogEnable 1 // 是否使能日志
#define rlogColorEnable 1 // 是否使能日志颜色
#define rlogLevel (rlogLvlWarning) // 日志打印等级
#define rlogTag "RyanMqttNet" // 日志tag
#include "platformNetwork.h"
#include "RyanMqttLog.h"
/**

@brief 连接mqtt服务器
@param userData
@param platformNetwork
@param host
@param port
@return RyanMqttError_e
成功返回RyanMqttSuccessError，失败返回错误信息
*/
RyanMqttError_e platformNetworkConnect(void *userData, platformNetwork_t *platformNetwork, const char *host, const char port)
{
RyanMqttError_e result = RyanMqttSuccessError;
struct addrinfo addrList = NULL;
struct addrinfo hints = {
.ai_family = AF_UNSPEC,
.ai_socktype = SOCK_STREAM,
.ai_protocol = IPPROTO_TCP};
if (getaddrinfo(host, port, &hints, &addrList) != 0)
{
result = RyanSocketFailedError;
goto exit;
}
platformNetwork->socket = socket(addrList->ai_family, addrList->ai_socktype, addrList->ai_protocol);
if (platformNetwork->socket < 0)
{
result = RyanSocketFailedError;
goto exit;
}
if (connect(platformNetwork->socket, addrList->ai_addr, addrList->ai_addrlen) != 0)
{
platformNetworkClose(userData, platformNetwork);
result = RyanMqttSocketConnectFailError;
goto exit;
}
exit:
if (NULL != addrList)
freeaddrinfo(addrList);
return result;
}
/
@brief 非阻塞接收数据
@param userData
@param platformNetwork
@param recvBuf
@param recvLen
@param timeout
@return RyanMqttError_e
socket错误返回 RyanSocketFailedError
接收超时或者接收数据长度不等于期待数据接受长度 RyanMqttRecvPacketTimeOutError
接收成功 RyanMqttSuccessError
*/
RyanMqttError_e platformNetworkRecvAsync(void *userData, platformNetwork_t *platformNetwork, char recvBuf, int recvLen, int timeout)
{
int32_t recvResult = 0;
int32_t offset = 0;
int32_t timeOut2 = timeout;
struct timeval tv = {0};
platformTimer_t timer = {0};
if (-1 == platformNetwork->socket)
return RyanSocketFailedError;
platformTimerCutdown(&timer, timeout);
while ((offset < recvLen) && (0 != timeOut2))
{
tv.tv_sec = timeOut2 / 1000;
tv.tv_usec = timeOut2 % 1000 * 1000;
if (tv.tv_sec <= 0 && tv.tv_usec <= 100)
{
tv.tv_sec = 0;
tv.tv_usec = 100;
}
setsockopt(platformNetwork->socket, SOL_SOCKET, SO_RCVTIMEO, (char )&tv, sizeof(struct timeval)); // 设置错做模式为非阻塞
recvResult = recv(platformNetwork->socket, recvBuf + offset, recvLen - offset, 0);
if (recvResult <= 0) // 小于零，表示错误，个别错误不代表socket错误
{
// 下列3种表示没问题,但需要推出发送
if ((errno == EAGAIN || // 套接字已标记为非阻塞，而接收操作被阻塞或者接收超时
errno == EWOULDBLOCK || // 发送时套接字发送缓冲区已满，或接收时套接字接收缓冲区为空
errno == EINTR)) // 操作被信号中断
break;
return RyanSocketFailedError;
}
offset += recvResult;
timeOut2 = platformTimerRemain(&timer);
}
if (offset != recvLen)
return RyanMqttRecvPacketTimeOutError;
return RyanMqttSuccessError;
}
/
@brief 非阻塞发送数据
@param userData
@param platformNetwork
@param sendBuf
@param sendLen
@param timeout
@return RyanMqttError_e
socket错误返回 RyanSocketFailedError
接收超时或者接收数据长度不等于期待数据接受长度 RyanMqttRecvPacketTimeOutError
接收成功 RyanMqttSuccessError
*/
RyanMqttError_e platformNetworkSendAsync(void *userData, platformNetwork_t *platformNetwork, char sendBuf, int sendLen, int timeout)
{
int32_t sendResult = 0;
int32_t offset = 0;
int32_t timeOut2 = timeout;
struct timeval tv = {0};
platformTimer_t timer = {0};
if (-1 == platformNetwork->socket)
return RyanSocketFailedError;
platformTimerCutdown(&timer, timeout);
while ((offset < sendLen) && (0 != timeOut2))
{
tv.tv_sec = timeOut2 / 1000;
tv.tv_usec = timeOut2 % 1000 * 1000;
if (tv.tv_sec <= 0 && tv.tv_usec <= 100)
{
tv.tv_sec = 0;
tv.tv_usec = 100;
}
setsockopt(platformNetwork->socket, SOL_SOCKET, SO_SNDTIMEO, (char )&tv, sizeof(struct timeval)); // 设置错做模式为非阻塞
sendResult = send(platformNetwork->socket, sendBuf + offset, sendLen - offset, 0);
if (sendResult <= 0) // 小于零，表示错误，个别错误不代表socket错误
{
// 下列3种表示没问题,但需要推出发送
if ((errno == EAGAIN || // 套接字已标记为非阻塞，而接收操作被阻塞或者接收超时
errno == EWOULDBLOCK || // 发送时套接字发送缓冲区已满，或接收时套接字接收缓冲区为空
errno == EINTR)) // 操作被信号中断
break;
return RyanSocketFailedError;
}
offset += sendResult;
timeOut2 = platformTimerRemain(&timer);
}
if (offset != sendLen)
return RyanMqttSendPacketTimeOutError;
return RyanMqttSuccessError;
}
/
@brief 断开mqtt服务器连接

@param userData
@param platformNetwork
@return RyanMqttError_e
*/
RyanMqttError_e platformNetworkClose(void *userData, platformNetwork_t *platformNetwork)
{
if (platformNetwork->socket >= 0)
{
closesocket(platformNetwork->socket);
platformNetwork->socket = -1;
}
return RyanMqttSuccessError;
}

3、总结
可以看到，RyanMqtt移植非常简单，有专门的platform层用来移植。

打开APP阅读更多精彩内容