阿里云物联网的设备监控与固件升级(OTA)实现

阿里云物联网的设备监控与固件升级(OTA)实现

阿里云物联网平台提供API接口（iotkit-Sdk开发包）方便设备接入其物联网平台，只要调用其函数接口就可以实现设备数据快速上网与云端操控。本文将就设备状态监控与固件升级展示物联网平台如何实现设备接入与维护的。

本文采用了阿里云V2.10的源代码开发包[https://github.com/aliyun/iotkit-embedded]，通过源代码编译了其静态库。

1、在本文案例中将实现对边缘设备内的某服务进行启动、停止的状态监控，和远程升级该服务。

首先看看如何实现服务的启动、停止以及状态查询，直接上代码：

win平台，CLogger为日志类，读者可用其他输出函数代替：

#include <windows.h>
#include <tchar.h>
#include <strsafe.h>
//#include <iostream>

#include "atlcomtime.h"
#pragma comment(lib, "advapi32.lib")
#include "Log.h"

VOID WINSVC::SvcQuery(char *svr, int &svc_state)
{
SC_HANDLE schSCManager;
SC_HANDLE schService;
// Get a handle to the SCM database.
schSCManager = OpenSCManager(
NULL, // local computer
NULL, // ServicesActive database
SC_MANAGER_ALL_ACCESS); // full access rights

if (NULL == schSCManager)
{
CLogger::createInstance()->Log(eSoftError, "Failed to open service manager (%d), %s %s %d!"
, GetLastError(), __FILE__, __FUNCTION__, __LINE__);
return;
}
// Get a handle to the service.
schService = OpenService(
schSCManager, // SCM database
svr, // name of service
SERVICE_QUERY_STATUS); // need query access

if (schService == NULL)
{
CLogger::createInstance()->Log(eSoftError, "Failed to get service(%s) and error(%d), %s %s %d!"
, svr, GetLastError(), __FILE__, __FUNCTION__, __LINE__);
CloseServiceHandle(schSCManager);
return;
}
//调用QueryServiceStatus函数
SERVICE_STATUS sStatus = { 0 };
if (!QueryServiceStatus(schService, &sStatus))
{
CloseServiceHandle(schService);
CloseServiceHandle(schSCManager);
return;
}
switch (sStatus.dwCurrentState)
{
case SERVICE_STOP_PENDING: case SERVICE_STOPPED:
svc_state = 1;
break;
case SERVICE_START_PENDING: case SERVICE_RUNNING: case SERVICE_CONTINUE_PENDING:
svc_state = 2;
break;
case SERVICE_PAUSE_PENDING: case SERVICE_PAUSED:
svc_state = 3;
break;
default:
svc_state = 0;
break;
}
CloseServiceHandle(schService);
CloseServiceHandle(schSCManager);
};

VOID WINSVC::SvcStart(char *svr)
{
SC_HANDLE schSCManager;
SC_HANDLE schService;
// Get a handle to the SCM database.
schSCManager = OpenSCManager(
NULL, // local computer
NULL, // ServicesActive database
SC_MANAGER_ALL_ACCESS); // full access rights

if (NULL == schSCManager)
{
CLogger::createInstance()->Log(eSoftError, "Failed to open service manager (%d), %s %s %d!"
, GetLastError(), __FILE__, __FUNCTION__, __LINE__);
return;
}
// Get a handle to the service.
schService = OpenService(
schSCManager, // SCM database
svr, // name of service
SERVICE_START | SERVICE_QUERY_STATUS); // need start and query access

if (schService == NULL)
{
CLogger::createInstance()->Log(eSoftError, "Failed to get service(%s) and error(%d), %s %s %d!"
, svr,GetLastError(), __FILE__, __FUNCTION__, __LINE__);
CloseServiceHandle(schSCManager);
return;
}
StartService(schService, 0, NULL);//开始Service
//调用QueryServiceStatus函数
SERVICE_STATUS sStatus = { 0 };
if (!QueryServiceStatus(schService, &sStatus))
{
CloseServiceHandle(schService);
CloseServiceHandle(schSCManager);
return;
}
if (SERVICE_RUNNING == sStatus.dwCurrentState || SERVICE_START_PENDING == sStatus.dwCurrentState)
{
CLogger::createInstance()->Log(eTipMessage, "start service(%s) success, %s %s %d!"
, svr, __FILE__, __FUNCTION__, __LINE__);
}
CloseServiceHandle(schService);
CloseServiceHandle(schSCManager);
};

VOID WINSVC::SvcStop(char *svr)
{
SC_HANDLE schSCManager;
SC_HANDLE schService;
// Get a handle to the SCM database.
schSCManager = OpenSCManager(
NULL, // local computer
NULL, // ServicesActive database
SC_MANAGER_ALL_ACCESS); // full access rights

if (NULL == schSCManager)
{
CLogger::createInstance()->Log(eSoftError, "Failed to open service manager(%d), %s %s %d!"
, GetLastError(), __FILE__, __FUNCTION__, __LINE__);
return;
}
// Get a handle to the service.
schService = OpenService(
schSCManager, // SCM database
svr, // name of service
SERVICE_STOP | SERVICE_QUERY_STATUS); // need stop or query access

if (schService == NULL)
{
CLogger::createInstance()->Log(eSoftError, "Failed to get service(%s) and error(%d), %s %s %d!"
, svr,GetLastError(), __FILE__, __FUNCTION__, __LINE__);
CloseServiceHandle(schSCManager);
return;
}
//调用QueryServiceStatus函数
SERVICE_STATUS sStatus = { 0 };
if (!QueryServiceStatus(schService, &sStatus))
{
CloseServiceHandle(schService);
CloseServiceHandle(schSCManager);
return;
}
if (SERVICE_RUNNING == sStatus.dwCurrentState || SERVICE_PAUSED == sStatus.dwCurrentState)
{
ControlService(schService, SERVICE_CONTROL_STOP, &sStatus);
}
if (!QueryServiceStatus(schService, &sStatus))
{
CloseServiceHandle(schService);
CloseServiceHandle(schSCManager);
return;
}
if (SERVICE_STOPPED == sStatus.dwCurrentState || SERVICE_STOP_PENDING == sStatus.dwCurrentState)
{
CLogger::createInstance()->Log(eTipMessage, "stop service(%s) success, %s %s %d!"
, svr, __FILE__, __FUNCTION__, __LINE__);
}
CloseServiceHandle(schService);
CloseServiceHandle(schSCManager);
};
linux平台：

#include <string.h>
#include <string>
#include <stdlib.h>
#include <stdio.h>
#include <signal.h>

void SvcStart(char *svr)
{
FILE* fp = NULL;
char command[128] = { 0 };
sprintf(command, "systemctl start %s", svr);
if ((fp = popen(command, "r")) == NULL)
{
return;
}
char buf[512] = { 0 };
if ((fgets(buf, 512, fp)) == NULL)

{
pclose(fp);
return;
}
printf("ret:%s ",buf);
pclose(fp);
};

void SvcStop(char *svr)
{
FILE* fp = NULL;
char command[128] = { 0 };
sprintf(command, "systemctl stop %s", svr);
if ((fp = popen(command, "r")) == NULL)
{
return;
}
char buf[512] = { 0 };
if ((fgets(buf, 512, fp)) == NULL)
{
pclose(fp);
return;
}
printf("ret:%s ",buf);
pclose(fp);
};

void SvcQuery(char* svr, int &svc_state)
{
svc_state = 0;
FILE* fp = NULL;
char command[128] = { 0 };
sprintf(command, "systemctl status %s | grep Active", svr);
if ((fp = popen(command, "r")) == NULL)
{
return;
}
char buf[512] = { 0 };
if ((fgets(buf, 512, fp)) == NULL)
{
pclose(fp);
return;
}
std::string comment = std::string(buf,strlen(buf));
//std::string::size_type _pos = comment.find("running");
//开机启动的状态,static不可被管理,disable未启动,enable启动
//dead关闭,exited已读取完系统配置,闲置 waiting等待, running正在进行, mounted挂载, plugged加载插件, failed系统配置错误
if(std::string::npos != comment.find("running"))
{
svc_state = 2;
}else if(std::string::npos != comment.find("listening")){
svc_state = 2;
}else{
svc_state = 1;
}
printf("ret:%s,state:%d ",buf,svc_state);
pclose(fp);
};
2、假定已经编译好了阿里云物联网平台的SDK包，注意包含OTA模块的编译，下来通过IOTKIT_SDK包实现将连接阿里云物联网平台。

首先要做的去阿里云物联网平台创建产品，

以及创建设备实例：

并记录三元组信息：ProductKey，DeviceName，DeviceSecret

定义设备的三元组信息以及IOT相关信息的代码：

#include "iot_import.h"
#include "iot_export.h"

#define PAYLOAD_FORMAT "{"id":"%lld","version":"1.0","method":"%s","params":%s}"
#define ALINK_METHOD_PROP_POST "thing.event.property.post"
#define ALINK_METHOD_EVENT_POST "thing.event.ControlFail.post"

#define ALINK_COMMENT_FORMAT "clientId%s&%sdeviceName%sproductKey%stimestamp%lld"
#define ALINK_TOPIC_DEV_LOGIN "/ext/session/%s/%s/combine/login"
#define ALINK_TOPIC_DEV_LOGIN_REPLY "/ext/session/%s/%s/combine/login_reply"
#define ALINK_TOPIC_EVENT_PRO_POST "/sys/%s/%s/thing/event/property/post"
#define ALINK_TOPIC_EVENT_PRO_POST_REPLY "/sys/%s/%s/thing/event/property/post_reply"
#define ALINK_TOPIC_SERVICE_PRO_SET "/sys/%s/%s/thing/service/property/set"

#define login_fomat
"{"id":"%lld","params":{ "productKey":"%s","deviceName":"%s","clientId":"%s&%s","timestamp":"%lld","signMethod":"hmacSha1","sign":"%s","cleanSession":"true"}}"
#define add_fomat
"{"id":"%lld","version":"1.0","params":[{"deviceName":"%s","productKey":"%s","sign":"%s","signmethod":"hmacSha1","timestamp":"%lld","clientId":"%d"}],"method":"thing.topo.add"}"
#define MQTT_MSGLEN (2048)

#define OTA_BUF_LEN (4096)

struct AliyunTriples
{
AliyunTriples()
: product_key("")
, product_secret("")
, device_name("")
, device_secret("")
{};
bool invalid()
{
return (product_key.empty()
|| product_secret.empty()
|| device_name.empty()
|| device_secret.empty());
};
std::string product_key; //产品key
std::string product_secret; //产品密钥
std::string device_name; //设备名
std::string device_secret; //设备密钥
};

#define EXAMPLE_TRACE(fmt, ...)
do {
HAL_Printf("%s|%03d :: ", __func__, __LINE__);
HAL_Printf(fmt, ##__VA_ARGS__);
HAL_Printf("%s", " ");
} while(0)
然后创建于物联网平台的链接对象，建立连接并上报版本号，如下实例所示：

//事件处理函数
void event_handle(void *pcontext, void *pclient, iotx_mqtt_event_msg_pt msg)
{
uintptr_t packet_id = (uintptr_t)msg->msg;
iotx_mqtt_topic_info_pt topic_info = (iotx_mqtt_topic_info_pt)msg->msg;

switch (msg->event_type) {
case IOTX_MQTT_EVENT_UNDEF:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: undefined event occur."
, __func__, __LINE__);
break;

case IOTX_MQTT_EVENT_DISCONNECT:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: MQTT disconnect."
, __func__, __LINE__);
break;

case IOTX_MQTT_EVENT_RECONNECT:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: MQTT reconnect."
, __func__, __LINE__);
break;

case IOTX_MQTT_EVENT_SUBCRIBE_SUCCESS:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: subscribe success, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_SUBCRIBE_TIMEOUT:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: subscribe wait ack timeout, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_SUBCRIBE_NACK:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: subscribe nack, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_UNSUBCRIBE_SUCCESS:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: unsubscribe success, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_UNSUBCRIBE_TIMEOUT:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: unsubscribe timeout, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_UNSUBCRIBE_NACK:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: unsubscribe nack, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_PUBLISH_SUCCESS:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: publish success, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_PUBLISH_TIMEOUT:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: publish timeout, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_PUBLISH_NACK:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: publish nack, packet-id=%u"
, __func__, __LINE__, (unsigned int)packet_id);
break;

case IOTX_MQTT_EVENT_PUBLISH_RECVEIVED:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: topic message arrived but without any related handle: topic=%.*s, topic_msg=%.*s",
__func__, __LINE__,
topic_info->topic_len,
topic_info->ptopic,
topic_info->payload_len,
topic_info->payload);
break;

case IOTX_MQTT_EVENT_BUFFER_OVERFLOW:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: buffer overflow, %s"
, __func__, __LINE__, msg->msg);
break;

default:
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: Should NOT arrive here."
, __func__, __LINE__);
break;
}
}

void IOToMqttAliyun::init()
{
//
if (NULL == (msg_buf = (char *)HAL_Malloc(MQTT_MSGLEN))) {
CLogger::createInstance()->Log(eTipMessage, "%s|%03d :: not enough memory"
, __func__, __LINE__);
}

if (NULL == (msg_readbuf = (char *)HAL_Malloc(MQTT_MSGLEN))) {
CLogger::createInstance()->Log(eTipMessage, "%s|%03d :: not enough memory"
, __func__, __LINE__);
}
IOT_OpenLog("mqtt");
IOT_SetLogLevel(IOT_LOG_DEBUG);

HAL_SetProductKey((char*)gatewayTriples.product_key.c_str());
HAL_SetProductSecret((char*)gatewayTriples.product_secret.c_str());
HAL_SetDeviceName((char*)gatewayTriples.device_name.c_str());
HAL_SetDeviceSecret((char*)gatewayTriples.device_secret.c_str());
}
void IOToMqttAliyun::uninit()
{
if (NULL != msg_buf) {
HAL_Free(msg_buf);
}

if (NULL != msg_readbuf) {
HAL_Free(msg_readbuf);
}
IOT_DumpMemoryStats(IOT_LOG_DEBUG);
IOT_CloseLog();
};

void IOToMqttAliyun::destroy()
{
if (NULL != h_ota) {
IOT_OTA_Deinit(h_ota);
}
if (NULL != pclient) {
IOT_MQTT_Destroy(&pclient);
}
}
//初始化、链接、上报
void IOToMqttAliyun::create()
{
iotx_conn_info_pt pconn_info;
iotx_mqtt_param_t mqtt_params;

/* Device AUTH */
//阿里云物联网平台的三元组信息：产品key，设备名，设备密钥
if (0 != IOT_SetupConnInfo(gatewayTriples.product_key.c_str()
, gatewayTriples.device_name.c_str()
, gatewayTriples.device_secret.c_str()
, (void **)&pconn_info))
{
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: AUTH request failed!"
, __func__, __LINE__);
HAL_SleepMs(10000);
return;
}
/* Initialize MQTT parameter */
memset(&mqtt_params, 0x0, sizeof(mqtt_params));

mqtt_params.port = pconn_info->port;
mqtt_params.host = pconn_info->host_name;
mqtt_params.client_id = pconn_info->client_id;
mqtt_params.username = pconn_info->username;
mqtt_params.password = pconn_info->password;
mqtt_params.pub_key = pconn_info->pub_key;

mqtt_params.request_timeout_ms = 2000;
mqtt_params.clean_session = 0;
mqtt_params.keepalive_interval_ms = 300000;
//
mqtt_params.pread_buf = msg_readbuf;
mqtt_params.read_buf_size = MQTT_MSGLEN;
mqtt_params.pwrite_buf = msg_buf;
mqtt_params.write_buf_size = MQTT_MSGLEN;

mqtt_params.handle_event.h_fp = event_handle;//事件处理函数
mqtt_params.handle_event.pcontext = NULL;

/* Construct a MQTT client with specify parameter */
pclient = IOT_MQTT_Construct(&mqtt_params);
if (NULL == pclient) {
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: MQTT construct failed"
, __func__, __LINE__);
HAL_SleepMs(10000);
}
//OTA初始化并上报版本号
//void *h_ota = NULL;
h_ota = IOT_OTA_Init(gatewayTriples.product_key.c_str()
, gatewayTriples.device_name.c_str(), pclient);
if (NULL == h_ota) {
EXAMPLE_TRACE("initialize OTA failed");
}
#ifdef WIN32
char version_def[128] = "iotx_win_ver_1.0.0";
#else
char version_def[128] = "iotx_linux_ver_1.0.0";
#endif
getVersion(version_def);
if (0 != IOT_OTA_ReportVersion(h_ota, version_def)) {
EXAMPLE_TRACE("report OTA version failed");
}
HAL_SleepMs(1000);
}
2）需要监控维护的服务配置及代码定义如下：

struct ServiceInfo
{
ServiceInfo()
: svc_name("")
, app_dir("")
, app_name("")
, aliyun_key("")
, upLoopTime(10)
, dll_lib(false)
{};
ServiceInfo(const ServiceInfo& rval)
{
svc_name = rval.svc_name;
app_dir = rval.app_dir;
app_name = rval.app_name;
aliyun_key = rval.aliyun_key;
upLoopTime = rval.upLoopTime;
dll_lib = rval.dll_lib;
};
ServiceInfo& operator=(const ServiceInfo &rval)
{
if (this != &rval) {
svc_name = rval.svc_name;
app_dir = rval.app_dir;
app_name = rval.app_name;
aliyun_key = rval.aliyun_key;
upLoopTime = rval.upLoopTime;
dll_lib = rval.dll_lib;
}
return *this;
};

std::string svc_name;
std::string app_dir;
std::string app_name;
std::string aliyun_key;
unsigned int upLoopTime;
bool dll_lib;
};

enum TopicType
{
TOPIC_DEV_LOGIN = 1, //子设备上线 Topic
TOPIC_DEV_LOGIN_REPLY = 2, //子设备上线 Topic_Reply
TOPIC_EVENT_PRO_POST = 3, //客户端上送数据 Topic
TOPIC_EVENT_PRO_POST_REPLY = 4, //客户端上送数据 Topic_Reply
TOPIC_SERVICE_PRO_SET = 5, //服务设置属性Topic
TOPIC_SERVICE_PRO_SET_REPLY = 6,
TOPIC_DEFAULT = 0
};

struct AliyunServiceDesc
{
TopicType topicType;
std::map<std::string,ServiceInfo> triples;//map<功能标识符,服务配置>
};
topic订购与回调函数实现，来自阿里云物联网平台的设值，将内容接受写入单体类（CacheAliyunMQTT）的缓存队列中：

std::map<std::string, AliyunServiceDesc> subTopicMaps;//类变量
void IOToMqttAliyun::subInit()
{
//other code
......
//sub topic init
ptr_CacheDataObj->getSvcInfo(svcTriples);//获取服务配置信息(svc.xml)
{
AliyunServiceDesc topicMapDesc;
topicMapDesc.topicType = TOPIC_SERVICE_PRO_SET;
for (std::map<int, SvcDesc>::iterator it = svcTriples.begin(); it != svcTriples.end(); ++it)
{
ServiceInfo svcinfo;
svcinfo.svc_name = it->second.svc_name;
svcinfo.app_dir = it->second.app_dir;
svcinfo.app_name = it->second.app_name;
topicMapDesc.triples[it->second.aliyun_key] = svcinfo;
}
char buf[128] = { 0 };
sprintf(buf, ALINK_TOPIC_SERVICE_PRO_SET
, gatewayTriples.product_key.c_str()
, gatewayTriples.device_name.c_str());
std::string alink_topic_service_pro_set = std::string(buf, strlen(buf));
subTopicMaps[alink_topic_service_pro_set] = topicMapDesc;
sprintf(buf, ALINK_TOPIC_EVENT_PRO_POST_REPLY
, gatewayTriples.product_key.c_str()
, gatewayTriples.device_name.c_str());
std::string alink_topic_service_pro_post = std::string(buf, strlen(buf));
subTopicMaps[alink_topic_service_pro_post] = topicMapDesc;
}
}
////////////////////////////////////////////////////////////////////////////////
void IOToMqttAliyun::subscribe()
{
int rc = 0;
for (std::map<std::string, AliyunServiceDesc>::iterator it = subTopicMaps.begin();
it != subTopicMaps.end(); ++it)
{
switch (it->second.topicType)
{
case TOPIC_EVENT_PRO_POST_REPLY:
rc = IOT_MQTT_Subscribe(pclient, it->first.c_str(), IOTX_MQTT_QOS0, push_reply_message_arrive, NULL);
break;
case TOPIC_SERVICE_PRO_SET:
rc = IOT_MQTT_Subscribe(pclient, it->first.c_str(), IOTX_MQTT_QOS0, service_set_message_arrive, NULL);
break;
default:
continue;
}
if (rc < 0) {
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: IOT_MQTT_Subscribe() failed, rc = %d"
, __func__, __LINE__, it->first.c_str(), rc);
}
else {
IOT_MQTT_Yield(pclient, 200);
printf("IOT_MQTT_Subscribe(%s) success! ", it->first.c_str());
}
}
}

void push_reply_message_arrive(void *pcontext, void *pclient, iotx_mqtt_event_msg_pt msg)
{
iotx_mqtt_topic_info_pt ptopic_info = (iotx_mqtt_topic_info_pt)msg->msg;

/* print topic name and topic message */
EXAMPLE_TRACE("--2--");
EXAMPLE_TRACE("packetId: %d", ptopic_info->packet_id);
EXAMPLE_TRACE("Topic: '%.*s' (Length: %d)",
ptopic_info->topic_len,
ptopic_info->ptopic,
ptopic_info->topic_len);
EXAMPLE_TRACE("Payload: '%.*s' (Length: %d)",
ptopic_info->payload_len,
ptopic_info->payload,
ptopic_info->payload_len);
EXAMPLE_TRACE("--2--");
};
//来自阿里云物联网平台的设值，将内容接受写入单体类（CacheAliyunMQTT）的缓存队列中
void service_set_message_arrive(void *pcontext, void *pclient, iotx_mqtt_event_msg_pt msg)
{
iotx_mqtt_topic_info_pt ptopic_info = (iotx_mqtt_topic_info_pt)msg->msg;
CacheAliyunMQTT *ptr_CacheAliyunMQTT = CacheAliyunMQTT::getInstance();
RCacheAliyun retDatas(std::string(ptopic_info->ptopic, ptopic_info->topic_len)
, std::string(ptopic_info->payload, ptopic_info->payload_len));
ptr_CacheAliyunMQTT->addRDS(retDatas);
{
/* print topic name and topic message */
EXAMPLE_TRACE("--3--");
EXAMPLE_TRACE("packetId: %d", ptopic_info->packet_id);
EXAMPLE_TRACE("Topic: '%.*s' (Length: %d)",
ptopic_info->topic_len,
ptopic_info->ptopic,
ptopic_info->topic_len);
EXAMPLE_TRACE("Payload: '%.*s' (Length: %d)",
ptopic_info->payload_len,
ptopic_info->payload,
ptopic_info->payload_len);
EXAMPLE_TRACE("--3--");
}
};

void IOToMqttAliyun::unsubscribe()
{
for (std::map<std::string, AliyunServiceDesc>::iterator it = subTopicMaps.begin();
it != subTopicMaps.end(); ++it)
{
IOT_MQTT_Unsubscribe(pclient, it->first.c_str());
IOT_MQTT_Yield(pclient, 200);
}
}
3、服务状态查询与发布，ProducerMqttAliyun类负责巡检服务状态，然后加入CacheAliyunMQTT（单体类）的缓存队列中，IOToMqttAliyun类从缓存队列中读取数据进行发送阿里云物联网平台。

void ProducerMqttAliyun::checkSvcState()
{
for (std::map<int, SvcDesc>::iterator it = svcTriples.begin(); it != svcTriples.end(); ++it)
{
int state = 0;
SvcQuery((char*)it->second.svc_name.c_str(), state);
bool valChange = false;
if (state > 0)
{
ServiceState e_state = static_cast<ServiceState>(state);
//printf("svc:%s,state:%d ",it->second.svc_name.c_str(), state);
if (it->second.svc_state != e_state)
{
valChange = true;
it->second.svc_state = e_state;
it->second.markT = static_cast<unsigned int>(time(NULL)) + it->second.upLoopTime;
}
else {
if (it->second.markT < static_cast<unsigned int>(time(NULL)))//刷新时变化
{
valChange = true;
it->second.markT = static_cast<unsigned int>(time(NULL)) + it->second.upLoopTime;
}
}
}
if (valChange) {
WCacheAliyun wcAliyun(it->first,(int)it->second.svc_state);
ptr_CacheAliyunMQTT->addWDS(wcAliyun);
}
}
}

void IOToMqttAliyun::send()
{
WCacheAliyun it;
if (ptr_CacheAliyunMQTT->getFirstWDS(it))
{
std::map<int, SvcDesc>::iterator it_svc = svcTriples.find(it.id);
if (it_svc != svcTriples.end())
{
char buf_params[128] = { 0 };
sprintf(buf_params, "{"%s":%d}", it_svc->second.aliyun_key.c_str(),it.val);
iotx_mqtt_topic_info_t topic_msg;
memset(&topic_msg, 0x0, sizeof(iotx_mqtt_topic_info_t));
int msg_len = 0;
char msg_pub[MQTT_MSGLEN] = { 0 };
topic_msg.qos = IOTX_MQTT_QOS0;
topic_msg.retain = 0;
topic_msg.dup = 0;
topic_msg.payload = (char *)msg_pub;
topic_msg.payload_len = msg_len;
memset(msg_pub, 0x0, MQTT_MSGLEN);
msg_len = sprintf(msg_pub, PAYLOAD_FORMAT, cnt++, ALINK_METHOD_PROP_POST, buf_params);
if (msg_len > 0) {
topic_msg.payload = (char *)msg_pub;
topic_msg.payload_len = msg_len;
char buf[128] = { 0 };
sprintf(buf, ALINK_TOPIC_EVENT_PRO_POST
, gatewayTriples.product_key.c_str()
, gatewayTriples.device_name.c_str());
std::string alink_topic_service_pro_post = std::string(buf, strlen(buf));
if (!alink_topic_service_pro_post.empty()) {
int rc = IOT_MQTT_Publish(pclient, alink_topic_service_pro_post.c_str(), &topic_msg);
if (rc < 0) {
//EXAMPLE_TRACE("error occur when publish");
CLogger::createInstance()->Log(eTipMessage
, "%s|%03d :: publish message fail: topic: %s payload(%d): %s rc = %d"
, __func__, __LINE__
, alink_topic_service_pro_post.c_str()
, topic_msg.payload_len, topic_msg.payload, rc);
}
else {
IOT_MQTT_Yield(pclient, 200);
EXAMPLE_TRACE("packet-id=%u, publish topic msg=%s", (uint32_t)rc, msg_pub);
}
}
}
}
ptr_CacheAliyunMQTT->removeFirstWDS();
}
}
4、实现阿里云物联网平台的下控执行指令，阿里云物联网平台物模型数据格式为JSON格式，可以将回调函数写入缓存队列的数据进行JSON解析，阿里云物联网平台v2.3以后的开发包，提供了JSON类（cJSON.h，cJSON.cpp)，可以实现JSON解析

void ConsumerMqttAliyun::receive()
{
RCacheAliyun item_cache;
if (ptr_ReceiveCacheAliyunMQTT->getFirstRDS(item_cache))
{
ptr_ReceiveCacheAliyunMQTT->removeFirstRDS();
printf("topic:%s payload:%s ", item_cache.topic.c_str(), item_cache.payload.c_str());
std::map<std::string, AliyunServiceDesc>::iterator it = subTopicMaps.find(item_cache.topic);
if (it != subTopicMaps.end())
{
printf("********************&******************** ");
cJSON *request_root = NULL;
request_root = cJSON_Parse(item_cache.payload.c_str());
if (request_root == NULL || !cJSON_IsObject(request_root)) {
printf("JSON Parse Error ");
return;
}
cJSON *item_propertyid = NULL, *item_params = NULL;
item_propertyid = cJSON_GetObjectItem(request_root, "params");
if (item_propertyid != NULL && cJSON_IsObject(item_propertyid))
{
//EXAMPLE_TRACE("Property text:%s Value: %s"
// ,item_propertyid->string, item_propertyid->valuestring);
for (int j = 0; j < cJSON_GetArraySize(item_propertyid); j++)
{
item_params = cJSON_GetArrayItem(item_propertyid, j);
if (item_params != NULL && cJSON_IsNumber(item_params))
{
printf("Property ID, index: %d, text:%s, Value: %.2f "
, j, item_params->string, item_params->valuedouble);
std::map<std::string, ServiceInfo>::iterator itp = it->second.triples.find(item_params->string);
if (itp != it->second.triples.end())
{
svc_control(itp->second.svc_name, static_cast<int>(item_params->valuedouble));
}
}
if (item_params != NULL && cJSON_IsBool(item_params))
{
printf("Property ID, index: %d, text:%s, Value: %d "
, j, item_params->string, (item_params->valuedouble > 0 ? 1 : 0));
}
if (item_params != NULL && cJSON_IsString(item_params))
{
printf("Property ID, index: %d, text:%s, Value: %s "
, j, item_params->string, item_params->valuestring);
}
cJSON_Delete(item_params);
}
}
cJSON_Delete(item_propertyid);
cJSON_Delete(request_root);
}
}
};

void ConsumerMqttAliyun::svc_control(std::string svc_name, int cmd)
{
switch (cmd)
{
case 1:
SvcStop((char*)svc_name.c_str());
break;
case 2:
SvcStart((char*)svc_name.c_str());
break;
case 3:
break;
default:
break;
}
}
5、受监控的服务状态在云端具体设备实例下的运行状态页面可以查看，如果要进行下控，可以通过监控运维栏目的在线调试页面或者数据分析栏目的数据空间可视化-〉2D/3D数据管理页面实现：

6、在阿里云物联网平台的监控运维栏目下的固件升级页面创建新的固件，将更新资料（受监控的服务的软件更新包-软件及配套文件，可以是全包或差分包）打包上传。通过指定升级或批量升级将更新包推送到边缘设备端。

边缘设备端的接口代码如下：

业务逻辑是不断巡检是否存在更新，如果存在更新，则加载更新包到本地，变更更新标识（线程自会调用脚本实现更新操作），并更新版本号及上报云端，版本号采用文件保存，在程序初始启动时需读取上报。

struct OTAAliyunCache : public OTAAliyun
{
OTAAliyunCache()
: OTAAliyun()
, updatef(false)
, appDir("")
, batfile("ExportZip.bat")
, pathdiv("")
{

};
bool updatef;
std::string appDir;
std::string batfile;
std::string pathdiv;
};
OTAAliyunCache otaInfo;

void IOToMqttAliyun::ota_down()
{
if (IOT_OTA_IsFetching(h_ota))
{
char buf_ota[OTA_BUF_LEN];
FILE *fp;
if (NULL == (fp = fopen(otaInfo.update_file.c_str(), "wb+"))) {
EXAMPLE_TRACE("open file failed");
return;
}
uint32_t firmware_valid;
uint32_t last_percent = 0, percent = 0;
char version[128], md5sum[33];
uint32_t len, size_downloaded, size_file;
/* get OTA information */
//IOT_OTA_Ioctl(h_ota, IOT_OTAG_FETCHED_SIZE, &size_downloaded, 4);
//IOT_OTA_Ioctl(h_ota, IOT_OTAG_FILE_SIZE, &size_file, 4);
//IOT_OTA_Ioctl(h_ota, IOT_OTAG_MD5SUM, md5sum, 33);
//IOT_OTA_Ioctl(h_ota, IOT_OTAG_VERSION, version, 128);
do {

len = IOT_OTA_FetchYield(h_ota, buf_ota, OTA_BUF_LEN, 1);
if (len > 0) {
if (1 != fwrite(buf_ota, len, 1, fp)) {
EXAMPLE_TRACE("write data to file failed");
break;
}
}
else {
IOT_OTA_ReportProgress(h_ota, IOT_OTAP_FETCH_FAILED, NULL);
EXAMPLE_TRACE("ota fetch fail");
}

/* get OTA information */
IOT_OTA_Ioctl(h_ota, IOT_OTAG_FETCHED_SIZE, &size_downloaded, 4);
IOT_OTA_Ioctl(h_ota, IOT_OTAG_FILE_SIZE, &size_file, 4);
IOT_OTA_Ioctl(h_ota, IOT_OTAG_MD5SUM, md5sum, 33);
IOT_OTA_Ioctl(h_ota, IOT_OTAG_VERSION, version, 128);

percent = (size_downloaded * 100) / size_file;
if (percent - last_percent > 1)
{
last_percent = percent;
IOT_OTA_ReportProgress(h_ota, (IOT_OTA_Progress_t)percent, NULL);//加载进度报告
printf("IOT_OTA_Progress:--%d-- ", percent);
IOT_OTA_ReportProgress(h_ota, (IOT_OTA_Progress_t)percent, "hello");
}
IOT_MQTT_Yield(pclient, 100);
} while (!IOT_OTA_IsFetchFinish(h_ota));
fclose(fp);
IOT_OTA_Ioctl(h_ota, IOT_OTAG_CHECK_FIRMWARE, &firmware_valid, 4);
if (0 == firmware_valid) {
EXAMPLE_TRACE("The firmware is invalid");
}
else {
EXAMPLE_TRACE("The firmware is valid");
if (strlen(version) > 0) {
setVersion(version);
if (0 != IOT_OTA_ReportVersion(h_ota, version)) {
EXAMPLE_TRACE("report OTA version failed");
}
}
otaInfo.updatef = true;
}
}
}
//更新事例
void IOToMqttAliyun::update()
{
if (!otaInfo.updatef)
return;
File *ptr_File = File::getInstance();
if(!ptr_File->isExist(otaInfo.batfile))
{
//解压
char cmd[512] = { 0 };
#ifdef WIN32
sprintf(cmd, "%s "
"cd %s "
"%s e %s -o%s -y "
"cd %s "
, otaInfo.update_dir.substr(0, 2).c_str()
, otaInfo.update_dir.c_str()
, otaInfo.zip_path.c_str()
, otaInfo.update_file.c_str()
, otaInfo.update_dir.c_str()
, otaInfo.appDir.c_str());
#else
sprintf(cmd, "#!/bin/sh "
"cd %s "
"%s -xzvf %s -C %s "
"cd %s "
, otaInfo.update_dir.c_str()
, otaInfo.zip_path.c_str()
, otaInfo.update_file.c_str()
, otaInfo.update_dir.c_str()
, otaInfo.appDir.c_str());
#endif // WIN32
printf("cmd:%s ", cmd);
if (!ptr_File->writeToFile(cmd, otaInfo.batfile, "w"))
{
printf("write %s fail! ", otaInfo.batfile.c_str());
}
}
system(otaInfo.batfile.c_str());

//停止 重命名 拷贝 启动
for (std::map<int, SvcDesc>::iterator it = svcTriples.begin();
it != svcTriples.end(); ++it)
{
//stop svc
SvcStop((char*)it->second.svc_name.c_str());
//重命名旧app,拷贝新app

char cmd[512] = { 0 };
#ifdef WIN32
std::string svc_batfile = otaInfo.update_dir + otaInfo.pathdiv + it->second.app_name + ".bat";
sprintf(cmd, "%s "
"cd %s "
"rename %s %s_%s "
"copy /y %s%s%s %s "
"cd %s "
, it->second.app_dir.substr(0, 2).c_str()
, it->second.app_dir.c_str()
, it->second.app_name.c_str(), getCurrentTime().c_str(), it->second.app_name.c_str()
, otaInfo.update_dir.c_str(), otaInfo.pathdiv.c_str(), it->second.app_name.c_str(), it->second.app_dir.c_str()
, otaInfo.appDir.c_str());
#else
std::string svc_batfile = otaInfo.update_dir + otaInfo.pathdiv + it->second.app_name + ".sh";
sprintf(cmd, "#!/bin/sh "
"cd %s "
"rename %s %s_%s "
"cp -f %s%s%s %s "
"cd %s "
, it->second.app_dir.c_str()
, it->second.app_name.c_str(), getCurrentTime().c_str(), it->second.app_name.c_str()
, otaInfo.update_dir.c_str(), otaInfo.pathdiv.c_str(), it->second.app_name.c_str(), it->second.app_dir.c_str()
, otaInfo.appDir.c_str());
#endif // WIN32
printf("cmd:%s ", cmd);
if (!ptr_File->writeToFile(cmd, svc_batfile, "w"))
{
printf("write %s fail! ", svc_batfile.c_str());
}
system(svc_batfile.c_str());
//start
SvcStart((char*)it->second.svc_name.c_str());
}
otaInfo.updatef = false;
}

int IOToMqttAliyun::getVersion(char* version_def)
{
if (NULL == version_def)
{
return 0;
}
FILE *fp;
if (NULL == (fp = fopen(otaInfo.version_file.c_str(), "r")))
{
EXAMPLE_TRACE("open file failed");
size_t size = strlen(version_def);
if (size <= 0)
{
return 0;
}
if (NULL != (fp = fopen(otaInfo.version_file.c_str(), "w")))
{
if (1 != fwrite(version_def, size, 1, fp))
{
EXAMPLE_TRACE("write data to file failed");
}
fclose(fp);
}
}
else
{
char version_buf[128] = { 0 };
fread(version_buf, 128, 1, fp);
if (strlen(version_buf) > 0)
{
memcpy(version_def, version_buf, strlen(version_buf));
}
fclose(fp);
}
return static_cast<int>(strlen(version_def));
}

void IOToMqttAliyun::setVersion(char* version_def)
{
if (NULL == version_def)
{
return;
}
size_t size = strlen(version_def);
if (size <= 0)
{
return;
}
FILE *fp;
if (NULL != (fp = fopen(otaInfo.version_file.c_str(), "w")))
{
if (1 != fwrite(version_def, size, 1, fp))
{
EXAMPLE_TRACE("write data to file failed");
}
fclose(fp);
}
}
7、本文只是实现简要功能化，更产品化更细节的如版本校验、更新前后依赖、更新是否成功再上报等读者自行考究。

经测试如下：

 
————————————————
版权声明：本文为CSDN博主「py_free」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/py8105/java/article/details/89919470