Add compatibility layer for FairMQ 1.3

ktf · ktf · commit 01c2aeff4eb3 · 2019-04-26T17:12:49.000+02:00
diff --git a/CCDB/src/ConditionsMQClient.cxx b/CCDB/src/ConditionsMQClient.cxx
@@ -21,6 +21,8 @@
 #include "CCDB/BackendOCDB.h"
 #include "CCDB/BackendRiak.h"
 #include "CCDB/ConditionsMQClient.h"
+#include "O2Device/Compatibility.h"
+
 #include <FairMQLogger.h>
 #include <options/FairMQProgOptions.h>
 
@@ -52,7 +54,7 @@ void ConditionsMQClient::Run()
 {
   Backend* backend;
 
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
 
     boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
     boost::posix_time::ptime startTime = boost::posix_time::microsec_clock::local_time();
diff --git a/CCDB/src/ConditionsMQServer.cxx b/CCDB/src/ConditionsMQServer.cxx
@@ -21,6 +21,7 @@
 #include "CCDB/Condition.h"
 #include "CCDB/ConditionsMQServer.h"
 #include "CCDB/IdPath.h"
+#include "O2Device/Compatibility.h"
 #include <FairMQLogger.h>
 #include <FairMQPoller.h>
 
@@ -92,7 +93,7 @@ void ConditionsMQServer::Run()
   std::unique_ptr<FairMQPoller> poller(
     fTransportFactory->CreatePoller(fChannels, { "data-put", "data-get", "broker-get" }));
 
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
 
     poller->Poll(-1);
 
diff --git a/Examples/flp2epn-distributed/src/EPNReceiver.cxx b/Examples/flp2epn-distributed/src/EPNReceiver.cxx
@@ -23,6 +23,7 @@
 #include <options/FairMQProgOptions.h>
 
 #include "FLP2EPNex_distributed/EPNReceiver.h"
+#include "O2Device/Compatibility.h"
 
 #include <iomanip>
 
@@ -110,7 +111,7 @@ void EPNReceiver::Run()
 
   FairMQChannel& ackOutChannel = fChannels.at(mAckChannelName).at(0);
 
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
     FairMQParts parts;
 
     if (Receive(parts, mInChannelName, 0, 100) > 0) {
diff --git a/Examples/flp2epn-distributed/src/FLPSender.cxx b/Examples/flp2epn-distributed/src/FLPSender.cxx
@@ -23,6 +23,7 @@
 #include <options/FairMQProgOptions.h>
 
 #include "FLP2EPNex_distributed/FLPSender.h"
+#include "O2Device/Compatibility.h"
 
 using namespace std;
 using namespace std::chrono;
@@ -71,7 +72,7 @@ void FLPSender::Run()
   // store the channel reference to avoid traversing the map on every loop iteration
   FairMQChannel& dataInChannel = fChannels.at(mInChannelName).at(0);
 
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
     // initialize f2e header
     auto* header = new f2eHeader;
     if (mTestMode > 0) {
diff --git a/Utilities/DataFlow/src/EPNReceiverDevice.cxx b/Utilities/DataFlow/src/EPNReceiverDevice.cxx
@@ -19,6 +19,7 @@
 #include "DataFlow/EPNReceiverDevice.h"
 #include "Headers/DataHeader.h"
 #include "Headers/SubframeMetadata.h"
+#include "O2Device/Compatibility.h"
 #include "TimeFrame/TimeFrame.h"
 
 #include <iomanip>
@@ -91,7 +92,7 @@ void EPNReceiverDevice::Run()
   std::multimap<TimeframeId, IndexElement> index;
   std::multimap<TimeframeId, FlpId> flpIds;
 
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
     FairMQParts subtimeframeParts;
     if (Receive(subtimeframeParts, mInChannelName, 0, 100) <= 0)
       continue;
diff --git a/Utilities/DataFlow/src/FLPSenderDevice.cxx b/Utilities/DataFlow/src/FLPSenderDevice.cxx
@@ -18,6 +18,7 @@
 #include "Headers/DataHeader.h"
 #include "Headers/SubframeMetadata.h"
 #include "DataFlow/FLPSenderDevice.h"
+#include "O2Device/Compatibility.h"
 
 using namespace std;
 using namespace std::chrono;
@@ -45,7 +46,7 @@ void FLPSenderDevice::Run()
   // store the channel reference to avoid traversing the map on every loop iteration
   //FairMQChannel& dataInChannel = fChannels.at(fInChannelName).at(0);
 
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
     // - Get the SubtimeframeMetadata
     // - Add the current FLP id to the SubtimeframeMetadata
     // - Forward to the EPN the whole subtimeframe
diff --git a/Utilities/DataFlow/src/TimeframeValidatorDevice.cxx b/Utilities/DataFlow/src/TimeframeValidatorDevice.cxx
@@ -20,6 +20,7 @@
 #include "TimeFrame/TimeFrame.h"
 #include "Headers/SubframeMetadata.h"
 #include "Headers/DataHeader.h"
+#include "O2Device/Compatibility.h"
 
 #include <options/FairMQProgOptions.h>
 
@@ -41,7 +42,7 @@ void o2::data_flow::TimeframeValidatorDevice::InitTask()
 
 void o2::data_flow::TimeframeValidatorDevice::Run()
 {
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
     FairMQParts timeframeParts;
     if (Receive(timeframeParts, mInChannelName, 0, 100) <= 0)
       continue;
diff --git a/Utilities/DataFlow/src/TimeframeWriterDevice.cxx b/Utilities/DataFlow/src/TimeframeWriterDevice.cxx
@@ -21,6 +21,8 @@
 #include "TimeFrame/TimeFrame.h"
 #include "Headers/SubframeMetadata.h"
 #include "Headers/DataHeader.h"
+#include "O2Device/Compatibility.h"
+
 #include <options/FairMQProgOptions.h>
 #include <boost/filesystem.hpp>
 
@@ -55,7 +57,7 @@ void TimeframeWriterDevice::Run()
   boost::filesystem::path p(mOutFileName);
   size_t streamedTimeframes = 0;
   bool needsNewFile = true;
-  while (CheckCurrentState(RUNNING) && mFileCount < mMaxFiles) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this) && mFileCount < mMaxFiles) {
     // In case we need to process more than one file,
     // the filename is split in basename and extension
     // and we call the files `<basename><count>.<extension>`.
diff --git a/Utilities/O2Device/include/O2Device/Compatibility.h b/Utilities/O2Device/include/O2Device/Compatibility.h
@@ -0,0 +1,43 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+#ifndef ALICEO2_UTILITIES_O2DEVICE_COMPATIBILITY_H_
+#define ALICEO2_UTILITIES_O2DEVICE_COMPATIBILITY_H_
+
+#include <type_traits>
+
+namespace o2
+{
+namespace compatibility
+{
+
+template <typename T>
+class FairMQ13
+{
+ private:
+  template <typename C>
+  static std::true_type test(decltype(&C::NewStatePending));
+  template <typename C>
+  static std::false_type test(...);
+
+ public:
+  static bool IsRunning(T* device)
+  {
+    if constexpr (std::is_same_v<decltype(test<T>(0)), std::true_type>) {
+      return device->NewStatePending();
+    } else if constexpr (std::is_same_v<decltype(test<T>(0)), std::false_type>) {
+      return device->CheckCurrentState(T::RUNNING);
+    }
+  }
+};
+
+} // namespace compatibility
+} // namespace o2
+
+#endif // ALICEO2_UTILITIES_O2DEVICE_COMPATIBILITY_H_
diff --git a/Utilities/O2MessageMonitor/src/O2MessageMonitor.cxx b/Utilities/O2MessageMonitor/src/O2MessageMonitor.cxx
@@ -31,10 +31,12 @@
 #include <FairMQLogger.h>
 #include <options/FairMQProgOptions.h>
 #include "O2MessageMonitor/O2MessageMonitor.h"
+#include "O2Device/Compatibility.h"
 
 using namespace std;
 using namespace o2::header;
 using namespace o2::base;
+using namespace o2::compatibility;
 
 //__________________________________________________________________________________________________
 void O2MessageMonitor::InitTask()
@@ -60,7 +62,7 @@ void O2MessageMonitor::Run()
 
   auto dataResource = o2::pmr::getTransportAllocator(subChannels[0].Transport());
 
-  while (CheckCurrentState(RUNNING) && (--mIterations) != 0) {
+  while (FairMQ13<FairMQDevice>::IsRunning(this) && (--mIterations) != 0) {
     this_thread::sleep_for(chrono::milliseconds(mDelay));
 
     O2Message message;
diff --git a/Utilities/aliceHLTwrapper/src/EventSampler.cxx b/Utilities/aliceHLTwrapper/src/EventSampler.cxx
@@ -26,6 +26,7 @@
 //  @brief  Sampler device for Alice HLT events in FairRoot/ALFA
 
 #include "aliceHLTwrapper/EventSampler.h"
+#include "O2Device/Compatibility.h"
 #include <FairMQLogger.h>
 #include <FairMQPoller.h>
 #include "aliceHLTwrapper/AliHLTDataTypes.h"
@@ -132,7 +133,7 @@ void EventSampler::Run()
 
   std::ofstream latencyLog(mLatencyLogFileName);
 
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
 
     // read input messages
     poller->Poll(mPollingTimeout);
@@ -230,7 +231,7 @@ void EventSampler::samplerLoop()
   int numOutputs = (fChannels.find("data-out") == fChannels.end() ? 0 : fChannels["data-out"].size());
 
   LOG(INFO) << "starting sampler loop, period " << mEventPeriod << " us";
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
     msg->Rebuild(sizeof(AliHLTComponentEventData));
     evtData = reinterpret_cast<AliHLTComponentEventData*>(msg->GetData());
     memset(evtData, 0, sizeof(AliHLTComponentEventData));
diff --git a/Utilities/aliceHLTwrapper/src/WrapperDevice.cxx b/Utilities/aliceHLTwrapper/src/WrapperDevice.cxx
@@ -27,6 +27,7 @@
 
 #include "aliceHLTwrapper/WrapperDevice.h"
 #include "aliceHLTwrapper/Component.h"
+#include "O2Device/Compatibility.h"
 #include <FairMQParts.h>
 #include <FairMQLogger.h>
 #include <FairMQPoller.h>
@@ -176,7 +177,7 @@ void WrapperDevice::Run()
 
   vector<FairMQParts> socketInputs(numInputs);
   int nReadCycles=0;
-  while (CheckCurrentState(RUNNING)) {
+  while (compatibility::FairMQ13<FairMQDevice>::IsRunning(this)) {
 
     // read input messages
     if (poller) {
diff --git a/cmake/O2Dependencies.cmake b/cmake/O2Dependencies.cmake
@@ -534,6 +534,7 @@ o2_define_bucket(
     ${FAIRROOT_INCLUDE_DIR}
     ${ROOT_INCLUDE_DIR}
     ${CMAKE_SOURCE_DIR}/Common/Utils/include
+    ${CMAKE_SOURCE_DIR}/Utilities/O2Device/include
 
     SYSTEMINCLUDE_DIRECTORIES
     ${PROTOBUF_INCLUDE_DIR}
