Update for new opmon interfaces

CMakeLists.txt

@@ -17,8 +17,8 @@ find_package(opmonlib REQUIRED)
 
 set(IPM_DEPENDENCIES ${CETLIB} ${CETLIB_EXCEPT} ers::ers logging::logging nlohmann_json::nlohmann_json utilities::utilities opmonlib::opmonlib cppzmq pthread)
 
+daq_protobuf_codegen( IPMInfo.proto )
 
-#daq_add_library(Receiver.cpp Sender.cpp LINK_LIBRARIES appfwk::appfwk logging::logging cppzmq)
 daq_add_library(Receiver.cpp Sender.cpp CallbackAdapter.cpp LINK_LIBRARIES ${IPM_DEPENDENCIES})
 
 daq_add_plugin(ZmqSender duneIPM LINK_LIBRARIES ipm)

include/ipm/Receiver.hpp

@@ -27,7 +27,7 @@
 #include "cetlib/compiler_macros.h"
 #include "ers/Issue.hpp"
 #include "nlohmann/json.hpp"
-#include "opmonlib/InfoCollector.hpp"
+#include "opmonlib/MonitorableObject.hpp"
 
 #include <atomic>
 #include <memory>
@@ -67,7 +67,7 @@ ERS_DECLARE_ISSUE(ipm,
 
 namespace dunedaq::ipm {
 
-class Receiver
+  class Receiver : public opmonlib::MonitorableObject
 {
 
 public:
@@ -108,9 +108,10 @@ class Receiver
   Receiver(Receiver&&) = delete;
   Receiver& operator=(Receiver&&) = delete;
 
-  void get_info(opmonlib::InfoCollector& ci, int /*level*/);
-
 protected:
+
+  void generate_opmon_data() override ;
+
   virtual Response receive_(const duration_t& timeout, bool no_tmoexcept_mode) = 0;
 
 private:

include/ipm/Sender.hpp

@@ -27,7 +27,7 @@
 #include "cetlib/compiler_macros.h"
 #include "ers/Issue.hpp"
 #include "nlohmann/json.hpp"
-#include "opmonlib/InfoCollector.hpp"
+#include "opmonlib/MonitorableObject.hpp"
 
 #include <atomic>
 #include <memory>
@@ -65,7 +65,7 @@ ERS_DECLARE_ISSUE(ipm,
 
 namespace dunedaq::ipm {
 
-class Sender
+  class Sender : public opmonlib::MonitorableObject 
 {
 
 public:
@@ -99,9 +99,10 @@ class Sender
   Sender(Sender&&) = delete;
   Sender& operator=(Sender&&) = delete;
 
-  void get_info(opmonlib::InfoCollector& ci, int /*level*/);
-
 protected:
+
+  void generate_opmon_data() override;
+
   virtual bool send_(const void* message,
                      message_size_t N,
                      const duration_t& timeout,

schema/ipm/IPMInfo.proto

@@ -0,0 +1,16 @@
+syntax = "proto3";
+
+
+package dunedaq.ipm.opmon;
+
+// Information from the sender	
+message SenderInfo {
+  uint64 bytes = 1;
+  uint64 messages = 2;   
+}
+
+// Information from the receiver	
+message ReceiverInfo {
+  uint64 bytes = 1;
+  uint64 messages = 2;   
+}

src/Receiver.cpp

@@ -7,7 +7,7 @@
  */
 
 #include "ipm/Receiver.hpp"
-#include "ipm/receiverinfo/InfoNljs.hpp"
+#include "ipm/IPMInfo.pb.h"
 
 dunedaq::ipm::Receiver::Response
 dunedaq::ipm::Receiver::receive(const duration_t& timeout, message_size_t bytes, bool no_tmoexcept_mode)
@@ -31,13 +31,13 @@ dunedaq::ipm::Receiver::receive(const duration_t& timeout, message_size_t bytes,
 }
 
 void
-dunedaq::ipm::Receiver::get_info(opmonlib::InfoCollector& ci, int /*level*/)
+dunedaq::ipm::Receiver::generate_opmon_data()
 {
 
-  receiverinfo::Info i;
+  opmon::ReceiverInfo i;
 
-  i.bytes = m_bytes.exchange(0);
-  i.messages = m_messages.exchange(0);
+  i.set_bytes(m_bytes.exchange(0));
+  i.set_messages(m_messages.exchange(0));
 
-  ci.add(i);
+  publish(std::move(i));
 }

src/Sender.cpp

@@ -7,7 +7,7 @@
  */
 
 #include "ipm/Sender.hpp"
-#include "ipm/senderinfo/InfoNljs.hpp"
+#include "ipm/IPMInfo.pb.h"
 
 #include <string>
 #include <vector>
@@ -40,13 +40,13 @@ dunedaq::ipm::Sender::send(const void* message,
 }
 
 void
-dunedaq::ipm::Sender::get_info(opmonlib::InfoCollector& ci, int /*level*/)
+dunedaq::ipm::Sender::generate_opmon_data()
 {
 
-  senderinfo::Info i;
+  opmon::SenderInfo i;
 
-  i.bytes = m_bytes.exchange(0);
-  i.messages = m_messages.exchange(0);
+  i.set_bytes(m_bytes.exchange(0));
+  i.set_messages(m_messages.exchange(0));
 
-  ci.add(i);
+  publish(std::move(i));
 }