zgj
140 lines
4.2 KiB
C++
140 lines
4.2 KiB
C++
/**
|
|
* @文件路径: \datacenter\datacenter\datacenter.cpp
|
|
* @作者名称: zjp
|
|
* @文件版本: V1.0.0
|
|
* @创建日期: 2025-02-20 08:55:09
|
|
* @简要说明:
|
|
* @编辑时间: 2025-02-21 12:33:40
|
|
* @Copyright: 2025 Shenzhen ShenBao Electrical instruments CO.,LTD. All Rights Reserved.
|
|
*/
|
|
#include <iostream>
|
|
#include <unistd.h>
|
|
#include <functional>
|
|
|
|
#include "datacenter.h"
|
|
#include "msg.h"
|
|
#include "datacenter/bizService/F_GetVersion.h"
|
|
#include "ToolUtils.h"
|
|
|
|
#include "Database.h"
|
|
#include "ConnectionPool.h"
|
|
|
|
ConnectionPool pool(POOLSIZE);
|
|
|
|
REGISTER_SERVICE(DataCenter);
|
|
|
|
int WriteResponseMsg(msg_t msg) {
|
|
printf("WriteResponseMsg %s\n", msg.topic.c_str());
|
|
return 0;
|
|
}
|
|
|
|
void DataCenter::datacenterinit(void) {
|
|
this->objmqttClient = new CMqttClient(&mQueue_In);
|
|
if (NULL == objmqttClient)
|
|
exit(-1);
|
|
|
|
this->objThreadPool = new ThreadPool(MAX_THREAD_POOL_SIZE);
|
|
if (NULL == objThreadPool){
|
|
delete objmqttClient;
|
|
exit(-1);
|
|
}
|
|
|
|
objmqttClient->connectServer();
|
|
|
|
datacenterOnline();
|
|
}
|
|
|
|
void DataCenter::datacenterOnline() {
|
|
msg_t intMsg,outMsg;
|
|
std::map<std::string, IBizService*> map = BizServiceFactovy::getBizServiceMap();
|
|
auto itor = map.find("F-Online");
|
|
//是否需要重试
|
|
if(itor != map.end()){
|
|
if(itor->second->HandleMsg(intMsg, outMsg, NULL) != 0)
|
|
return;
|
|
}
|
|
msg_mqtt_t outMqttMsg;
|
|
outMqttMsg.topic = outMsg.topic;
|
|
outMqttMsg.payload = outMsg.payload;
|
|
|
|
//mQueue_Out.push(outMqttMsg);
|
|
this->objmqttClient->publish(outMqttMsg.topic,outMqttMsg.payload);
|
|
}
|
|
std::string getPidFromTopic(std::string topic){
|
|
size_t pos = topic.find_last_of("/");
|
|
if(pos != std::string::npos){
|
|
return topic.substr(pos+1);
|
|
}
|
|
return "";
|
|
}
|
|
bool handleMsg(msg_mqtt_t inMqttMsg,MessageQueue &mQueue_Out) {
|
|
int fRet = -1;
|
|
msg_t intMsg,outMsg;
|
|
|
|
intMsg.topic = inMqttMsg.topic;
|
|
intMsg.payload = inMqttMsg.payload;
|
|
intMsg.subTopicV = inMqttMsg.subTopicV;
|
|
intMsg.timestamp = ToolUtils::getUtcTime(true);
|
|
|
|
std::string pidStr = getPidFromTopic(inMqttMsg.topic);
|
|
if(pidStr.empty()){
|
|
return false;
|
|
}
|
|
std::map<std::string, IBizService*> map = BizServiceFactovy::getBizServiceMap();
|
|
auto itor = map.find(pidStr);
|
|
if(itor == map.end()){
|
|
return false;
|
|
}
|
|
|
|
Database *conn = pool.getConnection(); // 获取数据库连接
|
|
sqlite3 *db = conn->GetHandle(); // 获取数据库句柄
|
|
fRet = itor->second->HandleMsg(intMsg, outMsg, (handparam_t *)db);
|
|
pool.returnConnection(conn); // 归还数据库连接
|
|
if (fRet == -1) {
|
|
return false;
|
|
}
|
|
|
|
msg_mqtt_t outMqttMsg;
|
|
outMqttMsg.topic = outMsg.topic;
|
|
outMqttMsg.payload = outMsg.payload;
|
|
mQueue_Out.push(outMqttMsg);
|
|
return true;
|
|
}
|
|
|
|
void DataCenter::updateSubTopicsMap(msg_mqtt_t &msgMqtt) {
|
|
if(this->tmpSubTopicsMap.find(msgMqtt.topic) != this->tmpSubTopicsMap.end()){
|
|
msgMqtt.subTopicV = this->tmpSubTopicsMap[msgMqtt.topic];
|
|
return;
|
|
}
|
|
std::vector<std::string>subTopicV;
|
|
ToolUtils::topic_split(msgMqtt.topic, '/', subTopicV);
|
|
msgMqtt.subTopicV = subTopicV;
|
|
this->tmpSubTopicsMap.insert(std::make_pair(msgMqtt.topic, subTopicV));
|
|
}
|
|
void DataCenter::ExecPoll(void *argc) {
|
|
DataCenter *that = (DataCenter *)argc;
|
|
if(NULL == that){
|
|
exit(-1);
|
|
}
|
|
|
|
while (!that->recvThreadFlag) {
|
|
msg_mqtt_t inMsg = that->mQueue_In.pop();
|
|
that->updateSubTopicsMap(inMsg);
|
|
std::cout << "Recv topic :" << inMsg.topic << "\n playload : " << inMsg.payload << std::endl;
|
|
that->objThreadPool->submitTask(1,handleMsg, inMsg, std::ref(that->mQueue_Out));
|
|
}
|
|
|
|
}
|
|
|
|
void DataCenter::ExecReponse(void *argc) {
|
|
DataCenter *that = (DataCenter *)argc;
|
|
if(NULL == that){
|
|
exit(-1);
|
|
}
|
|
|
|
while(!that->sendThreadFlag){
|
|
msg_mqtt_t outMsg = that->mQueue_Out.pop();
|
|
std::cout << "send topic :" << outMsg.topic << "\n playload : " << outMsg.payload << std::endl;
|
|
that->objmqttClient->publish(outMsg.topic,outMsg.payload);
|
|
}
|
|
} |