nop

.vscode/settings.json

@@ -95,7 +95,8 @@
         "csignal": "cpp",
         "__functional_03": "cpp",
         "variant": "cpp",
-        "format": "cpp"
+        "format": "cpp",
+        "*.ipp": "cpp"
     },
     "C_Cpp.errorSquiggles": "Disabled",
     "git.ignoreLimitWarning": true

opencv-motion-detect/evcloudsvc.cpp

@@ -37,6 +37,7 @@ const string mqtt_topic_report = "evsuits/report/";
 const string mqtt_topic_subscribe = "evsuits/request/";
 const string mqtt_topic_lastwill = "evsuits/lastwill/";
 const char kcArrMqttUrl[] = "tcp://admin:vJ3zHqWrHbrqxVMT@evcloudsvc.ilabservice.cloud:11883";
+const string kMqttClientId = "evcloudsvc";
 
 class EvCloudSvc {
 private:
@@ -75,10 +76,12 @@ private:
                 "msg": "[autogen]evcloudsvc detects cluster mgr {} offline for ipc {}",
                 "status": "active",
                 "time": 1591085518,
-                "type": "report"
+                "type": "report",
+                "ipc": "ipcSN"
             }
         )"_json;
         data["time"] = now;
+        data["ipc"] = ipcSn;
         bool hasIssue = false;
         if(v.count("mgrTerminal") == 0|| v["mgrTerminal"].size() == 0||v["mgrTerminal"].count("online") == 0) {
             data["catId"] = "AV_NO_CONFIG";
@@ -107,6 +110,7 @@ private:
                         auto data = j;
                         if(catId != EV_MSG_REPORT_CATID_AVMODOFFLINE) {
                             if(ret.count(catId) == 0) {
+                                data["ipc"] = ipcSn;
                                 ret[catId] = data;
                             }
                         }
@@ -118,6 +122,28 @@ private:
         return ret;
     }
 
+    void issueChecker(string ipcSn, json &ipcStatus){
+        auto issues = buildIssuesForIpc(ipcSn, ipcStatus);
+        if(allIssues.count(ipcSn) == 0) {
+            allIssues[ipcSn] = json();
+        }
+        bool report = false;
+
+        if(issues.size() > 0) {
+            for(auto &[kk,vv]: issues.items()) {
+                if(allIssues[ipcSn].count(kk) == 0) {
+                    report = true;
+                    break;
+                }
+            }
+        }
+        if(report) {
+            /// TODO:
+            allIssues[ipcSn] = issues;
+            MqttMgr::report_response_args(mqClient, mqtt_topic_report + kMqttClientId, 0, "ok", "issues", "", issues);
+        }
+    }
+
 
     void storeReportMsg(string selfId, string ipcSn, json &ipcStatus, string catId, string msg, bool active)
     {
@@ -146,7 +172,7 @@ private:
             ipcStatus["lastNReports"].erase(0);
         }
         // check status
-
+        issueChecker(ipcSn, ipcStatus);
     }
 
     int buildIpcStatus(json &conf)
@@ -822,23 +848,8 @@ private:
                         // }
                     }
 
-                    auto issues = buildIssuesForIpc(ipcSn, ipcStatus);
-                    if(allIssues.count(ipcSn) == 0) {
-                        allIssues[ipcSn] = json();
-                    }
-                    bool report = false;
-                    if(issues.size() > 0) {
-                        for(auto &[kk,vv]: issues.items()) {
-                            if(allIssues[ipcSn].count(kk) == 0) {
-                                report = true;
-                                break;
-                            }
-                        }
-                    }
-                    if(report) {
-                        /// TODO:
-
-                    }
+                    // TODO: report
+                    issueChecker(ipcSn, ipcStatus);
                 }
                 else {
                     spdlog::error("{} can't find ipc for report mod {}", devSn, modId);
@@ -973,7 +984,7 @@ private:
                     json jmeta = json::parse(meta);
                     if(jmeta["type"] == EV_MSG_META_TYPE_REPORT) {
                         json data = json::parse(body2str(body[3]));
-                        spdlog::warn("{} received report msg from {}: {}", devSn, selfId, data.dump());
+                        spdlog::info("{} received report msg from {}: {}", devSn, selfId, data.dump());
                         processReportMsg(selfId, data);
                     }
                     else {

opencv-motion-detect/mqtt_helper.cc

@@ -85,38 +85,11 @@ json make_online_msg(string sn)
 void on_connlost(void *context, char *cause)
 {
     MqttHelper *self = (MqttHelper *) context;
-    MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
     int rc = -1;
-
     spdlog::error("mqtt connection lost: {}", cause? cause: "unkown reason");
-
-    spdlog::warn("reconnecting {}", self->addr);
-    conn_opts.keepAliveInterval = 20;
-    conn_opts.cleansession = 1;
-    int cnt = 0;
-    const int kRetryIntervalMaxSeconds = 60 * 60;
-    const int kRetryIntervalMinSeconds = 10;
-    time_t t;
-    srand((unsigned) time(&t));
-    int sleepTime = 10;
-    while(rc != MQTTASYNC_SUCCESS){
-        if ((rc = MQTTAsync_connect(self->client, &conn_opts)) != MQTTASYNC_SUCCESS) {
-            string msg = fmt::format("mqtt failed to reconnect {}: {}",self->mqtt_url, MQTTAsync_strerror(rc));
-            spdlog::error(msg);
-            //self->state.set_exception(StrException(msg));
-            sleepTime = cnt* kRetryIntervalMinSeconds;
-            if(sleepTime > kRetryIntervalMaxSeconds) {
-                sleepTime = rand()%kRetryIntervalMaxSeconds;
-                if(sleepTime < kRetryIntervalMinSeconds){
-                    sleepTime = kRetryIntervalMinSeconds;
-                }
-            }
-            this_thread::sleep_for(chrono::seconds(sleepTime));
-            sleepTime++;
-        }else{
-            break;
-        }
-    }
+    // MqttHelper::AsyncResult as;
+    // as.state = MqttHelper::State::None;
+    // self->state.set_value(as);
 }
 
 int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message)

opencv-motion-detect/mqtt_helper.hpp

@@ -132,10 +132,7 @@ public:
         void *extra;
     };
 
-    /// State state = State::None; // 0: initial ; 1: ready; 2: disconnected; 3: destroyed
-    promise<AsyncResult> state;
-    MqttHelper(string mqtt_url, string id, string topic_lastwill, string topic_report, int kai = 20, int cs = 1): mqtt_url(mqtt_url), id(id), topic_lastwill(topic_lastwill), topic_report(topic_report)
-    {
+    void connect(){
         // make connection, throw excpetions
         MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
         MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
@@ -147,7 +144,7 @@ public:
             addr = "tcp://" + uri.Host + ":" + uri.Port;
         }
 
-        spdlog::debug("mqtt url: proto: {}, host: {}, port: {}, user: {}, pass: {}", uri.Protocol, uri.Host, uri.Port, uri.User, uri.Password);
+        spdlog::info("mqtt url: proto: {}, host: {}, port: {}, user: {}, pass: {}", uri.Protocol, uri.Host, uri.Port, uri.User, uri.Password);
 
         if(MQTTASYNC_SUCCESS != MQTTAsync_create(&client, addr.c_str(), (kIdPrefix+id).c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL)) {
             msg = "failed to async create mqtt";
@@ -181,6 +178,7 @@ public:
         }
         conn_opts.context = this;
         conn_opts.will = &lwm;
+        conn_opts.automaticReconnect = true;
         spdlog::info("trying to connect to {}", addr);
         if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS) {
             msg = fmt::format("Failed to start connect: {}", MQTTAsync_strerror(rc));
@@ -188,6 +186,8 @@ public:
             throw StrException(msg);
         }
 
+        spdlog::warn("mqtt connect: get_future");
+
         auto ar = state.get_future().get();
         if(ar.state != State::Ready) {
             string msg = fmt::format("ailed to initilaze mqtt: {}", MQTTAsync_strerror(ar.rc));
@@ -198,11 +198,17 @@ public:
             spdlog::info("initialze mqtt {} successfully", addr);
         }
 
-
         // reset state
         state = promise<AsyncResult>();
     }
 
+    /// State state = State::None; // 0: initial ; 1: ready; 2: disconnected; 3: destroyed
+    promise<AsyncResult> state;
+    MqttHelper(string mqtt_url, string id, string topic_lastwill, string topic_report, int kai = 20, int cs = 1): mqtt_url(mqtt_url), id(id), topic_lastwill(topic_lastwill), topic_report(topic_report)
+    {
+        connect();
+    }
+
     // can be called only state
     int subscribe(string topic, on_msg_fun_ptr_t on_msg, bool force = false)
     {