Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-16 10:01:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Tests for MQ examples

Browse Source
This commit is contained in:
Alexey Rybalchenko
2017-08-23 11:12:29 +02:00
parent f6ee1cdf57
commit bd8be1436e
57 changed files with 658 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
examples/MQ/6-multiple-channels/FairMQExample6Sink.cxx
View File

@@ -15,27 +15,52 @@
#include <memory> // unique_ptr
#include "FairMQExample6Sink.h"
#include "FairMQPoller.h"
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
using namespace std;
FairMQExample6Sink::FairMQExample6Sink()
: fReceivedData(false)
, fReceivedBroadcast(false)
, fMaxIterations(0)
, fNumIterations(0)
{
OnData("broadcast", &FairMQExample6Sink::HandleBroadcast);
OnData("data", &FairMQExample6Sink::HandleData);
}
void FairMQExample6Sink::InitTask()
{
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
}
bool FairMQExample6Sink::HandleBroadcast(FairMQMessagePtr& msg, int /*index*/)
{
LOG(INFO) << "Received broadcast: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
fReceivedBroadcast = true;
return true;
return CheckIterations();
}
bool FairMQExample6Sink::HandleData(FairMQMessagePtr& msg, int /*index*/)
{
LOG(INFO) << "Received message: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
fReceivedData = true;
return CheckIterations();
}
bool FairMQExample6Sink::CheckIterations()
{
if (fMaxIterations > 0)
{
if (fReceivedData && fReceivedBroadcast && ++fNumIterations >= fMaxIterations)
{
LOG(INFO) << "Configured maximum number of iterations reached & Received messages from both sources. Leaving RUNNING state.";
return false;
}
}
return true;
}