#include <PndMQBurstProcessor.h>

Inheritance diagram for PndMQBurstProcessor:

Public Member Functions
	PndMQBurstProcessor ()

virtual	~PndMQBurstProcessor ()

virtual void	UpdateParameters ()

virtual FairParGenericSet *	UpdateParameter (FairParGenericSet *thisPar)

virtual void	SetParameters ()

virtual void	ProcessData ()=0

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

Static Public Member Functions
static void	CustomCleanupParameters (void data, void hint)

static void	free_string (void data, void hint)

Protected Member Functions
virtual void	Run ()

Protected Attributes
BurstData	fBurstDataIn

BurstData	fBurstDataOut

int	fCurrentRunId

int	fNewRunId

TList *	fParCList

Private Attributes
bool	fHasBoostSerialization

Friends
class	boost::serialization::access

Detailed Description

Definition at line 50 of file PndMQBurstProcessor.h.

Constructor & Destructor Documentation

PndMQBurstProcessor::PndMQBurstProcessor ( )

inline

Definition at line 53 of file PndMQBurstProcessor.h.

                           :
         fHasBoostSerialization(false), fCurrentRunId(0), fNewRunId(0)
     {
         gSystem->ResetSignal(kSigInterrupt);
         gSystem->ResetSignal(kSigTermination);
 
         // Check if boost serialization is available if it is chosen
         using namespace baseMQ::tools::resolve;
         // coverity[pointless_expression]: suppress coverity warnings on apparant if(const).
         if (is_same<boost::archive::binary_iarchive, boost::archive::binary_iarchive>::value || is_same<boost::archive::binary_iarchive, boost::archive::text_iarchive>::value)
         {
             if (has_BoostSerialization<FairTimeStamp, void(boost::archive::binary_iarchive&, const unsigned int)>::value == 1)
             {
                 fHasBoostSerialization = true;
             }
         }
 
 
     }

virtual PndMQBurstProcessor::~PndMQBurstProcessor ( )

inlinevirtual

Definition at line 73 of file PndMQBurstProcessor.h.

     {
         fParCList->Clear();
     }

Member Function Documentation

void PndMQBurstProcessor::CustomCleanupParameters	(	void *	data,
		void *	hint
	)

static

Definition at line 26 of file PndMQBurstProcessor.cxx.

Referenced by UpdateParameter().

 {
     delete (std::string*)hint;
 }

void PndMQBurstProcessor::free_string	(	void *	data,
		void *	hint
	)

static

Definition at line 31 of file PndMQBurstProcessor.cxx.

Referenced by Run().

 {
         delete static_cast<std::string*>(hint);
 }

virtual void PndMQBurstProcessor::ProcessData ( )

pure virtual

Implemented in PndMQMvdPixelDigiProcessorBursts.

Referenced by Run().

void PndMQBurstProcessor::Run ( )

protectedvirtual

Definition at line 36 of file PndMQBurstProcessor.cxx.

References fBurstDataIn, fBurstDataOut, BurstHeader::fBurstID, fCurrentRunId, BurstData::fData, fHasBoostSerialization, BurstData::fHeader, fNewRunId, free_string(), BurstHeader::fRunID, ProcessData(), SetParameters(), and UpdateParameters().

 {
     if (fHasBoostSerialization)
     {
         while (CheckCurrentState(RUNNING))
         {
                  FairMQParts parts;
                  std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
                  if (Receive(msg, "data-in") >= 0)
                  {
                          LOG(INFO) << "Received data! " <<  msg->GetSize() << std::endl;
                                  string msgStr(static_cast<char*>(msg->GetData()), msg->GetSize());
                                  istringstream ibuffer(msgStr);
                                  if(!ibuffer.good())
                                          LOG(INFO) << "IBUFFER IS BAD!";
                                  try {
                                          boost::archive::binary_iarchive InputArchive(ibuffer);
 
                                         InputArchive >> fBurstDataIn;
                                  }
                                  catch (boost::archive::archive_exception& e)
                                 {
                                         LOG(ERROR) << e.what();
 //                                      continue;
                                 }
                                 fNewRunId = fBurstDataIn.fHeader.fRunID;
                                 if (fNewRunId != fCurrentRunId) {
                                         fCurrentRunId = fNewRunId;
                                         UpdateParameters();
                                         SetParameters();
                                 }
 
                                 if (fBurstDataIn.fData.size() < 1) continue;
 //                              LOG(INFO) << "BurstData size: " << fBurstDataIn.fData[0].size() << " BurstID: " << fBurstDataIn.fHeader.fBurstID;
 
                                 ProcessData();
                  }
                  if (fBurstDataOut.fData.size() > 0){
                          fBurstDataOut.fHeader.fBurstID = fBurstDataIn.fHeader.fBurstID;
                                 std::ostringstream obuffer;
                                 boost::archive::binary_oarchive OutputArchive(obuffer);
                                 OutputArchive << fBurstDataOut;
                                 std::string* strMsg = new std::string(obuffer.str());
                                 unique_ptr<FairMQMessage> msgOut(NewMessage(const_cast<char*>(strMsg->c_str()), strMsg->length(), free_string, strMsg));
                                 LOG(INFO) << "Data sent: " << fBurstDataOut.fHeader.fBranchName << " BurstID: " << fBurstDataOut.fHeader.fBurstID << " size: " << msgOut->GetSize();
                                 int event = 0;
 //                              for (auto eventItr : fBurstDataOut->fData){
 //                                      for (auto dataItr : eventItr){
 //                                              LOG(INFO) << event << " : " << dataItr->GetTimeStamp();
 //                                      }
 //                                      event++;
 //                              }
                                 Send(msgOut, "data-out");
                  }
         }
     }
     else
     {
         LOG(ERROR) << " Boost Serialization not ok";
     }
 }

template<class Archive >

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

inline

Definition at line 89 of file PndMQBurstProcessor.h.

     {
         ar& fBurstDataIn;
         ar& fBurstDataOut;
     }

virtual void PndMQBurstProcessor::SetParameters ( )

inlinevirtual

Reimplemented in PndMQMvdPixelDigiProcessorBursts.

Definition at line 81 of file PndMQBurstProcessor.h.

Referenced by Run().

81 {};

FairParGenericSet * PndMQBurstProcessor::UpdateParameter ( FairParGenericSet * thisPar )

virtual

Definition at line 106 of file PndMQBurstProcessor.cxx.

References CustomCleanupParameters(), and fCurrentRunId.

Referenced by UpdateParameters().

                                                                                   {
   std::string paramName = thisPar->GetName();
   //  boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
   std::string* reqStr = new std::string(paramName + "," + std::to_string(fCurrentRunId));
   LOG(WARN) << "Requesting parameter \"" << paramName << "\" for Run ID " << fCurrentRunId << " (" << thisPar << ")";
   std::unique_ptr<FairMQMessage> req(NewMessage(const_cast<char*>(reqStr->c_str()), reqStr->length(), CustomCleanupParameters, reqStr));
   std::unique_ptr<FairMQMessage> rep(NewMessage());
 
   if (Send(req,"param") > 0)
     {
       if (Receive(rep,"param") > 0)
         {
           TMessage2 tm(rep->GetData(), rep->GetSize());
           thisPar = (FairParGenericSet*)tm.ReadObject(tm.GetClass());
           LOG(WARN) << "Received parameter"<< paramName <<" from the server (" << thisPar << ")" << tm.GetClass()->GetName() << " DataSize: " << rep->GetSize();
           thisPar->print();
           return thisPar;
         }
     }
   return NULL;
 }

void PndMQBurstProcessor::UpdateParameters ( )

virtual

Definition at line 98 of file PndMQBurstProcessor.cxx.

References fParCList, and UpdateParameter().

Referenced by Run().

                                            {
   for ( int iparC = 0 ; iparC < fParCList->GetEntries() ; iparC++ ) {
     FairParGenericSet* tempObj = (FairParGenericSet*)(fParCList->At(iparC));
     fParCList->Remove(tempObj);
     fParCList->AddAt(UpdateParameter(tempObj),iparC);
   }
 }

Friends And Related Function Documentation

friend class boost::serialization::access

friend

Definition at line 109 of file PndMQBurstProcessor.h.

Member Data Documentation

BurstData PndMQBurstProcessor::fBurstDataIn

protected

Definition at line 97 of file PndMQBurstProcessor.h.

Referenced by PndMQMvdPixelDigiProcessorBursts::ProcessData(), and Run().

BurstData PndMQBurstProcessor::fBurstDataOut

protected

Definition at line 98 of file PndMQBurstProcessor.h.

Referenced by PndMQMvdPixelDigiProcessorBursts::ProcessData(), and Run().

int PndMQBurstProcessor::fCurrentRunId

protected

Definition at line 99 of file PndMQBurstProcessor.h.

Referenced by Run(), and UpdateParameter().

bool PndMQBurstProcessor::fHasBoostSerialization

private

Definition at line 110 of file PndMQBurstProcessor.h.

Referenced by Run().

int PndMQBurstProcessor::fNewRunId

protected

Definition at line 100 of file PndMQBurstProcessor.h.

Referenced by Run().

TList* PndMQBurstProcessor::fParCList

protected

Definition at line 101 of file PndMQBurstProcessor.h.

Referenced by PndMQMvdPixelDigiProcessorBursts::SetParameters(), and UpdateParameters().

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

Public Member Functions

Static Public Member Functions

Protected Member Functions

Protected Attributes

Private Attributes

Friends

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation

Member Data Documentation