Remove compile time transport interface switch

- Remove the compile time check of the transport implementation.
  The transport (zeromq/nanomsg) can be chosen at run time with:
  `device.SetTransport("zeromq"); // possible values are "zeromq" and "nanomsg"`.

  For devices that use FairMQProgOptions, the transport can be configured via cmd option:
  `--transport zeromq` or `--transport nanomsg`. Default values is "zeromq".
  The device receives the configured value with:
  `device.SetTransport(config.GetValue<std::string>("transport"));`

  Old method of setting transport still works. But the NANOMSG constant is not defined.

- Remove old `fairmq/prototest` directory. It was only used as a test for protobuf.
  The protobuf part of Tutorial3 does the same (with different values).

- Fix a bug in FairMQPollerNN, where the `revents` value was not initialized.
  This caused the `poller->CheckOutput()` to trigger when it should not.

fairmq/run/runBenchmarkSampler.cxx

@@ -22,12 +22,6 @@
 #include "FairMQProgOptions.h"
 #include "FairMQBenchmarkSampler.h"
 
-#ifdef NANOMSG
-#include "FairMQTransportFactoryNN.h"
-#else
-#include "FairMQTransportFactoryZMQ.h"
-#endif
-
 using namespace std;
 using namespace FairMQParser;
 using namespace boost::program_options;
@@ -65,11 +59,7 @@ int main(int argc, char** argv)
 
         LOG(INFO) << "PID: " << getpid();
 
-#ifdef NANOMSG
-        sampler.SetTransport(new FairMQTransportFactoryNN());
-#else
-        sampler.SetTransport(new FairMQTransportFactoryZMQ());
-#endif
+        sampler.SetTransport(config.GetValue<std::string>("transport"));
 
         sampler.SetProperty(FairMQBenchmarkSampler::Id, id);
         sampler.SetProperty(FairMQBenchmarkSampler::MsgSize, msgSize);

fairmq/run/runBuffer.cxx

@@ -19,23 +19,18 @@
 #include "FairMQLogger.h"
 #include "FairMQBuffer.h"
 
-#ifdef NANOMSG
-#include "FairMQTransportFactoryNN.h"
-#else
-#include "FairMQTransportFactoryZMQ.h"
-#endif
-
 using namespace std;
 
 typedef struct DeviceOptions
 {
     DeviceOptions() :
-        id(), ioThreads(0),
+        id(), ioThreads(0), transport(),
         inputSocketType(), inputBufSize(0), inputMethod(), inputAddress(),
         outputSocketType(), outputBufSize(0), outputMethod(), outputAddress() {}
 
     string id;
     int ioThreads;
+    string transport;
     string inputSocketType;
     int inputBufSize;
     string inputMethod;
@@ -56,6 +51,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
     desc.add_options()
         ("id", bpo::value<string>(), "Device ID")
         ("io-threads", bpo::value<int>()->default_value(1), "Number of I/O threads")
+        ("transport", bpo::value<string>()->default_value("zeromq"), "Transport (zeromq/nanomsg)")
         ("input-socket-type", bpo::value<string>()->required(), "Input socket type: sub/pull")
         ("input-buff-size", bpo::value<int>()->default_value(1000), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)")
         ("input-method", bpo::value<string>()->required(), "Input method: bind/connect")
@@ -69,7 +65,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
     bpo::variables_map vm;
     bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm);
 
-    if ( vm.count("help") )
+    if (vm.count("help"))
     {
         LOG(INFO) << "FairMQ Buffer" << endl << desc;
         return false;
@@ -77,35 +73,17 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
 
     bpo::notify(vm);
 
-    if ( vm.count("id") )
-        _options->id = vm["id"].as<string>();
-
-    if ( vm.count("io-threads") )
-        _options->ioThreads = vm["io-threads"].as<int>();
-
-    if ( vm.count("input-socket-type") )
-        _options->inputSocketType = vm["input-socket-type"].as<string>();
-
-    if ( vm.count("input-buff-size") )
-        _options->inputBufSize = vm["input-buff-size"].as<int>();
-
-    if ( vm.count("input-method") )
-        _options->inputMethod = vm["input-method"].as<string>();
-
-    if ( vm.count("input-address") )
-        _options->inputAddress = vm["input-address"].as<string>();
-
-    if ( vm.count("output-socket-type") )
-        _options->outputSocketType = vm["output-socket-type"].as<string>();
-
-    if ( vm.count("output-buff-size") )
-        _options->outputBufSize = vm["output-buff-size"].as<int>();
-
-    if ( vm.count("output-method") )
-        _options->outputMethod = vm["output-method"].as<string>();
-
-    if ( vm.count("output-address") )
-        _options->outputAddress = vm["output-address"].as<string>();
+    if (vm.count("id")) { _options->id = vm["id"].as<string>(); }
+    if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as<int>(); }
+    if (vm.count("transport"))  { _options->transport = vm["transport"].as<string>(); }
+    if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as<string>(); }
+    if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as<int>(); }
+    if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as<string>(); }
+    if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as<string>(); }
+    if (vm.count("output-socket-type")) { _options->outputSocketType = vm["output-socket-type"].as<string>(); }
+    if (vm.count("output-buff-size")) { _options->outputBufSize = vm["output-buff-size"].as<int>(); }
+    if (vm.count("output-method")) { _options->outputMethod = vm["output-method"].as<string>(); }
+    if (vm.count("output-address")) { _options->outputAddress = vm["output-address"].as<string>(); }
 
     return true;
 }
@@ -129,13 +107,7 @@ int main(int argc, char** argv)
 
     LOG(INFO) << "PID: " << getpid();
 
-#ifdef NANOMSG
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
-#else
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
-#endif
-
-    buffer.SetTransport(transportFactory);
+    buffer.SetTransport(options.transport);
 
     FairMQChannel inputChannel(options.inputSocketType, options.inputMethod, options.inputAddress);
     inputChannel.UpdateSndBufSize(options.inputBufSize);

fairmq/run/runMerger.cxx

@@ -19,23 +19,18 @@
 #include "FairMQLogger.h"
 #include "FairMQMerger.h"
 
-#ifdef NANOMSG
-#include "FairMQTransportFactoryNN.h"
-#else
-#include "FairMQTransportFactoryZMQ.h"
-#endif
-
 using namespace std;
 
 typedef struct DeviceOptions
 {
     DeviceOptions() :
-        id(), ioThreads(0), numInputs(0),
+        id(), ioThreads(0), transport(), numInputs(0),
         inputSocketType(), inputBufSize(), inputMethod(), inputAddress(),
         outputSocketType(), outputBufSize(0), outputMethod(), outputAddress() {}
 
     string id;
     int ioThreads;
+    string transport;
     int numInputs;
     vector<string> inputSocketType;
     vector<int> inputBufSize;
@@ -57,6 +52,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
     desc.add_options()
         ("id", bpo::value<string>()->required(), "Device ID")
         ("io-threads", bpo::value<int>()->default_value(1), "Number of I/O threads")
+        ("transport", bpo::value<string>()->default_value("zeromq"), "Transport (zeromq/nanomsg)")
         ("num-inputs", bpo::value<int>()->required(), "Number of Merger input sockets")
         ("input-socket-type", bpo::value<vector<string>>()->required(), "Input socket type: sub/pull")
         ("input-buff-size", bpo::value<vector<int>>()->required(), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)")
@@ -79,38 +75,18 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
 
     bpo::notify(vm);
 
-    if (vm.count("id"))
-        _options->id = vm["id"].as<string>();
-
-    if (vm.count("io-threads"))
-        _options->ioThreads = vm["io-threads"].as<int>();
-
-    if (vm.count("num-inputs"))
-        _options->numInputs = vm["num-inputs"].as<int>();
-
-    if (vm.count("input-socket-type"))
-        _options->inputSocketType = vm["input-socket-type"].as<vector<string>>();
-
-    if (vm.count("input-buff-size"))
-        _options->inputBufSize = vm["input-buff-size"].as<vector<int>>();
-
-    if (vm.count("input-method"))
-        _options->inputMethod = vm["input-method"].as<vector<string>>();
-
-    if (vm.count("input-address"))
-        _options->inputAddress = vm["input-address"].as<vector<string>>();
-
-    if (vm.count("output-socket-type"))
-        _options->outputSocketType = vm["output-socket-type"].as<string>();
-
-    if (vm.count("output-buff-size"))
-        _options->outputBufSize = vm["output-buff-size"].as<int>();
-
-    if (vm.count("output-method"))
-        _options->outputMethod = vm["output-method"].as<string>();
-
-    if (vm.count("output-address"))
-        _options->outputAddress = vm["output-address"].as<string>();
+    if (vm.count("id")) { _options->id = vm["id"].as<string>(); }
+    if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as<int>(); }
+    if (vm.count("transport"))  { _options->transport = vm["transport"].as<string>(); }
+    if (vm.count("num-inputs")) { _options->numInputs = vm["num-inputs"].as<int>(); }
+    if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as<vector<string>>(); }
+    if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as<vector<int>>(); }
+    if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as<vector<string>>(); }
+    if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as<vector<string>>(); }
+    if (vm.count("output-socket-type")) { _options->outputSocketType = vm["output-socket-type"].as<string>(); }
+    if (vm.count("output-buff-size")) { _options->outputBufSize = vm["output-buff-size"].as<int>(); }
+    if (vm.count("output-method")) { _options->outputMethod = vm["output-method"].as<string>(); }
+    if (vm.count("output-address")) { _options->outputAddress = vm["output-address"].as<string>(); }
 
     return true;
 }
@@ -134,13 +110,7 @@ int main(int argc, char** argv)
 
     LOG(INFO) << "PID: " << getpid();
 
-#ifdef NANOMSG
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
-#else
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
-#endif
-
-    merger.SetTransport(transportFactory);
+    merger.SetTransport(options.transport);
 
     for (unsigned int i = 0; i < options.inputAddress.size(); ++i)
     {

fairmq/run/runProxy.cxx

@@ -19,23 +19,18 @@
 #include "FairMQLogger.h"
 #include "FairMQProxy.h"
 
-#ifdef NANOMSG
-#include "FairMQTransportFactoryNN.h"
-#else
-#include "FairMQTransportFactoryZMQ.h"
-#endif
-
 using namespace std;
 
 typedef struct DeviceOptions
 {
     DeviceOptions() :
-        id(), ioThreads(0),
+        id(), ioThreads(0), transport(),
         inputSocketType(), inputBufSize(0), inputMethod(), inputAddress(),
         outputSocketType(), outputBufSize(0), outputMethod(), outputAddress() {}
 
     string id;
     int ioThreads;
+    string transport;
     string inputSocketType;
     int inputBufSize;
     string inputMethod;
@@ -56,6 +51,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
     desc.add_options()
         ("id", bpo::value<string>()->required(), "Device ID")
         ("io-threads", bpo::value<int>()->default_value(1), "Number of I/O threads")
+        ("transport", bpo::value<string>()->default_value("zeromq"), "Transport (zeromq/nanomsg)")
         ("input-socket-type", bpo::value<string>()->required(), "Input socket type: sub/pull")
         ("input-buff-size", bpo::value<int>()->default_value(1000), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)")
         ("input-method", bpo::value<string>()->required(), "Input method: bind/connect")
@@ -69,7 +65,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
     bpo::variables_map vm;
     bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm);
 
-    if ( vm.count("help") )
+    if (vm.count("help"))
     {
         LOG(INFO) << "FairMQ Proxy" << endl << desc;
         return false;
@@ -77,35 +73,17 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
 
     bpo::notify(vm);
 
-    if ( vm.count("id") )
-        _options->id = vm["id"].as<string>();
-
-    if ( vm.count("io-threads") )
-        _options->ioThreads = vm["io-threads"].as<int>();
-
-    if ( vm.count("input-socket-type") )
-        _options->inputSocketType = vm["input-socket-type"].as<string>();
-
-    if ( vm.count("input-buff-size") )
-        _options->inputBufSize = vm["input-buff-size"].as<int>();
-
-    if ( vm.count("input-method") )
-        _options->inputMethod = vm["input-method"].as<string>();
-
-    if ( vm.count("input-address") )
-        _options->inputAddress = vm["input-address"].as<string>();
-
-    if ( vm.count("output-socket-type") )
-        _options->outputSocketType = vm["output-socket-type"].as<string>();
-
-    if ( vm.count("output-buff-size") )
-        _options->outputBufSize = vm["output-buff-size"].as<int>();
-
-    if ( vm.count("output-method") )
-        _options->outputMethod = vm["output-method"].as<string>();
-
-    if ( vm.count("output-address") )
-        _options->outputAddress = vm["output-address"].as<string>();
+    if (vm.count("id")) { _options->id = vm["id"].as<string>(); }
+    if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as<int>(); }
+    if (vm.count("transport"))  { _options->transport = vm["transport"].as<string>(); }
+    if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as<string>(); }
+    if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as<int>(); }
+    if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as<string>(); }
+    if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as<string>(); }
+    if (vm.count("output-socket-type")) { _options->outputSocketType = vm["output-socket-type"].as<string>(); }
+    if (vm.count("output-buff-size")) { _options->outputBufSize = vm["output-buff-size"].as<int>(); }
+    if (vm.count("output-method")) { _options->outputMethod = vm["output-method"].as<string>(); }
+    if (vm.count("output-address")) { _options->outputAddress = vm["output-address"].as<string>(); }
 
     return true;
 }
@@ -129,13 +107,7 @@ int main(int argc, char** argv)
 
     LOG(INFO) << "PID: " << getpid();
 
-#ifdef NANOMSG
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
-#else
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
-#endif
-
-    proxy.SetTransport(transportFactory);
+    proxy.SetTransport(options.transport);
 
     FairMQChannel inputChannel(options.inputSocketType, options.inputMethod, options.inputAddress);
     inputChannel.UpdateSndBufSize(options.inputBufSize);

fairmq/run/runSink.cxx

@@ -21,12 +21,6 @@
 #include "FairMQProgOptions.h"
 #include "FairMQSink.h"
 
-#ifdef NANOMSG
-#include "FairMQTransportFactoryNN.h"
-#else
-#include "FairMQTransportFactoryZMQ.h"
-#endif
-
 using namespace std;
 using namespace FairMQParser;
 using namespace boost::program_options;
@@ -63,13 +57,7 @@ int main(int argc, char** argv)
 
         LOG(INFO) << "PID: " << getpid();
 
-#ifdef NANOMSG
-        FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
-#else
-        FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
-#endif
-
-        sink.SetTransport(transportFactory);
+        sink.SetTransport(config.GetValue<std::string>("transport"));
 
         sink.SetProperty(FairMQSink::Id, id);
         sink.SetProperty(FairMQSink::NumMsgs, numMsgs);

fairmq/run/runSplitter.cxx

@@ -19,24 +19,19 @@
 #include "FairMQLogger.h"
 #include "FairMQSplitter.h"
 
-#ifdef NANOMSG
-#include "FairMQTransportFactoryNN.h"
-#else
-#include "FairMQTransportFactoryZMQ.h"
-#endif
-
 using namespace std;
 
 typedef struct DeviceOptions
 {
     DeviceOptions() :
-        id(), ioThreads(0), numOutputs(0),
+        id(), ioThreads(0), transport(), numOutputs(0),
         inputSocketType(), inputBufSize(0), inputMethod(), inputAddress(),
         outputSocketType(), outputBufSize(), outputMethod(), outputAddress()
         {}
 
     string id;
     int ioThreads;
+    string transport;
     int numOutputs;
     string inputSocketType;
     int inputBufSize;
@@ -58,6 +53,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
     desc.add_options()
         ("id", bpo::value<string>()->required(), "Device ID")
         ("io-threads", bpo::value<int>()->default_value(1), "Number of I/O threads")
+        ("transport", bpo::value<string>()->default_value("zeromq"), "Transport (zeromq/nanomsg)")
         ("num-outputs", bpo::value<int>()->required(), "Number of Splitter output sockets")
         ("input-socket-type", bpo::value<string>()->required(), "Input socket type: sub/pull")
         ("input-buff-size", bpo::value<int>(), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)")
@@ -72,7 +68,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
     bpo::variables_map vm;
     bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm);
 
-    if ( vm.count("help") )
+    if (vm.count("help"))
     {
         LOG(INFO) << "FairMQ Splitter" << endl << desc;
         return false;
@@ -80,38 +76,18 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options)
 
     bpo::notify(vm);
 
-    if ( vm.count("id") )
-        _options->id = vm["id"].as<string>();
-
-    if ( vm.count("io-threads") )
-        _options->ioThreads = vm["io-threads"].as<int>();
-
-    if ( vm.count("num-outputs") )
-        _options->numOutputs = vm["num-outputs"].as<int>();
-
-    if ( vm.count("input-socket-type") )
-        _options->inputSocketType = vm["input-socket-type"].as<string>();
-
-    if ( vm.count("input-buff-size") )
-        _options->inputBufSize = vm["input-buff-size"].as<int>();
-
-    if ( vm.count("input-method") )
-        _options->inputMethod = vm["input-method"].as<string>();
-
-    if ( vm.count("input-address") )
-        _options->inputAddress = vm["input-address"].as<string>();
-
-    if ( vm.count("output-socket-type") )
-        _options->outputSocketType = vm["output-socket-type"].as<vector<string>>();
-
-    if ( vm.count("output-buff-size") )
-        _options->outputBufSize = vm["output-buff-size"].as<vector<int>>();
-
-    if ( vm.count("output-method") )
-        _options->outputMethod = vm["output-method"].as<vector<string>>();
-
-    if ( vm.count("output-address") )
-        _options->outputAddress = vm["output-address"].as<vector<string>>();
+    if (vm.count("id")) { _options->id = vm["id"].as<string>(); }
+    if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as<int>(); }
+    if (vm.count("transport"))  { _options->transport = vm["transport"].as<string>(); }
+    if (vm.count("num-outputs")) { _options->numOutputs = vm["num-outputs"].as<int>(); }
+    if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as<string>(); }
+    if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as<int>(); }
+    if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as<string>(); }
+    if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as<string>(); }
+    if (vm.count("output-socket-type")) { _options->outputSocketType = vm["output-socket-type"].as<vector<string>>(); }
+    if (vm.count("output-buff-size")) { _options->outputBufSize = vm["output-buff-size"].as<vector<int>>(); }
+    if (vm.count("output-method")) { _options->outputMethod = vm["output-method"].as<vector<string>>(); }
+    if (vm.count("output-address")) { _options->outputAddress = vm["output-address"].as<vector<string>>(); }
 
     return true;
 }
@@ -135,13 +111,7 @@ int main(int argc, char** argv)
 
     LOG(INFO) << "PID: " << getpid();
 
-#ifdef NANOMSG
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
-#else
-    FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
-#endif
-
-    splitter.SetTransport(transportFactory);
+    splitter.SetTransport(options.transport);
 
     FairMQChannel inputChannel(options.inputSocketType, options.inputMethod, options.inputAddress);
     inputChannel.UpdateSndBufSize(options.inputBufSize);