ctpdev: orbitreset and orbitsox via zmq II (#14147)

* dev: finishing orbitsox and orbitreset via zmq

* clang

* fix

* fix

* dev: removed infologger and using LOG(important)

* clang

* fix

Author: lietava

Detectors/CTP/workflowScalers/include/CTPWorkflowScalers/RunManager.h

@@ -18,6 +18,7 @@
 #include "DataFormatsCTP/Configuration.h"
 #include "BookkeepingApi/BkpClientFactory.h"
 #include "BookkeepingApi/BkpClient.h"
+
 using namespace o2::bkp::api;
 namespace o2
 {

Detectors/CTP/workflowScalers/src/RunManager.cxx

@@ -18,6 +18,7 @@
 #include <regex>
 #include "CommonUtils/StringUtils.h"
 #include <fairlogger/Logger.h>
+
 using namespace o2::ctp;
 ///
 /// Active run to keep cfg and saclers of active runs
@@ -232,10 +233,44 @@ int CTPRunManager::processMessage(std::string& topic, const std::string& message
     return 0;
   }
   if (topic.find("soxorbit") != std::string::npos) {
-    return 0;
+    std::vector<std::string> tokens = o2::utils::Str::tokenize(message, ' ');
+    int ret = 0;
+    if (tokens.size() == 3) {
+      long timestamp = std::stol(tokens[0]);
+      uint32_t runnumber = std::stoul(tokens[1]);
+      uint32_t orbit = std::stoul(tokens[2]);
+      ret = saveSoxOrbit(runnumber, orbit, timestamp);
+      std::string logmessage;
+      if (ret) {
+        logmessage = "Failed to update CCDB with SOX orbit.";
+      } else {
+        logmessage = "CCDB updated with SOX orbit.";
+      }
+      LOG(important) << logmessage << " run:" << runnumber << " sox orbit:" << orbit << " ts:" << timestamp;
+    } else {
+      LOG(error) << "Topic soxorbit dize !=3: " << message << " token size:" << tokens.size();
+      ret = 1;
+    }
+    return ret;
   }
   if (topic.find("orbitreset") != std::string::npos) {
-    return 0;
+    std::vector<std::string> tokens = o2::utils::Str::tokenize(message, ' ');
+    int ret = 0;
+    if (tokens.size() == 1) {
+      long timestamp = std::stol(tokens[0]);
+      ret = saveOrbitReset(timestamp);
+      std::string logmessage;
+      if (ret) {
+        logmessage = "Failed to update CCDB with orbitreset. ";
+      } else {
+        logmessage = "CCDB updated with orbitreset. ";
+      }
+      LOG(important) << logmessage << timestamp;
+    } else {
+      LOG(error) << "Topic orbit reset != 2: " << message << " token size:" << tokens.size();
+      ret = 1;
+    }
+    return ret;
   }
   static int nerror = 0;
   if (topic.find("sox") != std::string::npos) {

Detectors/CTP/workflowScalers/src/ctpCCDBManager.cxx

@@ -122,7 +122,7 @@ int ctpCCDBManager::saveSoxOrbit(uint32_t runNumber, uint32_t soxOrbit, long tim
   vect.push_back(timestamp);
   vect.push_back((uint64_t)runNumber);
   vect.push_back((uint64_t)soxOrbit);
-  long tmin = timestamp;
+  long tmin = timestamp / 1000;
   long tmax = tmin + 381928219;
   o2::ccdb::CcdbApi api;
   map<string, string> metadata; // can be empty
@@ -149,9 +149,10 @@ int ctpCCDBManager::saveOrbitReset(long timeStamp)
   if (timeStamp == 0) {
     auto now = std::chrono::system_clock::now();
     timeStamp = std::chrono::duration_cast<std::chrono::microseconds>(now.time_since_epoch()).count();
+    LOG(warn) << "Received timestamp = 0 , using current time:" << timeStamp;
   }
   vect.push_back(timeStamp);
-  long tmin = timeStamp;
+  long tmin = timeStamp / 1000;
   long tmax = tmin + 381928219;
   o2::ccdb::CcdbApi api;
   map<string, string> metadata; // can be empty