add cloud mq

LFtid1056/cloudfront/code/cfg_parser.cpp

@@ -173,6 +173,10 @@ std::string G_MQCONSUMER_TOPIC_LOG = "";//consumer topie
 std::string G_MQCONSUMER_TAG_LOG = "";//consumer tag
 std::string G_MQCONSUMER_KEY_LOG = "";//consumer key
 
+std::string G_MQCONSUMER_TOPIC_CLOUD = "";//consumer topie
+std::string G_MQCONSUMER_TAG_CLOUD = "";//consumer tag
+std::string G_MQCONSUMER_KEY_CLOUD = "";//consumer key
+
 //测试用的主题
 std::string G_ROCKETMQ_TOPIC_TEST = "";//topie
 std::string G_ROCKETMQ_TAG_TEST = "";//tag
@@ -315,6 +319,10 @@ void loadConfig(const std::string& filename) {
     strMap["RocketMq.ConsumerTagLOG"]       = &G_MQCONSUMER_TAG_LOG;
     strMap["RocketMq.ConsumerKeyLOG"]       = &G_MQCONSUMER_KEY_LOG;
 
+    strMap["RocketMq.ConsumerTopicCLOUD"]     = &G_MQCONSUMER_TOPIC_CLOUD;
+    strMap["RocketMq.ConsumerTagCLOUD"]       = &G_MQCONSUMER_TAG_CLOUD;
+    strMap["RocketMq.ConsumerKeyCLOUD"]       = &G_MQCONSUMER_KEY_CLOUD;
+
     strMap["RocketMq.Topic_Test"]           = &G_ROCKETMQ_TOPIC_TEST;
     strMap["RocketMq.Tag_Test"]             = &G_ROCKETMQ_TAG_TEST;
     strMap["RocketMq.Key_Test"]             = &G_ROCKETMQ_KEY_TEST;
@@ -2724,6 +2732,7 @@ bool compare_qvvr_and_file(const std::string& cfg_path, const std::vector<qvvr_d
 
     return false;
 }
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////数据转换函数
 // DataArrayItem to_json
 void to_json(nlohmann::json& j, const DataArrayItem& d) {
@@ -3077,4 +3086,11 @@ std::string normalize_mac(const std::string& mac) {
     std::string result = mac;
     result.erase(std::remove(result.begin(), result.end(), '-'), result.end());
     return result;
-}
+}
+
+////////////////////////////////////////////////////////////////////////////////////////
+
+
+
+
+

LFtid1056/cloudfront/code/rocketmq.cpp

@@ -1617,3 +1617,66 @@ void rocketmq_test_rc(Front* front)//用来测试补招
     queue_data_list.push_back(data);
 }
 
+////////////////////////////////////////////////////////////////////////////////////////////////////////////云前置新增功能
+
+//云前置功能
+rocketmq::ConsumeStatus cloudMessageCallback(const rocketmq::MQMessageExt& msg) {
+    //未初始化不处理消费
+    if (INITFLAG != 1) {
+        return rocketmq::RECONSUME_LATER;
+    }
+
+    std::string body = msg.getBody();
+    std::string key = msg.getKeys();
+
+    if (body.empty()) {
+        std::cerr << "Message body is NULL or empty." << std::endl;
+        return rocketmq::RECONSUME_LATER;
+    }
+
+    // 日志记录
+    DIY_INFOLOG("process", "【NORMAL】前置消费topic:%s_%s的云前置控制消息",FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_CLOUD.c_str());
+
+    std::cout << "rtdata Callback received message: " << body << std::endl;
+    if (!key.empty()) {
+        std::cout << "Message Key: " << key << std::endl;
+    } else {
+        std::cout << "Message Key: N/A" << std::endl;
+    }
+
+    // 消息解析
+    std::string devid, line;
+    bool realData = false, soeData = false;
+    int limit = 0;
+
+    if (!parseJsonMessageRT(body, devid, line, realData, soeData, limit)) {
+        std::cerr << "Failed to parse the JSON message." << std::endl;
+        DIY_ERRORLOG("process", "【ERROR】前置消费topic:%s_%s的云前置控制消息失败,消息的json格式不正确", FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str());
+        return rocketmq::RECONSUME_LATER;
+    }
+
+    // 加锁访问台账
+    int dev_index;
+    int mp_index;
+    if( !devid.empty() && !line.empty()){
+        std::lock_guard<std::mutex> lock(ledgermtx);
+        dev_index = find_dev_index_from_dev_id(devid);
+        mp_index  = find_mp_index_from_mp_id(line);
+    }
+    else{
+        std::cerr << "rtdata is NULL." << std::endl;
+        DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的云前置控制消息失败,消息的json结构不正确", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str());
+    }
+    
+
+    if (dev_index == -1 || mp_index == -1) {
+        std::cerr << "dev index or mp index is not found" << std::endl;
+        return rocketmq::RECONSUME_LATER;
+    }
+
+
+   //不再使用文件触发方式，直接调用接口向终端发起请求
+   ClientManager::instance().set_real_state_count(devid, 60,mp_index);//一秒询问一次，询问60次
+
+    return rocketmq::CONSUME_SUCCESS;
+}

LFtid1056/cloudfront/code/rocketmq.h

@@ -86,6 +86,10 @@ extern std::string G_MQCONSUMER_TOPIC_LOG;
 extern std::string G_MQCONSUMER_TAG_LOG;
 extern std::string G_MQCONSUMER_KEY_LOG;
 
+extern std::string G_MQCONSUMER_TOPIC_CLOUD;
+extern std::string G_MQCONSUMER_TAG_CLOUD;
+extern std::string G_MQCONSUMER_KEY_CLOUD;
+
 extern std::string G_LOG_TOPIC;
 extern std::string G_LOG_TAG;
 extern std::string G_LOG_KEY;