#include <PndMQMerger.h>

Inheritance diagram for PndMQMerger:

Public Member Functions
	PndMQMerger ()

virtual	~PndMQMerger ()

template<class Archive >
void	serialize (Archive &ar, const unsigned int version)

Protected Member Functions
virtual void	Run ()

virtual void	ProcessData (std::map< std::string, BurstData > &dataToProcess)=0

Protected Attributes
BurstData *	fOutputData

BurstData	fInputData

std::map< int, std::map < std::string, BurstData > >	fInputMap

bool	fHasBoostSerialization

std::vector< bool >	fRunningStatus

Friends
class	boost::serialization::access

Detailed Description

PndMQMerger.h

Since: 2012-12-06

Author: D. Klein, A. Rybalchenko

Definition at line 27 of file PndMQMerger.h.

Constructor & Destructor Documentation

PndMQMerger::PndMQMerger ( )

Definition at line 27 of file PndMQMerger.cxx.

References fHasBoostSerialization.

                          : fHasBoostSerialization(false), fOutputData(0)
 {
         using namespace baseMQ::tools::resolve;
         // coverity[pointless_expression]: suppress coverity warnings on apparant if(const).
         if (has_BoostSerialization<FairTimeStamp*, void(boost::archive::binary_iarchive&, const unsigned int)>::value == 1)
                 fHasBoostSerialization = true;
 }

PndMQMerger::~PndMQMerger ( )

virtual

Definition at line 35 of file PndMQMerger.cxx.

36 {

37 }

Member Function Documentation

virtual void PndMQMerger::ProcessData ( std::map< std::string, BurstData > & dataToProcess )

protectedpure virtual

Implemented in PndMQMergerTest.

Referenced by Run().

void PndMQMerger::Run ( )

protectedvirtual

Definition at line 45 of file PndMQMerger.cxx.

References BurstHeader::fBranchName, BurstHeader::fBurstID, BurstData::fHeader, fInputData, fInputMap, and ProcessData().

 {
     int direction = 0;
     int numInputs = fChannels.at("data-in").size();
 //    fRunningStatus.resize(numInputs);
 
     LOG(INFO) << "Number of Input Channels: " << numInputs;
 
 //    boost::this_thread::sleep(boost::posix_time::milliseconds(10000));
 
     std::unique_ptr<FairMQPoller> poller(fTransportFactory->CreatePoller(fChannels, {"data-in"}));
 
     int nMessages;
     while (CheckCurrentState(RUNNING))
     {
         poller->Poll(-1);
         for (int channelNr = 0; channelNr < numInputs; channelNr++){
                 LOG(INFO) << "---- Reading channel " << channelNr << " ----";
                 if (poller->CheckInput("data-in", channelNr)){
                                 std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
 
                                 nMessages++;
 
                                 LOG(INFO) << "Get Data for Channel: " << channelNr;
 
                                 if (Receive(msg, "data-in", channelNr) >= 0){
                                         LOG(INFO) << "---Data Received--- " << msg->GetSize();
 
                                         std::string msgStr(static_cast<char*>(msg->GetData()), msg->GetSize());
 //                                      LOG(INFO) << msgStr;
                                         std::istringstream ibuffer(msgStr);
 
                                         try {
                                                 boost::archive::binary_iarchive InputArchive(ibuffer);
                                                 InputArchive >> fInputData;
                                         }
                                         catch (boost::archive::archive_exception& e)
                                         {
                                                 LOG(ERROR) << e.what();
 //                                              continue;
                                         }
                                         LOG(INFO) << "Data: " << fInputData.fHeader.fBurstID << " " << fInputData.fHeader.fBranchName;
 //                                      LOG(INFO) << "DataSize: " << fInputData.fData.size() << " " << fInputData.fData[0].size();
 //                                      for (auto gapItr : fInputData.fData){
 //                                              for (auto dataItr : gapItr)
 //                                                      LOG(INFO) << "TimeStamp: " << dataItr->GetTimeStamp();
 //                                      }
 
                                         fInputMap[fInputData.fHeader.fBurstID][fInputData.fHeader.fBranchName] = fInputData;
                                         LOG(INFO) << "InputMap for BurstID: " << fInputData.fHeader.fBurstID << " " << fInputMap[fInputData.fHeader.fBurstID].size();
 
                                         if (fInputMap[fInputData.fHeader.fBurstID].size() == numInputs){                // all input channels have delivered data for this burstID
                                                 //std::map<std::string, BurstData> dataToProcess = fInputMap[fInputData.fHeader.fBurstID];
         //                                      fInputMap.erase(fInputMap.find(fInputData.fHeader.fBurstID));
                                                 ProcessData(fInputMap[fInputData.fHeader.fBurstID]);
 
         //                                      if (fOutputData != 0){
         //                                              std::ostringstream obuffer;
         //                                              boost::archive::binary_oarchive OutputArchive(obuffer);
         //                                              OutputArchive << *fOutputData;
         //                                              int outputSize = obuffer.str().length();
         //                                              unique_ptr<FairMQMessage> msgOut(NewMessage(const_cast<char*>(obuffer.str().c_str()), outputSize, CustomClean, fOutputData));
         //                                              Send(msgOut, "data-out");
         //                                      }
                                         }
                                 }
                 }
                 }
         }
 }

template<class Archive >

void PndMQMerger::serialize	(	Archive &	ar,
		const unsigned int	version
	)

inline

Definition at line 34 of file PndMQMerger.h.

References fInputData, and fOutputData.

         {
         ar& fInputData;
                 ar& fOutputData;
         }

Friends And Related Function Documentation

friend class boost::serialization::access

friend

Definition at line 49 of file PndMQMerger.h.

Member Data Documentation

bool PndMQMerger::fHasBoostSerialization

protected

Definition at line 50 of file PndMQMerger.h.

Referenced by PndMQMerger().

BurstData PndMQMerger::fInputData

protected

Definition at line 43 of file PndMQMerger.h.

Referenced by Run(), and serialize().

std::map<int, std::map<std::string, BurstData> > PndMQMerger::fInputMap

protected

Definition at line 44 of file PndMQMerger.h.

Referenced by Run().

BurstData* PndMQMerger::fOutputData

protected

Definition at line 42 of file PndMQMerger.h.

Referenced by serialize().

std::vector<bool> PndMQMerger::fRunningStatus

protected

Definition at line 53 of file PndMQMerger.h.

The documentation for this class was generated from the following files:

Public Member Functions

Protected Member Functions

Protected Attributes

Friends

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation

Member Data Documentation