diff --git a/CMakeLists.txt b/CMakeLists.txt
index 39428fc7170b0c504faee6cf53a2bd39c3afe211..f65eaf1c61de0f4a80418ed13ff8c21c6c08e169 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -129,6 +129,13 @@ if(FAIRROOT_FOUND)
endif()
+#Searching for Boost Package
+find_package(ZeroMQ)
+if(ZeroMQ_FOUND)
+ list(APPEND packages ZeroMQ)
+endif()
+
+
#Searching for Boost Package
find_package(Boost COMPONENTS serialization regex filesystem log log_setup container program_options thread)
if(Boost_FOUND)
diff --git a/services/histserv/Roadmap.md b/services/histserv/Roadmap.md
new file mode 100644
index 0000000000000000000000000000000000000000..70b473133767acf554020f7f1d43bc30a923caac
--- /dev/null
+++ b/services/histserv/Roadmap.md
@@ -0,0 +1,26 @@
+Histo source: class/lib + EventBuildChain
+
+- Members: hostname, port, ZMQ socket, list of known histo names
+
+- Interface method/constructor: init socket
+ => Type? PUB or PUSH?
+- Interface function: new entry <histo name, histo, (folder)>
+ => Check if histo already known
+ => No = add config part to message with <histo name, folder>
+ => Add part to message with <histo name, histo>
+ -> How do we stream the Histo1D object? boost streamer call?
+
+- Message format
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+Histo consummer: histserv
+
+- Members: ZMQ socket
+- Interface function: init socket
+ => Type? SUB or PULL?
+ => PULL
+- Exec method: ZMQ poll/get loop
+ => Blocking?
+ => Exit condition?
+- Config deserialization: boost?
+- Histos deserialization: boost?
+- Message processing: parts + calls to deserialization methods for Config and histo depending on size
diff --git a/services/histserv/app/Application.cxx b/services/histserv/app/Application.cxx
index d5ca5d33d95f3b3f3ba0d51c9308c6ee413e12fa..f3caeb2a77b25db91a7b8129c5ab695a5af63171 100644
--- a/services/histserv/app/Application.cxx
+++ b/services/histserv/app/Application.cxx
@@ -20,7 +20,11 @@
#include "TRootSniffer.h"
#include "TSystem.h"
+#include <boost/archive/binary_iarchive.hpp>
+#include <boost/iostreams/device/array.hpp>
+#include <boost/iostreams/stream.hpp>
#include <boost/serialization/utility.hpp>
+#include <boost/serialization/vector.hpp>
#include <mutex>
@@ -29,6 +33,9 @@
std::mutex mtx;
+namespace b_io = boost::iostreams;
+namespace b_ar = boost::archive;
+
namespace cbm::services::histserv
{
@@ -36,20 +43,18 @@ namespace cbm::services::histserv
Application::Application(ProgramOptions const& opt) : fOpt(opt)
{
/// Read options from executable
- LOG(info) << "Options for Application.";
- // FIXME: alternative to FairMQ fsChannelNameHistosInput = fConfig->GetValue<std::string>("ChNameIn");
- LOG(info) << " HTTP server port: " << fOpt.HttpPort();
+ LOG(info) << "Options for Application:";
+ LOG(info) << " Input ZMQ channel: " << fOpt.ComChan();
+ LOG(info) << " HTTP server port: " << fOpt.HttpPort();
if ("" != fOpt.HistoFile()) { //
- LOG(info) << "Output filename: " << fOpt.HistoFile() << (fOpt.Overwrite() ? " (in overwrite mode)" : "");
+ LOG(info) << " Output filename: " << fOpt.HistoFile() << (fOpt.Overwrite() ? " (in overwrite mode)" : "");
}
/// FIXME: SOMETHING_To_Replace_FairMQ!!!!!!!!!!!!!
/// FIXME: Initialize communication channels of SOMETHING_To_Replace_FairMQ
/// FIXME: Link channel to method in order to process received messages
- /*
- /// Method processing either combined Config+Data or Data only
- OnData(fsChannelNameHistosInput, &CbmMqHistoServer::ReceiveConfigAndData);
- * */
+ // fZmqSocket.set(zmq::sockopt::rcvhwm, int(hwm)); // FIXME: need for HWM?
+ fZmqSocket.bind(fOpt.ComChan().c_str()); // This side "binds" the socket => Other side should connect!!!!
fServer = new THttpServer(Form("http:%u", fOpt.HttpPort()));
/// To avoid the server sucking all Histos from gROOT when no output file is used
@@ -82,12 +87,85 @@ namespace cbm::services::histserv
fStopThread = false;
fThread = std::thread(&Application::UpdateHttpServer, this);
+ const std::chrono::milliseconds timeout {10};
+ /*
+ * Flesnet way (also cppzmq tour https://brettviren.github.io/cppzmq-tour)
+ * ZMQ Draft API! Need custom built ZMQ + #include <zmq_addon.hpp>!
+ zmq::poller_t in_poller;
+ in_poller.add(fZmqSocket, zmq::event_flags::pollin);
+ std::vector<decltype(poller)::poller_event> in_events(1);
+ */
+ /* Alternative taken from FairMQ Poller and zmq.hpp/monitor::check_event() */
+ // http://api.zeromq.org/2-1:zmq-poll: void *socket, int fd, short events, short revents
+ zmq_pollitem_t pollit {fZmqSocket.handle(), 0, ZMQ_POLLIN, 0};
+
while (!bHistoServerStop) { //
/// Infinite loop, this is a service which should survive until told otherwise after all
/// FIXME: Start listening to <SOMETHING?!?> to receive histograms and configuration
+ /*
+ * Flesnet way (also cppzmq tour https://brettviren.github.io/cppzmq-tour)
+ * ZMQ Draft API! Need custom built ZMQ + #include <zmq_addon.hpp>!
+ if (in_poller.wait_all(in_events, timeout)) {
+ zmq::message_t msg;
+ zmq::recv_result_t rres = fZmqSocket.recv(msg, zmq::recv_flags::none);
+ }
+ */
+ /* Alternative taken from FairMQ Poller and zmq.hpp/monitor::check_event() */
+ zmq::poll(&pollit, 1, timeout);
+ if (pollit.revents & ZMQ_POLLIN) {
+ std::vector<zmq::message_t> vMsg;
+ int more = 1;
+ size_t moreSize = sizeof(more);
+ while (more) {
+ vMsg.push_back(zmq::message_t());
+ int rc = zmq_msg_recv(vMsg.back().handle(), fZmqSocket.handle(), 0);
+ if (rc == -1) {
+ switch (zmq_errno()) {
+ case ETERM: {
+ LOG(debug) << "polling exited, reason: socket was terminated (" << zmq_strerror(errno) << ")";
+ more = 0;
+ bHistoServerStop = true;
+ break;
+ }
+ case EINTR: {
+ LOG(debug) << "polling interrupted by system call";
+ /// FIXME: not sure if should stop or go-on at this point
+ more = 0;
+ bHistoServerStop = true;
+ break;
+ }
+ default: {
+ LOG(error) << "polling failed, reason: " << zmq_strerror(errno);
+ more = 0;
+ bHistoServerStop = true;
+ break;
+ }
+ }
+ continue;
+ }
+ else {
+ zmq_getsockopt(fZmqSocket, ZMQ_RCVMORE, &more, &moreSize);
+ }
+ }
+
+ /// Check if multi-parts message
+ if (3 < vMsg.size()) { //
+ ReceiveConfigAndData(vMsg);
+ }
+ else if (1 == vMsg.size()) {
+ ReceiveData(vMsg[0]);
+ }
+ else if (0 < vMsg.size()) {
+ LOG(error) << "Invalid number of message parts received: should be either 1 or more than 3 vs "
+ << vMsg.size();
+ }
+ else {
+ LOG(error) << "polling told some data to expect but none successfully readout!";
+ }
+ }
- std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ //std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
// -------------------------------------------------------------------------------------------------------------------
@@ -130,24 +208,59 @@ namespace cbm::services::histserv
// -------------------------------------------------------------------------------------------------------------------
// ----- Server update background thread -------------------------------------------------------------------------
- bool Application::ReceiveData(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/)
+ bool Application::ReceiveData(zmq::message_t& msg)
{
LOG(debug) << "Application::ReceiveData => Processing histograms update";
- TObject* tempObject = nullptr;
/// FIXME: Something to replace FairMQ and extract the histograms!!!!
/// FIXME: Need something to replace the ROOT serializer which allowed to have any of TH1x, TH2x, TH3x or TProfile
- cbm::algo::Histo1D source(1, 0.0, 1.0, "dummy");
-
- /// copied from CbmTaskDigiEventQa::ToTH1D
- TH1D* result =
- new TH1D(source.Name().c_str(), source.Name().c_str(), source.NumBins(), source.MinValue(), source.MaxValue());
- for (uint32_t bin = 0; bin < source.NumBins(); bin++) {
- result->SetBinContent(bin, source.Content(bin));
+ /// FIXME: Need something to replace the TObjArray which allowed to have a mix of of TH1x, TH2x, TH3x or TProfile
+ b_io::basic_array_source<char> device(static_cast<char*>(msg.data()), msg.size());
+ b_io::stream<b_io::basic_array_source<char>> s(device);
+ b_ar::binary_iarchive iarch(s);
+
+ std::vector<cbm::algo::Histo1D> vHist;
+ iarch >> vHist;
+
+ for (auto& source : vHist) {
+ /// copied from CbmTaskDigiEventQa::ToTH1D
+ /// FIXME: Should be placed in a tools/interface/whatever library with all similar functions!!
+ /// FIXME: Reverse OP need to be implemented + CI unit tests for back and forth in each direction (ROOT <-> Algo)
+ /// FIXME: Lead to "Warning in <TROOT::Append>: Replacing existing TH1: xxxxxx (Potential memory leak)."
+ bool add = TH1::AddDirectoryStatus();
+ TH1::AddDirectory(false); // Needed to prevent ROOT from adding histogram to its internal registry
+ TH1D* result = new TH1D(source.Name().c_str(), source.Title().c_str(), // Titles in ROOT convention with ";" sep
+ source.NumBins(), source.MinValue(), source.MaxValue());
+ TH1::AddDirectory(add); // Needed to prevent ROOT from adding histogram to its internal registry
+ for (uint32_t bin = 0; bin < source.NumBins(); bin++) {
+ result->SetBinContent(1 + bin, source.Content(bin));
+ }
+ result->SetBinContent(0, source.Underflow());
+ result->SetBinContent(source.NumBins(), source.Overflow());
+ result->SetEntries(source.NumEntries());
+ if (!ReadHistogram<TH1>(result)) { //
+ return false;
+ }
+ delete result;
}
- result->SetEntries(source.NumEntries());
- tempObject = result;
+ /// If new histos received, try to prepare as many canvases as possible
+ /// Should be expensive on start and cheap afterward
+ if (!fbAllCanvasReady) {
+ LOG(debug) << "Application::ReceiveData => Checking for canvases updates";
+ for (uint32_t uCanv = 0; uCanv < fvpsCanvasConfig.size(); ++uCanv) {
+ /// Jump canvases already ready
+ if (fvbCanvasReady[uCanv]) { //
+ continue;
+ }
+
+ /// Now come the expensive part as we unpack its config and check each histo
+ fvbCanvasReady[uCanv] = PrepareCanvas(uCanv);
+ } // for( uint32_t uCanv = 0; uCanv < fvpsCanvasConfig.size(); ++uCanv )
+ } // if( !fbAllCanvasReady )
+
+ /*
+ TObject* tempObject = nullptr;
if (TString(tempObject->ClassName()).EqualTo("TObjArray")) {
std::lock_guard<std::mutex> lk(mtx);
TObjArray* arrayHisto = static_cast<TObjArray*>(tempObject);
@@ -199,23 +312,26 @@ namespace cbm::services::histserv
throw std::runtime_error(err_msg);
}
- fNMessages += 1;
-
if (nullptr != tempObject) delete tempObject;
+ */
+
+ fNMessages += 1;
LOG(debug) << "Application::ReceiveData => Finished processing histograms update";
return true;
}
- bool Application::ReceiveHistoConfig(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/)
+ bool Application::ReceiveHistoConfig(zmq::message_t& msg)
{
- std::pair<std::string, std::string> tempObject("", "");
-
- /// FIXME: Something to replace FairMQ and extract the histograms!!!!
- // Deserialize<BoostSerializer<std::pair<std::string, std::string>>>(*msg, tempObject);
- // BoostSerializer<std::pair<std::string, std::string>>().Deserialize(*msg, tempObject);
+ /// FIXME: Something to replace FairMQ and extract the config!!!!
+ // BoostSerializer<std::pair<std::string, std::string>>().Deserialize(msg, tempObject);
+ b_io::basic_array_source<char> device(static_cast<char*>(msg.data()), msg.size());
+ b_io::stream<b_io::basic_array_source<char>> s(device);
+ b_ar::binary_iarchive iarch(s);
+ std::pair<std::string, std::string> tempObject("", "");
+ iarch >> tempObject;
LOG(info) << " Received configuration for histo " << tempObject.first << " : " << tempObject.second;
@@ -243,13 +359,16 @@ namespace cbm::services::histserv
return true;
}
- bool Application::ReceiveCanvasConfig(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/)
+ bool Application::ReceiveCanvasConfig(zmq::message_t& msg)
{
- std::pair<std::string, std::string> tempObject("", "");
+ /// FIXME: Something to replace FairMQ and extract the config!!!!
+ // BoostSerializer<std::pair<std::string, std::string>>().Deserialize(msg, tempObject);
+ b_io::basic_array_source<char> device(static_cast<char*>(msg.data()), msg.size());
+ b_io::stream<b_io::basic_array_source<char>> s(device);
+ b_ar::binary_iarchive iarch(s);
- /// FIXME: Something to replace FairMQ and extract the histograms!!!!
- // Deserialize<BoostSerializer<std::pair<std::string, std::string>>>(*msg, tempObject);
- // BoostSerializer<std::pair<std::string, std::string>>().Deserialize(*msg, tempObject);
+ std::pair<std::string, std::string> tempObject("", "");
+ iarch >> tempObject;
LOG(info) << " Received configuration for canvas " << tempObject.first << " : " << tempObject.second;
@@ -279,44 +398,32 @@ namespace cbm::services::histserv
return true;
}
- bool Application::ReceiveConfigAndData(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/)
+ bool Application::ReceiveConfigAndData(std::vector<zmq::message_t>& vMsg)
{
/// FIXME: Something to replace FairMQ and extract the histograms!!!!
- /*
- /// Reject anything but
- /// - Header + Histo Config + Canvas Config + Histo Data
- /// - Histo data only
- if (4 != parts.Size()) {
- if (1 == parts.Size()) {
- /// Catch case of having only the histograms updates
- return ReceiveData(parts.At(0), 0);
- } // if( 1 == parts.Size() )
-
- fStopThread = true;
- std::string err_msg = "Application::ReceiveConfigAndData => Wrong number of parts: ";
- err_msg += parts.Size();
- err_msg += " instead of 4 (Header + Histo Config + Canvas config + Data) or 1 (Data)!";
- throw std::runtime_error(err_msg);
- } // if( parts.Size() < 4 )
+ LOG(debug) << "Application::ReceiveConfigAndData => Received composed message with " << vMsg.size() << " parts";
- LOG(info) << "Application::ReceiveConfigAndData => Received composed message with " << parts.Size() << " parts";
+ /// Header contains a pair of unsigned integers
+ /// FIXME: Something to replace FairMQ and extract the header!!!!
+ // BoostSerializer<std::pair<uint32_t, uint32_t>>().Deserialize(vMsg.at(0), pairHeader);
+ b_io::basic_array_source<char> device_header(static_cast<char*>(vMsg.at(0).data()), vMsg.at(0).size());
+ b_io::stream<b_io::basic_array_source<char>> s_header(device_header);
+ b_ar::binary_iarchive iarch_header(s_header);
- /// Header contains a pair of
std::pair<uint32_t, uint32_t> pairHeader;
- // Deserialize<BoostSerializer<std::pair<uint32_t, uint32_t>>>(*parts.At(0), pairHeader);
- BoostSerializer<std::pair<uint32_t, uint32_t>>().Deserialize(*parts.At(0), pairHeader);
-
- LOG(info) << "Application::ReceiveConfigAndData => Received configuration for " << pairHeader.first
- << " histos and " << pairHeader.second << " canvases";
+ iarch_header >> pairHeader;
+ LOG(debug) << "Application::ReceiveConfigAndData => Received configuration for " << pairHeader.first
+ << " histos and " << pairHeader.second << " canvases";
uint32_t uOffsetHistoConfig = pairHeader.first;
if (0 == pairHeader.first) {
uOffsetHistoConfig = 1;
- if (0 < (parts.At(uOffsetHistoConfig))->GetSize()) {
- fStopThread = true;
+ if (0 < vMsg[uOffsetHistoConfig].size()) {
+ fStopThread = true;
+ bHistoServerStop = true;
std::string err_msg = "Application::ReceiveConfigAndData => No histo config expected but corresponding message";
err_msg += " is not empty: ";
- err_msg += (parts.At(uOffsetHistoConfig))->GetSize();
+ err_msg += vMsg[uOffsetHistoConfig].size();
throw std::runtime_error(err_msg);
}
}
@@ -324,40 +431,42 @@ namespace cbm::services::histserv
uint32_t uOffsetCanvasConfig = pairHeader.second;
if (0 == pairHeader.second) {
uOffsetCanvasConfig = 1;
- if (0 < (parts.At(uOffsetHistoConfig + uOffsetCanvasConfig))->GetSize()) {
- fStopThread = true;
- std::string err_msg = "Application::ReceiveConfigAndData => No Canvas config expected but corresponding message";
- err_msg += " is not empty: ";
- err_msg += (parts.At(uOffsetHistoConfig + uOffsetCanvasConfig))->GetSize();
+ if (0 < vMsg[uOffsetHistoConfig + uOffsetCanvasConfig].size()) {
+ fStopThread = true;
+ bHistoServerStop = true;
+ std::string err_msg = "Application::ReceiveConfigAndData => No Canvas config expected but corresponding ";
+ err_msg += " message is not empty: ";
+ err_msg += vMsg[uOffsetHistoConfig + uOffsetCanvasConfig].size();
throw std::runtime_error(err_msg);
}
}
- if (static_cast<size_t>(parts.Size()) != 1 + uOffsetHistoConfig + uOffsetCanvasConfig + 1) {
- fStopThread = true;
- std::string err_msg = "Application::ReceiveConfigAndData => Number of parts not matching header: ";
- err_msg += parts.Size();
- err_msg += " instead of ";
+ if ((1 + uOffsetHistoConfig + uOffsetCanvasConfig + 1) != vMsg.size()) {
+ fStopThread = true;
+ bHistoServerStop = true;
+ std::string err_msg = "Application::ReceiveConfigAndData => Nb parts in message not matching configs numbers ";
+ err_msg += " declared in header";
+ err_msg += vMsg.size();
+ err_msg += " VS ";
err_msg += 1 + uOffsetHistoConfig + uOffsetCanvasConfig + 1;
throw std::runtime_error(err_msg);
- } // if( parts.Size() != 1 + pairHeader.first + pairHeader.second )
+ }
- /// Decode parts for histograms configuration
+ /// Decode parts for histograms configuration (auto-skip empty message if 0 declared in header)
for (uint32_t uHisto = 0; uHisto < pairHeader.first; ++uHisto) {
- ReceiveHistoConfig(parts.At(1 + uHisto), 0);
+ ReceiveHistoConfig(vMsg[1 + uHisto]);
} // for (UInt_t uHisto = 0; uHisto < pairHeader.first; ++uHisto)
+ LOG(debug) << "Application::ReceiveConfigAndData => Processed configuration for " << pairHeader.first << " histos";
- /// Decode parts for histograms configuration
+ /// Decode parts for histograms configuration (auto-skip empty message if 0 declared in header)
for (uint32_t uCanv = 0; uCanv < pairHeader.second; ++uCanv) {
- ReceiveCanvasConfig(parts.At(1 + uOffsetHistoConfig + uCanv), 0);
+ ReceiveCanvasConfig(vMsg[1 + uOffsetHistoConfig + uCanv]);
} // for (UInt_t uCanv = 0; uCanv < pairHeader.second; ++uCanv)
+ LOG(debug) << "Application::ReceiveConfigAndData => Processed configuration for " << pairHeader.second
+ << " canvases";
/// Decode the histograms data now that the configuration is loaded
- ReceiveData(parts.At(1 + uOffsetHistoConfig + uOffsetCanvasConfig), 0);
- *
- LOG(info) << "Application::ReceiveConfigAndData => Finished processing composed message with " << parts.Size()
- << " parts";
- */
+ ReceiveData(vMsg[1 + uOffsetHistoConfig + uOffsetCanvasConfig]);
return true;
}
@@ -432,7 +541,7 @@ namespace cbm::services::histserv
bool Application::PrepareCanvas(uint32_t uCanvIdx)
{
- LOG(debug) << " Extracting configuration for canvas index " << uCanvIdx;
+ LOG(info) << " Extracting configuration for canvas index " << uCanvIdx;
CanvasConfig conf(ExtractCanvasConfigFromString(fvpsCanvasConfig[uCanvIdx].second));
/// First check if all objects to be drawn are present
@@ -546,29 +655,35 @@ namespace cbm::services::histserv
/// Register the histos in the HTTP server
for (UInt_t uHisto = 0; uHisto < fvHistos.size(); ++uHisto) {
- /// Make sure we end up in chosen folder
- TString sFolder = fvHistos[uHisto].second.data();
- if (nullptr == gDirectory->Get(sFolder)) { //
- gDirectory->mkdir(sFolder);
+ /// catch case of histograms declared in config but not yet received
+ if (nullptr != fvHistos[uHisto].first) {
+ /// Make sure we end up in chosen folder
+ TString sFolder = fvHistos[uHisto].second.data();
+ if (nullptr == gDirectory->Get(sFolder)) { //
+ gDirectory->mkdir(sFolder);
+ }
+ gDirectory->cd(sFolder);
+
+ /// Write plot
+ fvHistos[uHisto].first->Write();
}
- gDirectory->cd(sFolder);
-
- /// Write plot
- fvHistos[uHisto].first->Write();
histoFile->cd();
} // for( UInt_t uHisto = 0; uHisto < fvHistos.size(); ++uHisto )
for (UInt_t uCanvas = 0; uCanvas < fvCanvas.size(); ++uCanvas) {
- /// Make sure we end up in chosen folder
- TString sFolder = fvCanvas[uCanvas].second.data();
- if (nullptr == gDirectory->Get(sFolder)) { //
- gDirectory->mkdir(sFolder);
+ /// catch case of canvases declared in config but for which not all histos were yet received
+ if (nullptr != fvCanvas[uCanvas].first) {
+ /// Make sure we end up in chosen folder
+ TString sFolder = fvCanvas[uCanvas].second.data();
+ if (nullptr == gDirectory->Get(sFolder)) { //
+ gDirectory->mkdir(sFolder);
+ }
+ gDirectory->cd(sFolder);
+
+ /// Write plot
+ fvCanvas[uCanvas].first->Write();
}
- gDirectory->cd(sFolder);
-
- /// Write plot
- fvCanvas[uCanvas].first->Write();
histoFile->cd();
} // for( UInt_t uHisto = 0; uHisto < fvCanvas.size(); ++uHisto )
diff --git a/services/histserv/app/Application.h b/services/histserv/app/Application.h
index 2b8168df58fdbc9c9dba17d0f659ad759d2e1bd3..fb52caec36c9c6a36839765401d89218110bb1c3 100644
--- a/services/histserv/app/Application.h
+++ b/services/histserv/app/Application.h
@@ -8,9 +8,9 @@
#include "THttpServer.h"
#include "TObjArray.h"
-#include <memory>
#include <string>
#include <thread>
+#include <zmq.hpp>
#include "ProgramOptions.h"
@@ -43,15 +43,12 @@ namespace cbm::services::histserv
void UpdateHttpServer();
private:
- const std::string& OutputFile() const;
- const std::string& ParamFile() const;
- const std::string& SetupTag() const;
//const std::string& ConfigFile() const;
- bool ReceiveData(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/);
- bool ReceiveHistoConfig(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/);
- bool ReceiveCanvasConfig(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/);
- bool ReceiveConfigAndData(/*FIXME_SOMETHING_To_Replace_FairMQ& parts*/);
+ bool ReceiveData(zmq::message_t& msg);
+ bool ReceiveHistoConfig(zmq::message_t& msg);
+ bool ReceiveCanvasConfig(zmq::message_t& msg);
+ bool ReceiveConfigAndData(std::vector<zmq::message_t>& vMsg);
template<class HistoT>
bool ReadHistogram(HistoT* pHist);
@@ -67,6 +64,10 @@ namespace cbm::services::histserv
std::thread fThread;
bool fStopThread = false;
+ /// Interface
+ zmq::context_t fZmqContext {1};
+ zmq::socket_t fZmqSocket {fZmqContext, ZMQ_PULL};
+
/// Array of histograms with unique names
TObjArray fArrayHisto;
/// Vector of string with ( HistoName, FolderPath ) to configure the histogram
diff --git a/services/histserv/app/CMakeLists.txt b/services/histserv/app/CMakeLists.txt
index b614872a252d4a900e0b7d2ca64a2a34822da4c4..3d240995634bd1583d81586d23f26460f34802de 100644
--- a/services/histserv/app/CMakeLists.txt
+++ b/services/histserv/app/CMakeLists.txt
@@ -38,6 +38,8 @@ target_link_libraries(histserv_nofairmq
ROOT::Hist
ROOT::RIO
ROOT::RHTTP
+ libzmq
+ cppzmq
)
install(TARGETS histserv_nofairmq DESTINATION bin)
diff --git a/services/histserv/app/ProgramOptions.cxx b/services/histserv/app/ProgramOptions.cxx
index d5826bbe284c49fb3331d05d9b862264729a134c..cfe378e94af9b114fa763a04b2959f744ec929a5 100644
--- a/services/histserv/app/ProgramOptions.cxx
+++ b/services/histserv/app/ProgramOptions.cxx
@@ -34,8 +34,9 @@ namespace cbm::services::histserv
// --- Define configuration options
po::options_description config("Configuration");
auto config_add = config.add_options();
- config_add("input,i", po::value<string>(&fsChanHistosIn)->value_name("<???????>"),
- "name or host:port or whatever is needed for input channel (histos/canvases config and data)");
+ config_add("input,i", po::value<string>(&fsChanHistosIn)->value_name("<protocol://xxxxxx>"),
+ "name or host:port or whatever is needed for input channel (histos/canvases config and data), "
+ " cf http://api.zeromq.org/2-1:zmq-bind");
config_add("port,p", po::value<uint32_t>(&fuHttpServerPort)->default_value(8080),
"port on which the http ROOT server (JSroot) will be available");
config_add("output,o", po::value<string>(&fsHistoFileName)->value_name("<file name>"),