DPL: send consumed-timeframes metric on a channel metric-feedback if it exists

ktf · davidrohr · commit 46c57457d23a · 2021-11-08T19:49:21.000+01:00
diff --git a/Framework/Core/src/CommonDataProcessors.cxx b/Framework/Core/src/CommonDataProcessors.cxx
@@ -27,6 +27,7 @@
 #include "Framework/InputSpec.h"
 #include "Framework/Logger.h"
 #include "Framework/OutputSpec.h"
+#include "Framework/RawDeviceService.h"
 #include "Framework/Variant.h"
 #include "../../../Algorithm/include/Algorithm/HeaderStack.h"
 #include "Framework/OutputObjHeader.h"
@@ -44,6 +45,8 @@
 #include <ROOT/RDataFrame.hxx>
 #include <ROOT/RArrowDS.hxx>
 #include <ROOT/RVec.hxx>
+
+#include <FairMQDevice.h>
 #include <chrono>
 #include <fstream>
 #include <functional>
@@ -506,6 +509,15 @@ DataProcessorSpec CommonDataProcessors::getDummySink(std::vector<InputSpec> cons
     .algorithm = AlgorithmSpec{adaptStateful([](CallbackService& callbacks) {
       auto dataConsumed = [](ServiceRegistry& services) {
         services.get<DataProcessingStats>().consumedTimeframes++;
+        auto device = services.get<RawDeviceService>().device();
+        auto channel = device->fChannels.find("metric-feedback");
+        if (channel != device->fChannels.end()) {
+          FairMQMessagePtr payload(device->NewMessage());
+          int64_t* consumed = (int64_t*)malloc(sizeof(int64_t));
+          *consumed = services.get<DataProcessingStats>().consumedTimeframes;
+          payload->Rebuild(consumed, sizeof(int64_t), nullptr, nullptr);
+          channel->second[0].Send(payload);
+        }
       };
       callbacks.set(CallbackService::Id::DataConsumed, dataConsumed);
 
