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

Update JSON files & readme, use FairMQDevicePtr, cleanup.

Browse Source
This commit is contained in:
Alexey Rybalchenko
2016-10-28 15:42:15 +02:00
committed by Mohammad Al-Turany
parent d2606f0821
commit 72eb775983
21 changed files with 41 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -22,36 +22,22 @@ using namespace std;
FairMQExample6Sink::FairMQExample6Sink()
{
OnData("broadcast", &FairMQExample6Sink::HandleBroadcast);
OnData("data", &FairMQExample6Sink::HandleData);
}
void FairMQExample6Sink::Run()
bool FairMQExample6Sink::HandleBroadcast(FairMQMessagePtr& msg, int /*index*/)
{
std::unique_ptr<FairMQPoller> poller(fTransportFactory->CreatePoller(fChannels, { "data", "broadcast" }));
LOG(INFO) << "Received broadcast: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
while (CheckCurrentState(RUNNING))
{
poller->Poll(100);
return true;
}
if (poller->CheckInput("broadcast", 0))
{
FairMQMessagePtr msg(NewMessage());
bool FairMQExample6Sink::HandleData(FairMQMessagePtr& msg, int /*index*/)
{
LOG(INFO) << "Received message: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
if (Receive(msg, "broadcast") > 0)
{
LOG(INFO) << "Received broadcast: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
}
}
if (poller->CheckInput("data", 0))
{
FairMQMessagePtr msg(NewMessage());
if (Receive(msg, "data") > 0)
{
LOG(INFO) << "Received message: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
}
}
}
return true;
}
FairMQExample6Sink::~FairMQExample6Sink()