PndMQMerger.cxx

Go to the documentation of this file.

/********************************************************************************
 *    Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
 *                                                                              *
 *              This software is distributed under the terms of the             * 
 *         GNU Lesser General Public Licence version 3 (LGPL) version 3,        *  
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/archive/binary_oarchive.hpp>

#include "baseMQtools.h"

#include "FairMQLogger.h"
#include "PndMQMerger.h"
#include "FairMQPoller.h"

using namespace std;

PndMQMerger::PndMQMerger() : 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()
{
}

void CustomClean(void* data, void *hint)
{
        LOG(INFO) << "FREEMESSAGE called for data: " << static_cast<BurstData*>(hint)->fHeader.fBranchName;
        delete static_cast<BurstData*>(hint);
}

void PndMQMerger::Run()
{
    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");
        //                                      }
                                        }
                                }
                }
                }
        }
}