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

Enable new callback API

Browse Source 
- OnData() channel data handler.
 - ConditionalRun() for devices without incoming data.
 - Header file with common main(), to be extended with getDevice/addCustomOptions.
 - Update examples (MQ/Tutorial3) to use the new API and config.
 - NewSimpleMessage() for simpler creation of small messages (additional copy).
 - Replace SetProperty/GetProperty with fConfig access.
 - Runtime configurable channel names for common devices.
 - Configurable logging interval per channel.
 - FairMQMultiplier for distributing same data to multiple outputs.
 - Cleanup state machine messages.
 - Cmd option to toggle signal handling.
 - Simpler API for send/receive timeouts.
 - Enable --log-to-file.
 - Fix coverity issues, warnings.
 - Various code cleanup and minor tweaks.
This commit is contained in:
Alexey Rybalchenko
2016-08-10 09:47:53 +02:00
parent a01c4368b7
commit a1424e13fb
53 changed files with 501 additions and 1263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
examples/MQ/5-req-rep/FairMQExample5Client.cxx
View File

@@ -12,11 +12,13 @@
* @author A. Rybalchenko
*/
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <memory> // unique_ptr
#include <thread> // this_thread::sleep_for
#include <chrono>
#include "FairMQExample5Client.h"
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
using namespace std;
@@ -25,79 +27,41 @@ FairMQExample5Client::FairMQExample5Client()
{
}
void FairMQExample5Client::InitTask()
{
fText = fConfig->GetValue<string>("text");
}
bool FairMQExample5Client::ConditionalRun()
{
this_thread::sleep_for(chrono::seconds(1));
string* text = new string(fText);
// create message object with a pointer to the data buffer,
// its size,
// custom deletion function (called when transfer is done),
// and pointer to the object managing the data buffer
FairMQMessagePtr request(NewMessage(const_cast<char*>(text->c_str()), // data
text->length(), // size
[](void* /*data*/, void* object) { delete static_cast<string*>(object); }, // deletion callback
text)); // object that manages the data
FairMQMessagePtr reply(NewMessage());
LOG(INFO) << "Sending \"" << fText << "\" to server.";
if (Send(request, "data") > 0)
{
if (Receive(reply, "data") >= 0)
{
LOG(INFO) << "Received reply from server: \"" << string(static_cast<char*>(reply->GetData()), reply->GetSize()) << "\"";
return true;
}
}
return false;
}
FairMQExample5Client::~FairMQExample5Client()
{
}
void FairMQExample5Client::CustomCleanup(void* /*data*/, void *hint)
{
delete static_cast<string*>(hint);
}
void FairMQExample5Client::Run()
{
while (CheckCurrentState(RUNNING))
{
boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
string* text = new string(fText);
unique_ptr<FairMQMessage> request(NewMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text));
unique_ptr<FairMQMessage> reply(NewMessage());
LOG(INFO) << "Sending \"" << fText << "\" to server.";
if (Send(request, "data") > 0)
{
if (Receive(reply, "data") >= 0)
{
LOG(INFO) << "Received reply from server: \"" << string(static_cast<char*>(reply->GetData()), reply->GetSize()) << "\"";
}
}
}
}
void FairMQExample5Client::SetProperty(const int key, const string& value)
{
switch (key)
{
case Text:
fText = value;
break;
default:
FairMQDevice::SetProperty(key, value);
break;
}
}
string FairMQExample5Client::GetProperty(const int key, const string& default_ /*= ""*/)
{
switch (key)
{
case Text:
return fText;
break;
default:
return FairMQDevice::GetProperty(key, default_);
}
}
void FairMQExample5Client::SetProperty(const int key, const int value)
{
switch (key)
{
default:
FairMQDevice::SetProperty(key, value);
break;
}
}
int FairMQExample5Client::GetProperty(const int key, const int default_ /*= 0*/)
{
switch (key)
{
default:
return FairMQDevice::GetProperty(key, default_);
}
}

15
examples/MQ/5-req-rep/FairMQExample5Client.h
View File

@@ -22,25 +22,14 @@
class FairMQExample5Client : public FairMQDevice
{
public:
enum
{
Text = FairMQDevice::Last,
Last
};
FairMQExample5Client();
virtual ~FairMQExample5Client();
static void CustomCleanup(void* data, void* hint);
virtual void SetProperty(const int key, const std::string& value);
virtual std::string GetProperty(const int key, const std::string& default_ = "");
virtual void SetProperty(const int key, const int value);
virtual int GetProperty(const int key, const int default_ = 0);
protected:
std::string fText;
virtual void Run();
virtual bool ConditionalRun();
virtual void InitTask();
};
#endif /* FAIRMQEXAMPLECLIENT_H_ */

39
examples/MQ/5-req-rep/FairMQExample5Server.cxx
View File

@@ -12,9 +12,6 @@
* @author A. Rybalchenko
*/
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include "FairMQExample5Server.h"
#include "FairMQLogger.h"
@@ -22,32 +19,28 @@ using namespace std;
FairMQExample5Server::FairMQExample5Server()
{
OnData("data", &FairMQExample5Server::HandleData);
}
void FairMQExample5Server::CustomCleanup(void* /*data*/, void *hint)
bool FairMQExample5Server::HandleData(FairMQMessagePtr& request, int /*index*/)
{
delete static_cast<string*>(hint);
}
LOG(INFO) << "Received request from client: \"" << string(static_cast<char*>(request->GetData()), request->GetSize()) << "\"";
void FairMQExample5Server::Run()
{
while (CheckCurrentState(RUNNING))
string* text = new string("Thank you for the \"" + string(static_cast<char*>(request->GetData()), request->GetSize()) + "\"!");
LOG(INFO) << "Sending reply to client.";
FairMQMessagePtr reply(NewMessage(const_cast<char*>(text->c_str()), // data
text->length(), // size
[](void* /*data*/, void* object) { delete static_cast<string*>(object); }, // deletion callback
text)); // object that manages the data
if (Send(reply, "data") > 0)
{
unique_ptr<FairMQMessage> request(NewMessage());
if (Receive(request, "data") >= 0)
{
LOG(INFO) << "Received request from client: \"" << string(static_cast<char*>(request->GetData()), request->GetSize()) << "\"";
string* text = new string("Thank you for the \"" + string(static_cast<char*>(request->GetData()), request->GetSize()) + "\"!");
LOG(INFO) << "Sending reply to client.";
unique_ptr<FairMQMessage> reply(NewMessage(const_cast<char*>(text->c_str()), text->length(), CustomCleanup, text));
Send(reply, "data");
}
return true;
}
return false;
}
FairMQExample5Server::~FairMQExample5Server()

4
examples/MQ/5-req-rep/FairMQExample5Server.h
View File

@@ -23,10 +23,8 @@ class FairMQExample5Server : public FairMQDevice
FairMQExample5Server();
virtual ~FairMQExample5Server();
static void CustomCleanup(void *data, void* hint);
protected:
virtual void Run();
bool HandleData(FairMQMessagePtr&, int);
};
#endif /* FAIRMQEXAMPLE5SERVER_H_ */

56
examples/MQ/5-req-rep/runExample5Client.cxx
View File

@@ -5,55 +5,19 @@
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* runExampleClient.cxx
*
* @since 2013-04-23
* @author D. Klein, A. Rybalchenko
*/
#include "boost/program_options.hpp"
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
#include "runFairMQDevice.h"
#include "FairMQExample5Client.h"
using namespace boost::program_options;
namespace bpo = boost::program_options;
int main(int argc, char** argv)
void addCustomOptions(bpo::options_description& options)
{
try
{
std::string text;
options_description clientOptions("Client options");
clientOptions.add_options()
("text", value<std::string>(&text)->default_value("Hello"), "Text to send out");
FairMQProgOptions config;
config.AddToCmdLineOptions(clientOptions);
config.ParseAll(argc, argv);
FairMQExample5Client client;
client.CatchSignals();
client.SetConfig(config);
client.SetProperty(FairMQExample5Client::Text, text);
client.ChangeState("INIT_DEVICE");
client.WaitForEndOfState("INIT_DEVICE");
client.ChangeState("INIT_TASK");
client.WaitForEndOfState("INIT_TASK");
client.ChangeState("RUN");
client.InteractiveStateLoop();
}
catch (std::exception& e)
{
LOG(ERROR) << "Unhandled Exception reached the top of main: "
<< e.what() << ", application will now exit";
return 1;
}
return 0;
options.add_options()
("text", bpo::value<std::string>()->default_value("Hello"), "Text to send out");
}
FairMQDevice* getDevice(const FairMQProgOptions& /*config*/)
{
return new FairMQExample5Client();
}

46
examples/MQ/5-req-rep/runExample5Server.cxx
View File

@@ -5,45 +5,17 @@
* GNU Lesser General Public Licence version 3 (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* runExampleServer.cxx
*
* @since 2013-04-23
* @author D. Klein, A. Rybalchenko
*/
#include <iostream>
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
#include "runFairMQDevice.h"
#include "FairMQExample5Server.h"
int main(int argc, char** argv)
namespace bpo = boost::program_options;
void addCustomOptions(bpo::options_description& /*options*/)
{
try
{
FairMQProgOptions config;
config.ParseAll(argc, argv);
FairMQExample5Server server;
server.CatchSignals();
server.SetConfig(config);
server.ChangeState("INIT_DEVICE");
server.WaitForEndOfState("INIT_DEVICE");
server.ChangeState("INIT_TASK");
server.WaitForEndOfState("INIT_TASK");
server.ChangeState("RUN");
server.InteractiveStateLoop();
}
catch (std::exception& e)
{
LOG(ERROR) << "Unhandled Exception reached the top of main: "
<< e.what() << ", application will now exit";
return 1;
}
return 0;
}
FairMQDevice* getDevice(const FairMQProgOptions& /*config*/)
{
return new FairMQExample5Server();
}