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 device configuration

Browse Source 
- Move general config files out of example directory to fairmq/run.
 - Use FairMQProgOptions for MQ example 5.
 - Add SendPartAsync() for non-blocking send of a message part.
This commit is contained in:
Alexey Rybalchenko
2015-11-11 11:06:11 +01:00
committed by Mohammad Al-Turany
parent f0a878a43a
commit 837e035457
4 changed files with 130 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,7 +14,11 @@
#include <iostream>
#include "boost/program_options.hpp"
#include "FairMQLogger.h"
#include "FairMQParser.h"
#include "FairMQProgOptions.h"
#include "FairMQExample5Server.h"
#ifdef NANOMSG
@@ -24,40 +28,57 @@
#endif
using namespace std;
using namespace boost::program_options;
int main(int argc, char** argv)
{
FairMQExample5Server server;
server.CatchSignals();
LOG(INFO) << "PID: " << getpid();
FairMQProgOptions config;
try
{
if (config.ParseAll(argc, argv))
{
return 0;
}
string filename = config.GetValue<string>("config-json-file");
string id = config.GetValue<string>("id");
config.UserParser<FairMQParser::JSON>(filename, id);
server.fChannels = config.GetFairMQMap();
LOG(INFO) << "PID: " << getpid();
#ifdef NANOMSG
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
#else
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
#endif
server.SetTransport(transportFactory);
server.SetTransport(transportFactory);
server.SetProperty(FairMQExample5Server::Id, "server");
server.SetProperty(FairMQExample5Server::NumIoThreads, 1);
server.SetProperty(FairMQExample5Server::Id, id);
FairMQChannel replyChannel("rep", "bind", "tcp://*:5005");
replyChannel.UpdateSndBufSize(10000);
replyChannel.UpdateRcvBufSize(10000);
replyChannel.UpdateRateLogging(1);
server.ChangeState("INIT_DEVICE");
server.WaitForEndOfState("INIT_DEVICE");
server.fChannels["data"].push_back(replyChannel);
server.ChangeState("INIT_TASK");
server.WaitForEndOfState("INIT_TASK");
server.ChangeState("INIT_DEVICE");
server.WaitForEndOfState("INIT_DEVICE");
server.ChangeState("INIT_TASK");
server.WaitForEndOfState("INIT_TASK");
server.ChangeState("RUN");
server.InteractiveStateLoop();
server.ChangeState("RUN");
server.InteractiveStateLoop();
}
catch (exception& e)
{
LOG(ERROR) << e.what();
LOG(INFO) << "Command line options are the following: ";
config.PrintHelp();
return 1;
}
return 0;
}