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

Add orthogonal OK/ERROR states.

Browse Source 
Replace state check mutex with atomic.

Update DDS example documentation.
This commit is contained in:
Alexey Rybalchenko
2015-08-24 17:35:30 +02:00
committed by Mohammad Al-Turany
parent a7ab33a10e
commit fbf7dbf2ba
38 changed files with 838 additions and 615 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
fairmq/examples/2-sampler-processor-sink/FairMQExample2Processor.cxx
View File

@@ -18,6 +18,8 @@
#include "FairMQExample2Processor.h"
#include "FairMQLogger.h"
using namespace std;
FairMQExample2Processor::FairMQExample2Processor()
: fText()
{
@@ -25,25 +27,33 @@ FairMQExample2Processor::FairMQExample2Processor()
void FairMQExample2Processor::CustomCleanup(void *data, void *object)
{
delete (std::string*)object;
delete (string*)object;
}
void FairMQExample2Processor::Run()
{
// Check if we are still in the RUNNING state
while (CheckCurrentState(RUNNING))
{
FairMQMessage* input = fTransportFactory->CreateMessage();
fChannels.at("data-in").at(0).Receive(input);
// Create empty message to hold the input
unique_ptr<FairMQMessage> input(fTransportFactory->CreateMessage());
LOG(INFO) << "Received data, processing...";
// Receive the message (blocks until received or interrupted (e.g. by state change)).
// Returns size of the received message or -1 if interrupted.
if (fChannels.at("data-in").at(0).Receive(input) > 0)
{
LOG(INFO) << "Received data, processing...";
std::string* text = new std::string(static_cast<char*>(input->GetData()), input->GetSize());
*text += " (modified by " + fId + ")";
// Modify the received string
string* text = new string(static_cast<char*>(input->GetData()), input->GetSize());
*text += " (modified by " + fId + ")";
delete input;
// Create output message
unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text));
FairMQMessage* msg = fTransportFactory->CreateMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text);
fChannels.at("data-out").at(0).Send(msg);
// Send out the output message
fChannels.at("data-out").at(0).Send(msg);
}
}
}

13
fairmq/examples/2-sampler-processor-sink/FairMQExample2Sampler.cxx
View File

@@ -18,6 +18,8 @@
#include "FairMQExample2Sampler.h"
#include "FairMQLogger.h"
using namespace std;
FairMQExample2Sampler::FairMQExample2Sampler()
: fText()
{
@@ -25,18 +27,19 @@ FairMQExample2Sampler::FairMQExample2Sampler()
void FairMQExample2Sampler::CustomCleanup(void *data, void *object)
{
delete (std::string*)object;
delete (string*)object;
}
void FairMQExample2Sampler::Run()
{
// Check if we are still in the RUNNING state
while (CheckCurrentState(RUNNING))
{
boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
std::string* text = new std::string(fText);
string* text = new string(fText);
FairMQMessage* msg = fTransportFactory->CreateMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text);
unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text));
LOG(INFO) << "Sending \"" << fText << "\"";
@@ -48,7 +51,7 @@ FairMQExample2Sampler::~FairMQExample2Sampler()
{
}
void FairMQExample2Sampler::SetProperty(const int key, const std::string& value)
void FairMQExample2Sampler::SetProperty(const int key, const string& value)
{
switch (key)
{
@@ -61,7 +64,7 @@ void FairMQExample2Sampler::SetProperty(const int key, const std::string& value)
}
}
std::string FairMQExample2Sampler::GetProperty(const int key, const std::string& default_ /*= ""*/)
string FairMQExample2Sampler::GetProperty(const int key, const string& default_ /*= ""*/)
{
switch (key)
{

15
fairmq/examples/2-sampler-processor-sink/FairMQExample2Sink.cxx
View File

@@ -28,15 +28,14 @@ void FairMQExample2Sink::Run()
{
while (CheckCurrentState(RUNNING))
{
FairMQMessage* msg = fTransportFactory->CreateMessage();
unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
fChannels.at("data-in").at(0).Receive(msg);
LOG(INFO) << "Received message: \""
<< string(static_cast<char*>(msg->GetData()), msg->GetSize())
<< "\"";
delete msg;
if (fChannels.at("data-in").at(0).Receive(msg) > 0)
{
LOG(INFO) << "Received message: \""
<< string(static_cast<char*>(msg->GetData()), msg->GetSize())
<< "\"";
}
}
}