PndMQMergerTest Class Reference

#include <PndMQMergerTest.h>

Inheritance diagram for PndMQMergerTest:

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

Protected Member Functions
virtual void	ProcessData (std::map< std::string, BurstData > &dataToProcess)
virtual void	Run ()

Protected Attributes
BurstData *	fOutputData
BurstData	fInputData
std::map< int, std::map < std::string, BurstData > >	fInputMap
bool	fHasBoostSerialization
std::vector< bool >	fRunningStatus

Detailed Description

PndMQMergerTest.h

Since

2012-12-06

Author

D. Klein, A. Rybalchenko

Definition at line 27 of file PndMQMergerTest.h.

Constructor & Destructor Documentation

PndMQMergerTest::PndMQMergerTest

(

)

Definition at line 26 of file PndMQMergerTest.cxx.

                                 : PndMQMerger()
{
}

Member Function Documentation

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

protectedvirtual

Implements PndMQMerger.

Definition at line 31 of file PndMQMergerTest.cxx.

{
        for (auto itr : dataToProcess){
                LOG(INFO) << "Data in Burst: " << itr.first;
//              for (auto eventItr : itr.second.fData)
//                      for (auto dataItr : eventItr)
//                              LOG(INFO) << dataItr->GetTimeStamp();
        }
        LOG(INFO) << "Finished!";
}

void PndMQMerger::Run ( )

protectedvirtualinherited

Definition at line 45 of file PndMQMerger.cxx.

{
    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  )

inlineinherited

Definition at line 34 of file PndMQMerger.h.

References PndMQMerger::fInputData, and PndMQMerger::fOutputData.

        {
        ar& fInputData;
                ar& fOutputData;
        }

Member Data Documentation

bool PndMQMerger::fHasBoostSerialization

protectedinherited

Definition at line 50 of file PndMQMerger.h.

Referenced by PndMQMerger::PndMQMerger().

BurstData PndMQMerger::fInputData

protectedinherited

Definition at line 43 of file PndMQMerger.h.

Referenced by PndMQMerger::serialize().

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

protectedinherited

Definition at line 44 of file PndMQMerger.h.

BurstData* PndMQMerger::fOutputData

protectedinherited

Definition at line 42 of file PndMQMerger.h.

Referenced by PndMQMerger::serialize().

std::vector<bool> PndMQMerger::fRunningStatus

protectedinherited

Definition at line 53 of file PndMQMerger.h.

---

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

• PndMQMergerTest.h
• PndMQMergerTest.cxx