move zeromq and nanomsg implementation files to separate directories

fairmq/nanomsg/FairMQMessageNN.cxx

@@ -0,0 +1,135 @@
+/**
+ * FairMQMessageNN.cxx
+ *
+ * @since 2013-12-05
+ * @author: A. Rybalchenko
+ */
+
+#include <cstring>
+
+#include <nanomsg/nn.h>
+
+#include "FairMQMessageNN.h"
+#include "FairMQLogger.h"
+
+
+FairMQMessageNN::FairMQMessageNN() :
+  fSize(0),
+  fMessage(NULL)
+{
+}
+
+FairMQMessageNN::FairMQMessageNN(size_t size)
+{
+  fMessage = nn_allocmsg(size, 0);
+  if(!fMessage){
+    stringstream logmsg;
+    logmsg << "failed allocating message, reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  }
+  fSize = size;
+}
+
+FairMQMessageNN::FairMQMessageNN(void* data, size_t size)
+{
+  fMessage = nn_allocmsg(size, 0);
+  if(!fMessage){
+    stringstream logmsg;
+    logmsg << "failed allocating message, reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  }
+  memcpy (fMessage, data, size);
+  fSize = size;
+}
+
+void FairMQMessageNN::Rebuild()
+{
+  Clear();
+  fSize = 0;
+  fMessage = NULL;
+}
+
+void FairMQMessageNN::Rebuild(size_t size)
+{
+  Clear();
+  fMessage = nn_allocmsg(size, 0);
+  if(!fMessage){
+    stringstream logmsg;
+    logmsg << "failed allocating message, reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  }
+  fSize = size;
+}
+
+void FairMQMessageNN::Rebuild(void* data, size_t size)
+{
+  Clear();
+  fMessage = nn_allocmsg(size, 0);
+  if(!fMessage){
+    stringstream logmsg;
+    logmsg << "failed allocating message, reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  }
+  fSize = size;
+}
+
+void* FairMQMessageNN::GetMessage()
+{
+  return fMessage;
+}
+
+void* FairMQMessageNN::GetData()
+{
+  return fMessage;
+}
+
+size_t FairMQMessageNN::GetSize()
+{
+  return fSize;
+}
+
+void FairMQMessageNN::SetMessage(void* data, size_t size)
+{
+  fMessage = data;
+  fSize = size;
+}
+
+void FairMQMessageNN::Copy(FairMQMessage* msg)
+{
+  if(fMessage){
+    int rc = nn_freemsg(fMessage);
+    if( rc < 0 ){
+      stringstream logmsg;
+      logmsg << "failed freeing message, reason: " << nn_strerror(errno);
+      FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+    }
+  }
+
+  size_t size = msg->GetSize();
+
+  fMessage = nn_allocmsg(size, 0);
+  if(!fMessage){
+    stringstream logmsg;
+    logmsg << "failed allocating message, reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  }
+  std::memcpy (fMessage, msg->GetMessage(), size);
+  fSize = size;
+}
+
+void FairMQMessageNN::Clear()
+{
+  int rc = nn_freemsg(fMessage);
+  if (rc < 0) {
+    stringstream logmsg;
+    logmsg << "failed freeing message, reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  } else {
+    fMessage = NULL;
+    fSize = 0;
+  }
+}
+
+FairMQMessageNN::~FairMQMessageNN()
+{
+}

fairmq/nanomsg/FairMQMessageNN.h

@@ -0,0 +1,43 @@
+/**
+ * FairMQMessageNN.h
+ *
+ * @since 2013-12-05
+ * @author: A. Rybalchenko
+ */
+
+#ifndef FAIRMQMESSAGENN_H_
+#define FAIRMQMESSAGENN_H_
+
+#include <cstddef>
+
+#include "FairMQMessage.h"
+
+
+class FairMQMessageNN : public FairMQMessage
+{
+  public:
+    FairMQMessageNN();
+    FairMQMessageNN(size_t size);
+    FairMQMessageNN(void* data, size_t size);
+
+    virtual void Rebuild();
+    virtual void Rebuild(size_t size);
+    virtual void Rebuild(void* data, size_t site);
+
+    virtual void* GetMessage();
+    virtual void* GetData();
+    virtual size_t GetSize();
+
+    virtual void Copy(FairMQMessage* msg);
+
+    void SetMessage(void* data, size_t size);
+    void Clear();
+
+    virtual ~FairMQMessageNN();
+
+  private:
+    void* fMessage;
+    size_t fSize;
+};
+
+#endif /* FAIRMQMESSAGENN_H_ */

fairmq/nanomsg/FairMQPollerNN.cxx

@@ -0,0 +1,39 @@
+/**
+ * FairMQPollerNN.cxx
+ *
+ * @since 2014-01-23
+ * @author A. Rybalchenko
+ */
+
+#include <nanomsg/nn.h>
+
+#include "FairMQPollerNN.h"
+
+FairMQPollerNN::FairMQPollerNN(const vector<FairMQSocket*>& inputs)
+{
+  fNumItems = inputs.size();
+  items = new nn_pollfd[fNumItems];
+
+  for (int i = 0; i < fNumItems; i++) {
+    items[i].fd = inputs.at(i)->GetSocket(1);
+    items[i].events = NN_POLLIN;
+  }
+}
+
+void FairMQPollerNN::Poll(int timeout)
+{
+  nn_poll(items, fNumItems, timeout);
+}
+
+bool FairMQPollerNN::CheckInput(int index)
+{
+  if (items[index].revents & NN_POLLIN) 
+    return true;
+
+  return false;
+}
+
+FairMQPollerNN::~FairMQPollerNN()
+{
+  if (items != NULL) delete [] items;
+}

fairmq/nanomsg/FairMQPollerNN.h

@@ -0,0 +1,33 @@
+/**
+ * FairMQPollerNN.h
+ *
+ * @since 2014-01-23
+ * @author A. Rybalchenko
+ */
+
+#ifndef FAIRMQPOLLERNN_H_
+#define FAIRMQPOLLERNN_H_
+
+#include <vector>
+
+#include "FairMQPoller.h"
+#include "FairMQSocket.h"
+
+using std::vector;
+
+class FairMQPollerNN : public FairMQPoller
+{
+  public:
+    FairMQPollerNN(const vector<FairMQSocket*>& inputs);
+
+    virtual void Poll(int timeout);
+    virtual bool CheckInput(int index);
+
+    virtual ~FairMQPollerNN();
+
+  private:
+    nn_pollfd* items;
+    int fNumItems;
+};
+
+#endif /* FAIRMQPOLLERNN_H_ */

fairmq/nanomsg/FairMQSocketNN.cxx

@@ -0,0 +1,161 @@
+/**
+ * FairMQSocketNN.cxx
+ *
+ * @since 2012-12-05
+ * @author A. Rybalchenko
+ */
+
+#include <sstream>
+
+#include "FairMQSocketNN.h"
+#include "FairMQLogger.h"
+
+FairMQSocketNN::FairMQSocketNN(const string& type, int num) :
+  fBytesTx(0),
+  fBytesRx(0),
+  fMessagesTx(0),
+  fMessagesRx(0)
+{
+  stringstream id;
+  id << type << "." << num;
+  fId = id.str();
+
+  fSocket = nn_socket (AF_SP, GetConstant(type));
+  if (type == "sub") {
+    nn_setsockopt(fSocket, NN_SUB, NN_SUB_SUBSCRIBE, NULL, 0);
+  }
+
+  stringstream logmsg;
+  logmsg << "created socket #" << fId;
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+}
+
+string FairMQSocketNN::GetId()
+{
+  return fId;
+}
+
+void FairMQSocketNN::Bind(const string& address)
+{
+  stringstream logmsg;
+  logmsg << "bind socket #" << fId << " on " << address;
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  int eid = nn_bind(fSocket, address.c_str());
+  if (eid < 0) {
+    stringstream logmsg2;
+    logmsg2 << "failed binding socket #" << fId << ", reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg2.str());
+  }
+}
+
+void FairMQSocketNN::Connect(const string& address)
+{
+  stringstream logmsg;
+  logmsg << "connect socket #" << fId << " to " << address;
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  int eid = nn_connect(fSocket, address.c_str());
+  if (eid < 0) {
+    stringstream logmsg2;
+    logmsg2 << "failed connecting socket #" << fId << ", reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg2.str());
+  }
+}
+
+size_t FairMQSocketNN::Send(FairMQMessage* msg)
+{
+  void* ptr = msg->GetMessage();
+  int rc = nn_send(fSocket, &ptr, NN_MSG, 0);
+  if (rc < 0) {
+    stringstream logmsg;
+    logmsg << "failed sending on socket #" << fId << ", reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  } else {
+    fBytesTx += rc;
+    ++fMessagesTx;
+  }
+
+  return rc;
+}
+
+size_t FairMQSocketNN::Receive(FairMQMessage* msg)
+{
+  void* ptr = msg->GetMessage();
+  int rc = nn_recv(fSocket, &ptr, NN_MSG, 0);
+  if (rc < 0) {
+    stringstream logmsg;
+    logmsg << "failed receiving on socket #" << fId << ", reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  } else {
+    fBytesRx += rc;
+    ++fMessagesRx;
+    msg->SetMessage(ptr, rc);
+  }
+
+  return rc;
+}
+
+void* FairMQSocketNN::GetSocket()
+{
+  return NULL;// dummy method to compy with the interface. functionality not possible in zeromq.
+}
+
+int FairMQSocketNN::GetSocket(int nothing)
+{
+  return fSocket;
+}
+
+void FairMQSocketNN::Close()
+{
+  nn_close(fSocket);
+}
+
+void FairMQSocketNN::SetOption(const string& option, const void* value, size_t valueSize)
+{
+  int rc = nn_setsockopt(fSocket, NN_SOL_SOCKET, GetConstant(option), value, valueSize);
+  if (rc < 0) {
+    stringstream logmsg;
+    logmsg << "failed setting socket option, reason: " << nn_strerror(errno);
+    FairMQLogger::GetInstance()->Log(FairMQLogger::ERROR, logmsg.str());
+  }
+}
+
+unsigned long FairMQSocketNN::GetBytesTx()
+{
+  return fBytesTx;
+}
+
+unsigned long FairMQSocketNN::GetBytesRx()
+{
+  return fBytesRx;
+}
+
+unsigned long FairMQSocketNN::GetMessagesTx()
+{
+  return fMessagesTx;
+}
+
+unsigned long FairMQSocketNN::GetMessagesRx()
+{
+  return fMessagesRx;
+}
+
+int FairMQSocketNN::GetConstant(const string& constant)
+{
+  if (constant == "sub") return NN_SUB;
+  if (constant == "pub") return NN_PUB;
+  if (constant == "xsub") return NN_SUB; // TODO: is there XPUB, XSUB for nanomsg?
+  if (constant == "xpub") return NN_PUB;
+  if (constant == "push") return NN_PUSH;
+  if (constant == "pull") return NN_PULL;
+  if (constant == "snd-hwm") return NN_SNDBUF;
+  if (constant == "rcv-hwm") return NN_RCVBUF;
+
+  return -1;
+}
+
+FairMQSocketNN::~FairMQSocketNN()
+{
+  Close();
+}

fairmq/nanomsg/FairMQSocketNN.h

@@ -0,0 +1,56 @@
+/**
+ * FairMQSocketNN.h
+ *
+ * @since 2013-12-05
+ * @author A. Rybalchenko
+ */
+
+#ifndef FAIRMQSOCKETNN_H_
+#define FAIRMQSOCKETNN_H_
+
+#include <nanomsg/nn.h>
+#include <nanomsg/pipeline.h>
+#include <nanomsg/pubsub.h>
+
+#include "FairMQSocket.h"
+
+
+class FairMQSocketNN : public FairMQSocket
+{
+  public:
+    FairMQSocketNN(const string& type, int num);
+
+    virtual string GetId();
+
+    virtual void Bind(const string& address);
+    virtual void Connect(const string& address);
+
+    virtual size_t Send(FairMQMessage* msg);
+    virtual size_t Receive(FairMQMessage* msg);
+
+    virtual void* GetSocket();
+    virtual int GetSocket(int nothing);
+    virtual void Close();
+
+    virtual void SetOption(const string& option, const void* value, size_t valueSize);
+
+    unsigned long GetBytesTx();
+    unsigned long GetBytesRx();
+    unsigned long GetMessagesTx();
+    unsigned long GetMessagesRx();
+
+    static int GetConstant(const string& constant);
+
+
+    virtual ~FairMQSocketNN();
+
+  private:
+    int fSocket;
+    string fId;
+    unsigned long fBytesTx;
+    unsigned long fBytesRx;
+    unsigned long fMessagesTx;
+    unsigned long fMessagesRx;
+};
+
+#endif /* FAIRMQSOCKETNN_H_ */

fairmq/nanomsg/FairMQTransportFactoryNN.cxx

@@ -0,0 +1,38 @@
+/**
+ * FairMQTransportFactoryNN.cxx
+ *
+ * @since 2014-01-20
+ * @author: A. Rybalchenko
+ */
+
+#include "FairMQTransportFactoryNN.h"
+
+FairMQTransportFactoryNN::FairMQTransportFactoryNN()
+{
+
+}
+
+FairMQMessage* FairMQTransportFactoryNN::CreateMessage()
+{
+  return new FairMQMessageNN();
+}
+
+FairMQMessage* FairMQTransportFactoryNN::CreateMessage(size_t size)
+{
+  return new FairMQMessageNN(size);
+}
+
+FairMQMessage* FairMQTransportFactoryNN::CreateMessage(void* data, size_t size)
+{
+  return new FairMQMessageNN(data, size);
+}
+
+FairMQSocket* FairMQTransportFactoryNN::CreateSocket(const string& type, int num)
+{
+  return new FairMQSocketNN(type, num);
+}
+
+FairMQPoller* FairMQTransportFactoryNN::CreatePoller(const vector<FairMQSocket*>& inputs)
+{
+  return new FairMQPollerNN(inputs);
+}

fairmq/nanomsg/FairMQTransportFactoryNN.h

@@ -0,0 +1,33 @@
+/**
+ * FairMQTransportFactoryNN.h
+ *
+ * @since 2014-01-20
+ * @author: A. Rybalchenko
+ */
+
+#ifndef FAIRMQTRANSPORTFACTORYNN_H_
+#define FAIRMQTRANSPORTFACTORYNN_H_
+
+#include <vector>
+
+#include "FairMQTransportFactory.h"
+#include "FairMQMessageNN.h"
+#include "FairMQSocketNN.h"
+#include "FairMQPollerNN.h"
+
+class FairMQTransportFactoryNN : public FairMQTransportFactory
+{
+  public:
+    FairMQTransportFactoryNN();
+
+    virtual FairMQMessage* CreateMessage();
+    virtual FairMQMessage* CreateMessage(size_t size);
+    virtual FairMQMessage* CreateMessage(void* data, size_t size);
+    virtual FairMQSocket* CreateSocket(const string& type, int num);
+    virtual FairMQPoller* CreatePoller(const vector<FairMQSocket*>& inputs);
+
+
+    virtual ~FairMQTransportFactoryNN() {};
+};
+
+#endif /* FAIRMQTRANSPORTFACTORYNN_H_ */

fairmq/nanomsg/runBenchmarkSampler.cxx

@@ -0,0 +1,115 @@
+/**
+ * runBenchmarkSampler.cxx
+ *
+ * @since 2013-04-23
+ * @author D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQBenchmarkSampler.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQBenchmarkSampler sampler;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  sampler.ChangeState(FairMQBenchmarkSampler::STOP);
+  sampler.ChangeState(FairMQBenchmarkSampler::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc != 9 ) {
+    cout << "Usage: bsampler ID eventSize eventRate numIoTreads\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
+              << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  sampler.SetTransport(transportFactory);
+
+  int i = 1;
+
+  sampler.SetProperty(FairMQBenchmarkSampler::Id, argv[i]);
+  ++i;
+
+  int eventSize;
+  stringstream(argv[i]) >> eventSize;
+  sampler.SetProperty(FairMQBenchmarkSampler::EventSize, eventSize);
+  ++i;
+
+  int eventRate;
+  stringstream(argv[i]) >> eventRate;
+  sampler.SetProperty(FairMQBenchmarkSampler::EventRate, eventRate);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  sampler.SetProperty(FairMQBenchmarkSampler::NumIoThreads, numIoThreads);
+  ++i;
+
+  sampler.SetProperty(FairMQBenchmarkSampler::NumInputs, 0);
+  sampler.SetProperty(FairMQBenchmarkSampler::NumOutputs, 1);
+
+
+  sampler.ChangeState(FairMQBenchmarkSampler::INIT);
+
+
+  sampler.SetProperty(FairMQBenchmarkSampler::OutputSocketType, argv[i], 0);
+  ++i;
+  int outputSndBufSize;
+  stringstream(argv[i]) >> outputSndBufSize;
+  sampler.SetProperty(FairMQBenchmarkSampler::OutputSndBufSize, outputSndBufSize, 0);
+  ++i;
+  sampler.SetProperty(FairMQBenchmarkSampler::OutputMethod, argv[i], 0);
+  ++i;
+  sampler.SetProperty(FairMQBenchmarkSampler::OutputAddress, argv[i], 0);
+  ++i;
+
+
+  sampler.ChangeState(FairMQBenchmarkSampler::SETOUTPUT);
+  sampler.ChangeState(FairMQBenchmarkSampler::SETINPUT);
+  sampler.ChangeState(FairMQBenchmarkSampler::RUN);
+
+
+
+  char ch;
+  cin.get(ch);
+
+  sampler.ChangeState(FairMQBenchmarkSampler::STOP);
+  sampler.ChangeState(FairMQBenchmarkSampler::END);
+
+  return 0;
+}
+

fairmq/nanomsg/runBuffer.cxx

@@ -0,0 +1,115 @@
+/**
+ * runBuffer.cxx
+ *
+ * @since 2012-10-26
+ * @author: D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQBuffer.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQBuffer buffer;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  buffer.ChangeState(FairMQBuffer::STOP);
+  buffer.ChangeState(FairMQBuffer::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc != 11 ) {
+    cout << "Usage: buffer \tID numIoTreads\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  buffer.SetTransport(transportFactory);
+
+  int i = 1;
+
+  buffer.SetProperty(FairMQBuffer::Id, argv[i]);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  buffer.SetProperty(FairMQBuffer::NumIoThreads, numIoThreads);
+  ++i;
+  buffer.SetProperty(FairMQBuffer::NumInputs, 1);
+  buffer.SetProperty(FairMQBuffer::NumOutputs, 1);
+
+
+  buffer.ChangeState(FairMQBuffer::INIT);
+
+
+  buffer.SetProperty(FairMQBuffer::InputSocketType, argv[i], 0);
+  ++i;
+  int inputRcvBufSize;
+  stringstream(argv[i]) >> inputRcvBufSize;
+  buffer.SetProperty(FairMQBuffer::InputRcvBufSize, inputRcvBufSize, 0);
+  ++i;
+  buffer.SetProperty(FairMQBuffer::InputMethod, argv[i], 0);
+  ++i;
+  buffer.SetProperty(FairMQBuffer::InputAddress, argv[i], 0);
+  ++i;
+
+  buffer.SetProperty(FairMQBuffer::OutputSocketType, argv[i], 0);
+  ++i;
+  int outputSndBufSize;
+  stringstream(argv[i]) >> outputSndBufSize;
+  buffer.SetProperty(FairMQBuffer::OutputSndBufSize, outputSndBufSize, 0);
+  ++i;
+  buffer.SetProperty(FairMQBuffer::OutputMethod, argv[i], 0);
+  ++i;
+  buffer.SetProperty(FairMQBuffer::OutputAddress, argv[i], 0);
+  ++i;
+
+
+  buffer.ChangeState(FairMQBuffer::SETOUTPUT);
+  buffer.ChangeState(FairMQBuffer::SETINPUT);
+  buffer.ChangeState(FairMQBuffer::RUN);
+
+
+
+  char ch;
+  cin.get(ch);
+
+  buffer.ChangeState(FairMQBuffer::STOP);
+  buffer.ChangeState(FairMQBuffer::END);
+
+  return 0;
+}
+

fairmq/nanomsg/runMerger.cxx

@@ -0,0 +1,126 @@
+/**
+ * runMerger.cxx
+ *
+ * @since 2012-12-06
+ * @author D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQMerger.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQMerger merger;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  merger.ChangeState(FairMQMerger::STOP);
+  merger.ChangeState(FairMQMerger::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc != 15 ) {
+    cout << "Usage: merger \tID numIoTreads\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  merger.SetTransport(transportFactory);
+
+  int i = 1;
+
+  merger.SetProperty(FairMQMerger::Id, argv[i]);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  merger.SetProperty(FairMQMerger::NumIoThreads, numIoThreads);
+  ++i;
+
+  merger.SetProperty(FairMQMerger::NumInputs, 2);
+  merger.SetProperty(FairMQMerger::NumOutputs, 1);
+
+
+  merger.ChangeState(FairMQMerger::INIT);
+
+
+  merger.SetProperty(FairMQMerger::InputSocketType, argv[i], 0);
+  ++i;
+  int inputRcvBufSize;
+  stringstream(argv[i]) >> inputRcvBufSize;
+  merger.SetProperty(FairMQMerger::InputRcvBufSize, inputRcvBufSize, 0);
+  ++i;
+  merger.SetProperty(FairMQMerger::InputMethod, argv[i], 0);
+  ++i;
+  merger.SetProperty(FairMQMerger::InputAddress, argv[i], 0);
+  ++i;
+
+  merger.SetProperty(FairMQMerger::InputSocketType, argv[i], 1);
+  ++i;
+  stringstream(argv[i]) >> inputRcvBufSize;
+  merger.SetProperty(FairMQMerger::InputRcvBufSize, inputRcvBufSize, 1);
+  ++i;
+  merger.SetProperty(FairMQMerger::InputMethod, argv[i], 1);
+  ++i;
+  merger.SetProperty(FairMQMerger::InputAddress, argv[i], 1);
+  ++i;
+
+  merger.SetProperty(FairMQMerger::OutputSocketType, argv[i], 0);
+  ++i;
+  int outputSndBufSize;
+  stringstream(argv[i]) >> outputSndBufSize;
+  merger.SetProperty(FairMQMerger::OutputSndBufSize, outputSndBufSize, 0);
+  ++i;
+  merger.SetProperty(FairMQMerger::OutputMethod, argv[i], 0);
+  ++i;
+  merger.SetProperty(FairMQMerger::OutputAddress, argv[i], 0);
+  ++i;
+
+
+  merger.ChangeState(FairMQMerger::SETOUTPUT);
+  merger.ChangeState(FairMQMerger::SETINPUT);
+  merger.ChangeState(FairMQMerger::RUN);
+
+
+  char ch;
+  cin.get(ch);
+
+  merger.ChangeState(FairMQMerger::STOP);
+  merger.ChangeState(FairMQMerger::END);
+
+  return 0;
+}
+

fairmq/nanomsg/runNToOneMerger.cxx

@@ -0,0 +1,124 @@
+/**
+ * runNToOneMerger.cxx
+ *
+ * @since 2012-12-06
+ * @author D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQMerger.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQMerger merger;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  merger.ChangeState(FairMQMerger::STOP);
+  merger.ChangeState(FairMQMerger::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc < 16 || (argc-8)%4!=0 ) {
+    cout << "Usage: merger \tID numIoTreads numInputs\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\t...\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
+              << argc << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  merger.SetTransport(transportFactory);
+
+  int i = 1;
+
+  merger.SetProperty(FairMQMerger::Id, argv[i]);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  merger.SetProperty(FairMQMerger::NumIoThreads, numIoThreads);
+  ++i;
+
+  int numInputs;
+  stringstream(argv[i]) >> numInputs;
+  merger.SetProperty(FairMQMerger::NumInputs, numInputs);
+  ++i;
+
+  merger.SetProperty(FairMQMerger::NumOutputs, 1);
+
+
+  merger.ChangeState(FairMQMerger::INIT);
+
+
+  for (int iInput = 0; iInput < numInputs; iInput++ ) {
+    merger.SetProperty(FairMQMerger::InputSocketType, argv[i], iInput);
+    ++i;
+    int inputRcvBufSize;
+    stringstream(argv[i]) >> inputRcvBufSize;
+    merger.SetProperty(FairMQMerger::InputRcvBufSize, inputRcvBufSize, iInput);
+    ++i;
+    merger.SetProperty(FairMQMerger::InputMethod, argv[i], iInput);
+    ++i;
+    merger.SetProperty(FairMQMerger::InputAddress, argv[i], iInput);
+    ++i;
+  }
+
+  merger.SetProperty(FairMQMerger::OutputSocketType, argv[i], 0);
+  ++i;
+  int outputSndBufSize;
+  stringstream(argv[i]) >> outputSndBufSize;
+  merger.SetProperty(FairMQMerger::OutputSndBufSize, outputSndBufSize, 0);
+  ++i;
+  merger.SetProperty(FairMQMerger::OutputMethod, argv[i], 0);
+  ++i;
+  merger.SetProperty(FairMQMerger::OutputAddress, argv[i], 0);
+  ++i;
+
+
+  merger.ChangeState(FairMQMerger::SETOUTPUT);
+  merger.ChangeState(FairMQMerger::SETINPUT);
+  merger.ChangeState(FairMQMerger::RUN);
+
+
+  char ch;
+  cin.get(ch);
+
+  merger.ChangeState(FairMQMerger::STOP);
+  merger.ChangeState(FairMQMerger::END);
+
+  return 0;
+}
+

fairmq/nanomsg/runOneToNSplitter.cxx

@@ -0,0 +1,122 @@
+/**
+ * runOneToNSplitter.cxx
+ *
+ * @since 2012-12-06
+ * @author D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQSplitter.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQSplitter splitter;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  splitter.ChangeState(FairMQSplitter::STOP);
+  splitter.ChangeState(FairMQSplitter::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc < 16 || (argc - 8) % 4 != 0 ) { // argc{ name, id, threads, nout, insock, inbuff, inmet, inadd, ... out}
+    cout << "Usage: splitter \tID numIoTreads numOutputs\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
+              << "\t\t..." << argc << " arguments provided" << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  splitter.SetTransport(transportFactory);
+
+  int i = 1;
+
+  splitter.SetProperty(FairMQSplitter::Id, argv[i]);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  splitter.SetProperty(FairMQSplitter::NumIoThreads, numIoThreads);
+  ++i;
+
+  splitter.SetProperty(FairMQSplitter::NumInputs, 1);
+
+  int numOutputs;
+  stringstream(argv[i]) >> numOutputs;
+  splitter.SetProperty(FairMQSplitter::NumOutputs, numOutputs);
+  ++i;
+
+
+  splitter.ChangeState(FairMQSplitter::INIT);
+
+
+  splitter.SetProperty(FairMQSplitter::InputSocketType, argv[i], 0);
+  ++i;
+  int inputRcvBufSize;
+  stringstream(argv[i]) >> inputRcvBufSize;
+  splitter.SetProperty(FairMQSplitter::InputRcvBufSize, inputRcvBufSize, 0);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::InputMethod, argv[i], 0);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::InputAddress, argv[i], 0);
+  ++i;
+
+  int outputSndBufSize;
+  for (int iOutput = 0; iOutput < numOutputs; iOutput++) {
+    splitter.SetProperty(FairMQSplitter::OutputSocketType, argv[i], iOutput);
+    ++i;
+    stringstream(argv[i]) >> outputSndBufSize;
+    splitter.SetProperty(FairMQSplitter::OutputSndBufSize, outputSndBufSize, iOutput);
+    ++i;
+    splitter.SetProperty(FairMQSplitter::OutputMethod, argv[i], iOutput);
+    ++i;
+    splitter.SetProperty(FairMQSplitter::OutputAddress, argv[i], iOutput);
+    ++i;
+  }
+
+  splitter.ChangeState(FairMQSplitter::SETOUTPUT);
+  splitter.ChangeState(FairMQSplitter::SETINPUT);
+  splitter.ChangeState(FairMQSplitter::RUN);
+
+
+  char ch;
+  cin.get(ch);
+
+  splitter.ChangeState(FairMQSplitter::STOP);
+  splitter.ChangeState(FairMQSplitter::END);
+
+  return 0;
+}
+

fairmq/nanomsg/runProxy.cxx

@@ -0,0 +1,115 @@
+/**
+ * runProxy.cxx
+ *
+ * @since 2013-10-07
+ * @author A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQProxy.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQProxy proxy;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  proxy.ChangeState(FairMQProxy::STOP);
+  proxy.ChangeState(FairMQProxy::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc != 11 ) {
+    cout << "Usage: proxy \tID numIoTreads\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  proxy.SetTransport(transportFactory);
+
+  int i = 1;
+
+  proxy.SetProperty(FairMQProxy::Id, argv[i]);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  proxy.SetProperty(FairMQProxy::NumIoThreads, numIoThreads);
+  ++i;
+
+  proxy.SetProperty(FairMQProxy::NumInputs, 1);
+  proxy.SetProperty(FairMQProxy::NumOutputs, 1);
+
+
+  proxy.ChangeState(FairMQProxy::INIT);
+
+
+  proxy.SetProperty(FairMQProxy::InputSocketType, argv[i], 0);
+  ++i;
+  int inputRcvBufSize;
+  stringstream(argv[i]) >> inputRcvBufSize;
+  proxy.SetProperty(FairMQProxy::InputRcvBufSize, inputRcvBufSize, 0);
+  ++i;
+  proxy.SetProperty(FairMQProxy::InputMethod, argv[i], 0);
+  ++i;
+  proxy.SetProperty(FairMQProxy::InputAddress, argv[i], 0);
+  ++i;
+
+  proxy.SetProperty(FairMQProxy::OutputSocketType, argv[i], 0);
+  ++i;
+  int outputSndBufSize;
+  stringstream(argv[i]) >> outputSndBufSize;
+  proxy.SetProperty(FairMQProxy::OutputSndBufSize, outputSndBufSize, 0);
+  ++i;
+  proxy.SetProperty(FairMQProxy::OutputMethod, argv[i], 0);
+  ++i;
+  proxy.SetProperty(FairMQProxy::OutputAddress, argv[i], 0);
+  ++i;
+
+
+  proxy.ChangeState(FairMQProxy::SETOUTPUT);
+  proxy.ChangeState(FairMQProxy::SETINPUT);
+  proxy.ChangeState(FairMQProxy::RUN);
+
+
+  char ch;
+  cin.get(ch);
+
+  proxy.ChangeState(FairMQProxy::STOP);
+  proxy.ChangeState(FairMQProxy::END);
+
+  return 0;
+}
+

fairmq/nanomsg/runSink.cxx

@@ -0,0 +1,104 @@
+/**
+ * runSink.cxx
+ *
+ * @since 2013-01-21
+ * @author D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQSink.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQSink sink;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  sink.ChangeState(FairMQSink::STOP);
+  sink.ChangeState(FairMQSink::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc != 7 ) {
+    cout << "Usage: sink \tID numIoTreads\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  sink.SetTransport(transportFactory);
+
+  int i = 1;
+
+  sink.SetProperty(FairMQSink::Id, argv[i]);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  sink.SetProperty(FairMQSink::NumIoThreads, numIoThreads);
+  ++i;
+
+  sink.SetProperty(FairMQSink::NumInputs, 1);
+  sink.SetProperty(FairMQSink::NumOutputs, 0);
+
+
+  sink.ChangeState(FairMQSink::INIT);
+
+
+  sink.SetProperty(FairMQSink::InputSocketType, argv[i], 0);
+  ++i;
+  int inputRcvBufSize;
+  stringstream(argv[i]) >> inputRcvBufSize;
+  sink.SetProperty(FairMQSink::InputRcvBufSize, inputRcvBufSize, 0);
+  ++i;
+  sink.SetProperty(FairMQSink::InputMethod, argv[i], 0);
+  ++i;
+  sink.SetProperty(FairMQSink::InputAddress, argv[i], 0);
+  ++i;
+
+
+  sink.ChangeState(FairMQSink::SETOUTPUT);
+  sink.ChangeState(FairMQSink::SETINPUT);
+  sink.ChangeState(FairMQSink::RUN);
+
+
+  char ch;
+  cin.get(ch);
+
+  sink.ChangeState(FairMQSink::STOP);
+  sink.ChangeState(FairMQSink::END);
+
+  return 0;
+}
+

fairmq/nanomsg/runSplitter.cxx

@@ -0,0 +1,126 @@
+/**
+ * runSplitter.cxx
+ *
+ * @since 2012-12-06
+ * @author D. Klein, A. Rybalchenko
+ */
+
+#include <iostream>
+#include <csignal>
+
+#include "FairMQLogger.h"
+#include "FairMQSplitter.h"
+#include "FairMQTransportFactoryNN.h"
+
+using std::cout;
+using std::cin;
+using std::endl;
+using std::stringstream;
+
+
+FairMQSplitter splitter;
+
+static void s_signal_handler (int signal)
+{
+  cout << endl << "Caught signal " << signal << endl;
+
+  splitter.ChangeState(FairMQSplitter::STOP);
+  splitter.ChangeState(FairMQSplitter::END);
+
+  cout << "Shutdown complete. Bye!" << endl;
+  exit(1);
+}
+
+static void s_catch_signals (void)
+{
+  struct sigaction action;
+  action.sa_handler = s_signal_handler;
+  action.sa_flags = 0;
+  sigemptyset(&action.sa_mask);
+  sigaction(SIGINT, &action, NULL);
+  sigaction(SIGTERM, &action, NULL);
+}
+
+int main(int argc, char** argv)
+{
+  if ( argc != 15 ) {
+    cout << "Usage: splitter \tID numIoTreads\n"
+              << "\t\tinputSocketType inputRcvBufSize inputMethod inputAddress\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n"
+              << "\t\toutputSocketType outputSndBufSize outputMethod outputAddress\n" << endl;
+    return 1;
+  }
+
+  s_catch_signals();
+
+  stringstream logmsg;
+  logmsg << "PID: " << getpid();
+  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, logmsg.str());
+
+  FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+  splitter.SetTransport(transportFactory);
+
+  int i = 1;
+
+  splitter.SetProperty(FairMQSplitter::Id, argv[i]);
+  ++i;
+
+  int numIoThreads;
+  stringstream(argv[i]) >> numIoThreads;
+  splitter.SetProperty(FairMQSplitter::NumIoThreads, numIoThreads);
+  ++i;
+
+  splitter.SetProperty(FairMQSplitter::NumInputs, 1);
+  splitter.SetProperty(FairMQSplitter::NumOutputs, 2);
+
+
+  splitter.ChangeState(FairMQSplitter::INIT);
+
+
+  splitter.SetProperty(FairMQSplitter::InputSocketType, argv[i], 0);
+  ++i;
+  int inputRcvBufSize;
+  stringstream(argv[i]) >> inputRcvBufSize;
+  splitter.SetProperty(FairMQSplitter::InputRcvBufSize, inputRcvBufSize, 0);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::InputMethod, argv[i], 0);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::InputAddress, argv[i], 0);
+  ++i;
+
+  splitter.SetProperty(FairMQSplitter::OutputSocketType, argv[i], 0);
+  ++i;
+  int outputSndBufSize;
+  stringstream(argv[i]) >> outputSndBufSize;
+  splitter.SetProperty(FairMQSplitter::OutputSndBufSize, outputSndBufSize, 0);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::OutputMethod, argv[i], 0);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::OutputAddress, argv[i], 0);
+  ++i;
+
+  splitter.SetProperty(FairMQSplitter::OutputSocketType, argv[i], 1);
+  ++i;
+  stringstream(argv[i]) >> outputSndBufSize;
+  splitter.SetProperty(FairMQSplitter::OutputSndBufSize, outputSndBufSize, 1);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::OutputMethod, argv[i], 1);
+  ++i;
+  splitter.SetProperty(FairMQSplitter::OutputAddress, argv[i], 1);
+  ++i;
+
+
+  splitter.ChangeState(FairMQSplitter::SETOUTPUT);
+  splitter.ChangeState(FairMQSplitter::SETINPUT);
+  splitter.ChangeState(FairMQSplitter::RUN);
+
+
+  char ch;
+  cin.get(ch);
+
+  splitter.ChangeState(FairMQSplitter::STOP);
+  splitter.ChangeState(FairMQSplitter::END);
+
+  return 0;
+}
+