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

Adjust example 1 docs

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-06-17 10:27:38 +02:00
parent d1c51e0f1f
commit beb510ded8
2 changed files with 17 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
examples/1-1/Sampler.cxx
View File

@ -17,8 +17,7 @@ Sampler::Sampler()
: fText() : fText()
, fMaxIterations(0) , fMaxIterations(0)
, fNumIterations(0) , fNumIterations(0)
{ {}
}
void Sampler::InitTask() void Sampler::InitTask()
{ {
@ -32,25 +31,22 @@ bool Sampler::ConditionalRun()
// create a copy of the data with new(), that will be deleted after the transfer is complete // create a copy of the data with new(), that will be deleted after the transfer is complete
string* text = new string(fText); string* text = new string(fText);
// create message object with a pointer to the data buffer, // create message object with a pointer to the data buffer, its size,
// its size,
// custom deletion function (called when transfer is done), // custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer // and pointer to the object managing the data buffer
FairMQMessagePtr msg(NewMessage(const_cast<char*>(text->c_str()), FairMQMessagePtr msg(NewMessage(
const_cast<char*>(text->c_str()),
text->length(), text->length(),
[](void* /*data*/, void* object) { delete static_cast<string*>(object); }, [](void* /*data*/, void* object) { delete static_cast<string*>(object); },
text)); text));
LOG(info) << "Sending \"" << fText << "\""; LOG(info) << "Sending \"" << fText << "\"";
// in case of error or transfer interruption, return false to go to IDLE state // in case of error or transfer interruption, return false to go to the Ready state
// successfull transfer will return number of bytes transfered (can be 0 if sending an empty message). // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
if (Send(msg, "data") < 0) if (Send(msg, "data") < 0) {
{
return false; return false;
} } else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations)
{
LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state."; LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
return false; return false;
} }
@ -58,8 +54,6 @@ bool Sampler::ConditionalRun()
return true; return true;
} }
Sampler::~Sampler() Sampler::~Sampler() {}
{
}
} // namespace example_1_1 } // namespace example_1_1

13
examples/1-1/Sink.cxx
View File

@ -33,23 +33,22 @@ void Sink::InitTask()
fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");
} }
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0) // handler is called whenever a message arrives on "data", with a reference to the message and a
// sub-channel index (here 0)
bool Sink::HandleData(FairMQMessagePtr& msg, int /*index*/) bool Sink::HandleData(FairMQMessagePtr& msg, int /*index*/)
{ {
LOG(info) << "Received: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\""; LOG(info) << "Received: \"" << string(static_cast<char*>(msg->GetData()), msg->GetSize()) << "\"";
if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
{
LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state."; LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
return false; return false;
} }
// return true if want to be called again (otherwise return false go to IDLE state) // return true if you want the handler to be called again (otherwise return false go to the
// Ready state)
return true; return true;
} }
Sink::~Sink() Sink::~Sink() {}
{
}
} // namespace example_1_1 } // namespace example_1_1