Add new Send/Receive methods with smart pointers and no flag checks.

fairmq/FairMQChannel.cxx

@@ -36,6 +36,8 @@ FairMQChannel::FairMQChannel()
     , fPoller(nullptr)
     , fCmdSocket(nullptr)
     , fTransportFactory(nullptr)
+    , fNoBlockFlag(0)
+    , fSndMoreFlag(0)
 {
 }
 
@@ -52,6 +54,8 @@ FairMQChannel::FairMQChannel(const string& type, const string& method, const str
     , fPoller(nullptr)
     , fCmdSocket(nullptr)
     , fTransportFactory(nullptr)
+    , fNoBlockFlag(0)
+    , fSndMoreFlag(0)
 {
 }
 
@@ -347,10 +351,21 @@ bool FairMQChannel::InitCommandInterface(FairMQTransportFactory* factory)
     fTransportFactory = factory;
 
     fCmdSocket = fTransportFactory->CreateSocket("sub", "device-commands", 1);
-    fCmdSocket->Connect("inproc://commands");
+    if (fCmdSocket)
+    {
+        fCmdSocket->Connect("inproc://commands");
 
-    fPoller = fTransportFactory->CreatePoller(*fSocket, *fCmdSocket);
+        fNoBlockFlag = fCmdSocket->NOBLOCK;
-    return true;
+        fSndMoreFlag = fCmdSocket->SNDMORE;
+
+        fPoller = fTransportFactory->CreatePoller(*fSocket, *fCmdSocket);
+
+        return true;
+    }
+    else
+    {
+        return false;
+    }
 }
 
 void FairMQChannel::ResetChannel()
@@ -359,6 +374,57 @@ void FairMQChannel::ResetChannel()
     // TODO: implement channel resetting
 }
 
+int FairMQChannel::Send(const unique_ptr<FairMQMessage>& msg) const
+{
+    fPoller->Poll(-1);
+
+    if (fPoller->CheckInput(0))
+    {
+        HandleCommand();
+        return -1;
+    }
+
+    if (fPoller->CheckOutput(1))
+    {
+        return fSocket->Send(msg.get(), 0);
+    }
+
+    return -1;
+}
+
+int FairMQChannel::SendAsync(const unique_ptr<FairMQMessage>& msg) const
+{
+    return fSocket->Send(msg.get(), fNoBlockFlag);
+}
+
+int FairMQChannel::SendPart(const unique_ptr<FairMQMessage>& msg) const
+{
+    return fSocket->Send(msg.get(), fSndMoreFlag);
+}
+
+int FairMQChannel::Receive(const unique_ptr<FairMQMessage>& msg) const
+{
+    fPoller->Poll(-1);
+
+    if (fPoller->CheckInput(0))
+    {
+        HandleCommand();
+        return -1;
+    }
+
+    if (fPoller->CheckInput(1))
+    {
+        return fSocket->Receive(msg.get(), 0);
+    }
+
+    return -1;
+}
+
+int FairMQChannel::ReceiveAsync(const unique_ptr<FairMQMessage>& msg) const
+{
+    return fSocket->Receive(msg.get(), fNoBlockFlag);
+}
+
 int FairMQChannel::Send(FairMQMessage* msg, const string& flag) const
 {
     if (flag == "")

fairmq/FairMQChannel.h

@@ -16,6 +16,7 @@
 #define FAIRMQCHANNEL_H_
 
 #include <string>
+#include <memory> // unique_ptr
 
 #include <boost/thread/mutex.hpp>
 
@@ -59,6 +60,14 @@ class FairMQChannel
     FairMQSocket* fSocket;
 
     // Wrappers for the socket methods to simplify the usage of channels
+    int Send(const std::unique_ptr<FairMQMessage>& msg) const;
+    int SendAsync(const std::unique_ptr<FairMQMessage>& msg) const;
+    int SendPart(const std::unique_ptr<FairMQMessage>& msg) const;
+
+    int Receive(const std::unique_ptr<FairMQMessage>& msg) const;
+    int ReceiveAsync(const std::unique_ptr<FairMQMessage>& msg) const;
+
+    // DEPRECATED socket method wrappers with raw pointers and flag checks
     int Send(FairMQMessage* msg, const std::string& flag = "") const;
     int Send(FairMQMessage* msg, const int flags) const;
     int Receive(FairMQMessage* msg, const std::string& flag = "") const;
@@ -84,6 +93,9 @@ class FairMQChannel
 
     FairMQTransportFactory* fTransportFactory;
 
+    int fNoBlockFlag;
+    int fSndMoreFlag;
+
     bool HandleCommand() const;
 
     // use static mutex to make the class easily copyable

fairmq/FairMQConfigurable.h

@@ -25,11 +25,12 @@ class FairMQConfigurable
         Last = 1
     };
     FairMQConfigurable();
+    virtual ~FairMQConfigurable();
+
     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);
-    virtual ~FairMQConfigurable();
 };
 
 #endif /* FAIRMQCONFIGURABLE_H_ */

fairmq/FairMQDevice.cxx

@@ -418,6 +418,11 @@ int FairMQDevice::GetProperty(const int key, const int default_ /*= 0*/)
     }
 }
 
+void FairMQDevice::SetTransport(unique_ptr<FairMQTransportFactory>& factory)
+{
+    fTransportFactory = factory.get();
+}
+
 void FairMQDevice::SetTransport(FairMQTransportFactory* factory)
 {
     fTransportFactory = factory;
@@ -637,8 +642,6 @@ void FairMQDevice::ResetWrapper()
 
 void FairMQDevice::Reset()
 {
-    LOG(DEBUG) << "Resetting Device...";
-
     // iterate over the channels map
     for (auto mi = fChannels.begin(); mi != fChannels.end(); ++mi)
     {
@@ -657,8 +660,6 @@ void FairMQDevice::Reset()
             vi->fCmdSocket = nullptr;
         }
     }
-
-    LOG(DEBUG) << "Device reset finished!";
 }
 
 void FairMQDevice::Terminate()
@@ -725,4 +726,6 @@ FairMQDevice::~FairMQDevice()
         delete fCmdSocket;
         fCmdSocket = nullptr;
     }
+
+    delete fTransportFactory;
 }

fairmq/FairMQDevice.h

@@ -16,6 +16,7 @@
 #define FAIRMQDEVICE_H_
 
 #include <vector>
+#include <memory> // unique_ptr
 #include <string>
 #include <iostream>
 #include <unordered_map>
@@ -60,6 +61,7 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
     virtual int GetProperty(const int key, const int default_ = 0);
 
     virtual void SetTransport(FairMQTransportFactory* factory);
+    virtual void SetTransport(std::unique_ptr<FairMQTransportFactory>& factory);
 
     static bool SortSocketsByAddress(const FairMQChannel &lhs, const FairMQChannel &rhs);
 

fairmq/FairMQMessage.h

@@ -16,6 +16,7 @@
 #define FAIRMQMESSAGE_H_
 
 #include <cstddef> // for size_t
+#include <memory> // unique_ptr
 
 typedef void (fairmq_free_fn) (void *data, void *hint);
 
@@ -33,6 +34,7 @@ class FairMQMessage
 
     virtual void CloseMessage() = 0;
     virtual void Copy(FairMQMessage* msg) = 0;
+    virtual void Copy(const std::unique_ptr<FairMQMessage>& msg) = 0;
 
     virtual ~FairMQMessage() {};
 };

fairmq/devices/FairMQBenchmarkSampler.cxx

@@ -38,14 +38,15 @@ void FairMQBenchmarkSampler::Run()
     boost::thread resetEventCounter(boost::bind(&FairMQBenchmarkSampler::ResetEventCounter, this));
 
     void* buffer = operator new[](fEventSize);
-    FairMQMessage* baseMsg = fTransportFactory->CreateMessage(buffer, fEventSize);
+
+    unique_ptr<FairMQMessage> baseMsg(fTransportFactory->CreateMessage(buffer, fEventSize));
 
     // store the channel reference to avoid traversing the map on every loop iteration
     const FairMQChannel& dataChannel = fChannels.at("data-out").at(0);
 
     while (CheckCurrentState(RUNNING))
     {
-        FairMQMessage* msg = fTransportFactory->CreateMessage();
+        unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
         msg->Copy(baseMsg);
 
         dataChannel.Send(msg);
@@ -56,12 +57,8 @@ void FairMQBenchmarkSampler::Run()
         {
             boost::this_thread::sleep(boost::posix_time::milliseconds(1));
         }
-
-        delete msg;
     }
 
-    delete baseMsg;
-
     try {
         resetEventCounter.interrupt();
         resetEventCounter.join();

fairmq/devices/FairMQBenchmarkSampler.h

@@ -28,8 +28,7 @@ class FairMQBenchmarkSampler : public FairMQDevice
   public:
     enum
     {
-        InputFile = FairMQDevice::Last,
+        EventSize = FairMQDevice::Last,
-        EventSize,
         EventRate,
         Last
     };

fairmq/devices/FairMQBuffer.cxx

@@ -32,14 +32,12 @@ void FairMQBuffer::Run()
 
     while (CheckCurrentState(RUNNING))
     {
-        FairMQMessage* msg = fTransportFactory->CreateMessage();
+        std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
 
         if (dataInChannel.Receive(msg) > 0)
         {
             dataOutChannel.Send(msg);
         }
-
-        delete msg;
     }
 }
 

fairmq/devices/FairMQMerger.cxx

@@ -29,28 +29,32 @@ FairMQMerger::~FairMQMerger()
 
 void FairMQMerger::Run()
 {
-    FairMQPoller* poller = fTransportFactory->CreatePoller(fChannels.at("data-in"));
+    std::unique_ptr<FairMQPoller> poller(fTransportFactory->CreatePoller(fChannels.at("data-in")));
 
     // store the channel references to avoid traversing the map on every loop iteration
     const FairMQChannel& dataOutChannel = fChannels.at("data-out").at(0);
-    FairMQChannel* dataInChannels[fChannels.at("data-in").size()];
+    std::vector<FairMQChannel*> dataInChannels(fChannels.at("data-in").size());
     for (int i = 0; i < fChannels.at("data-in").size(); ++i)
     {
-        dataInChannels[i] = &(fChannels.at("data-in").at(i));
+        dataInChannels.at(i) = &(fChannels.at("data-in").at(i));
     }
 
     while (CheckCurrentState(RUNNING))
     {
-        FairMQMessage* msg = fTransportFactory->CreateMessage();
+        std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
 
         poller->Poll(100);
 
+        // Loop over the data input channels.
         for (int i = 0; i < fChannels.at("data-in").size(); ++i)
         {
+            // Check if the channel has data ready to be received.
             if (poller->CheckInput(i))
             {
+                // Try receiving the data.
                 if (dataInChannels[i]->Receive(msg) > 0)
                 {
+                    // If data was received, send it to output.
                     if (dataOutChannel.Send(msg) < 0)
                     {
                         LOG(DEBUG) << "Blocking send interrupted by a command";
@@ -64,9 +68,5 @@ void FairMQMerger::Run()
                 }
             }
         }
-
-        delete msg;
     }
-
-    delete poller;
 }

fairmq/devices/FairMQProxy.cxx

@@ -28,7 +28,7 @@ FairMQProxy::~FairMQProxy()
 
 void FairMQProxy::Run()
 {
-    FairMQMessage* msg = fTransportFactory->CreateMessage();
+    std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
 
     // store the channel references to avoid traversing the map on every loop iteration
     const FairMQChannel& dataInChannel = fChannels.at("data-in").at(0);
@@ -41,6 +41,4 @@ void FairMQProxy::Run()
             dataOutChannel.Send(msg);
         }
     }
-
-    delete msg;
 }

fairmq/devices/FairMQSink.cxx

@@ -29,11 +29,9 @@ void FairMQSink::Run()
 
     while (CheckCurrentState(RUNNING))
     {
-        FairMQMessage* msg = fTransportFactory->CreateMessage();
+        std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
 
         dataChannel.Receive(msg);
-
-        delete msg;
     }
 }
 

fairmq/devices/FairMQSplitter.cxx

@@ -41,7 +41,7 @@ void FairMQSplitter::Run()
 
     while (CheckCurrentState(RUNNING))
     {
-        FairMQMessage* msg = fTransportFactory->CreateMessage();
+        std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
 
         if (dataInChannel.Receive(msg) > 0)
         {
@@ -52,7 +52,5 @@ void FairMQSplitter::Run()
                 direction = 0;
             }
         }
-
-        delete msg;
     }
 }

fairmq/devices/GenericFileSink.h

@@ -71,17 +71,18 @@ class GenericFileSink : public FairMQDevice, public InputPolicy, public OutputPo
     {
         int receivedMsg = 0;
 
-        while (GetCurrentState() == RUNNING)
+        // store the channel reference to avoid traversing the map on every loop iteration
-        {
+        const FairMQChannel& inputChannel = fChannels["data-in"].at(0);
-            FairMQMessage* msg = fTransportFactory->CreateMessage();
 
-            if (fChannels["data-in"].at(0).Receive(msg) > 0)
+        while (CheckCurrentState(RUNNING))
+        {
+            std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
+
+            if (inputChannel.Receive(msg) > 0)
             {
-                OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg));
+                OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg.get()));
                 receivedMsg++;
             }
-
-            delete msg;
         }
 
         MQLOG(INFO) << "Received " << receivedMsg << " messages!";

fairmq/devices/GenericFileSink.tpl

@@ -44,15 +44,18 @@ void GenericFileSink<InputPolicy, OutputPolicy>::Run()
 {
     int receivedMsg = 0;
 
+    // store the channel reference to avoid traversing the map on every loop iteration
+    const FairMQChannel& inputChannel = fChannels["data-in"].at(0);
+
     while (CheckCurrentState(RUNNING))
     {
-        FairMQMessage* msg = fTransportFactory->CreateMessage();
+        std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
-        if (fChannels.at("data-in").at(0).Receive(msg) > 0)
+
+        if (inputChannel.Receive(msg) > 0)
         {
-            OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg));
+            OutputPolicy::AddToFile(InputPolicy::DeSerializeMsg(msg.get()));
             receivedMsg++;
         }
-        delete msg;
     }
 
     MQLOG(INFO) << "Received " << receivedMsg << " messages!";

fairmq/devices/GenericMerger.h

@@ -38,22 +38,25 @@ class GenericMerger : public FairMQDevice, public MergerPolicy, public InputPoli
 
     virtual void Run()
     {
-        FairMQPoller* poller = fTransportFactory->CreatePoller(fChannels["data-in"]);
+        std::unique_ptr<FairMQPoller> poller(fTransportFactory->CreatePoller(fChannels["data-in"]));
 
         int received = 0;
 
         while (GetCurrentState() == RUNNING)
         {
-            FairMQMessage* msg = fTransportFactory->CreateMessage();
+            std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
             // MergerPolicy::
             poller->Poll(fBlockingTime);
 
-            for (int i = 0; i < fChannels["datain"].size(); i++)
+            for (int i = 0; i < fChannels.at("data-in").size(); i++)
             {
                 if (poller->CheckInput(i))
                 {
-                    received = fChannels["data-in"].at(i).Receive(msg)
+                    received = fChannels.at("data-in").at(i).Receive(msg)
-                    MergerPolicy::Merge(InputPolicy::DeSerializeMsg(msg));
+                    if (received > 0)
+                    {
+                        MergerPolicy::Merge(InputPolicy::DeSerializeMsg(msg));
+                    }
                 }
 
                 OutputPolicy::SetMessage(msg);
@@ -64,11 +67,7 @@ class GenericMerger : public FairMQDevice, public MergerPolicy, public InputPoli
                     received = 0;
                 }
             }
-
-            delete msg;
         }
-
-        delete poller;
     }
 };
 

fairmq/devices/GenericProcessor.h

@@ -120,31 +120,29 @@ class GenericProcessor : public FairMQDevice, public InputPolicy, public OutputP
         int receivedMsgs = 0;
         int sentMsgs = 0;
 
+        const FairMQChannel& inputChannel = fChannels["data-in"].at(0);
+        const FairMQChannel& outputChannel = fChannels["data-out"].at(0);
+
         while (CheckCurrentState(RUNNING))
         {
-            FairMQMessage* msg = fTransportFactory->CreateMessage();
+            std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
 
             ++receivedMsgs;
 
-            if (fChannels["data-in"].at(0).Receive(msg) > 0)
+            if (inputChannel.Receive(msg) > 0)
             {
                 // InputPolicy::DeSerializeMsg(msg) --> deserialize data of msg and fill output container
                 // TaskPolicy::ExecuteTask( ... )   --> process output container
-                TaskPolicy::ExecuteTask(InputPolicy::DeSerializeMsg(msg));
+                TaskPolicy::ExecuteTask(InputPolicy::DeSerializeMsg(msg.get()));
 
                 // OutputPolicy::fMessage point to msg
-                OutputPolicy::SetMessage(msg);
+                OutputPolicy::SetMessage(msg.get());
 
                 // TaskPolicy::GetOutputData() --> Get processed output container
                 // OutputPolicy::message(...)  --> Serialize output container and fill fMessage
-                fChannels["data-out"].at(0).Send(OutputPolicy::SerializeMsg(TaskPolicy::GetOutputData()));
+                outputChannel.Send(OutputPolicy::SerializeMsg(TaskPolicy::GetOutputData()));
                 sentMsgs++;
             }
-
-            if (msg)
-            {
-                msg->CloseMessage();
-            }
         }
 
         MQLOG(INFO) << "Received " << receivedMsgs << " and sent " << sentMsgs << " messages!";

fairmq/devices/GenericSampler.h

@@ -13,7 +13,7 @@
  */
 
 #ifndef GENERICSAMPLER_H
-#define	GENERICSAMPLER_H
+#define GENERICSAMPLER_H
 
 #include <vector>
 #include <iostream>
@@ -82,7 +82,6 @@ class base_GenericSampler : public FairMQDevice, public T, public U
         typedef source_type source_type;
         typedef serialization_type serialization_type;
     };
-    
     */
 
     virtual void SetTransport(FairMQTransportFactory* factory);
@@ -104,8 +103,6 @@ class base_GenericSampler : public FairMQDevice, public T, public U
     int GetCurrentIndex() const;
     void SetContinuous(bool flag);
 
-    
-    /// ///////////////////////////////////////////////////////////////////////////////////////
     /*
        register the tasks you want to process and, which will be
        called by ExecuteTasks() member function. The registration is done by filling
@@ -130,25 +127,23 @@ class base_GenericSampler : public FairMQDevice, public T, public U
      
      To communicate with the Host derived class via callback, three methods from the host class are callable (only
      after binding these methods in the GenericSampler<I,O>::InitTask() )
-     
     */
     template<typename RegistrationManager>
     void RegisterTask(RegistrationManager manage)
     {
-        manage(this,fTaskList);
+        manage(this, fTaskList);
-        LOG(DEBUG)<<"Current Number of registered tasks = "<<fTaskList.size();
+        LOG(DEBUG) << "Current Number of registered tasks = " << fTaskList.size();
     }
-    /// ///////////////////////////////////////////////////////////////////////////////////////
+
     void ExecuteTasks()
     {
         for(const auto& p : fTaskList)
         {
-            LOG(DEBUG)<<"Execute Task "<< p.first;
+            LOG(DEBUG) << "Execute Task " << p.first;
             p.second();
         }
     }
 
-    
   protected:
     virtual void InitTask();
     virtual void Run();
@@ -160,30 +155,29 @@ class base_GenericSampler : public FairMQDevice, public T, public U
     int fEventRate;
     int fEventCounter;
     bool fContinuous;
-    std::map<key_type, task_type > fTaskList; // to handle Task list
+    std::map<key_type, task_type> fTaskList; // to handle Task list
-    
 
     // automatically enable or disable the call of policy function members for binding of host functions.
     // this template functions use SFINAE to detect the existence of the policy function signature.
-    template<typename S = source_type ,FairMQ::tools::enable_if_hasNot_BindSendPart<S> = 0 >
+    template<typename S = source_type,FairMQ::tools::enable_if_hasNot_BindSendPart<S> = 0>
-    void BindingSendPart(){}
+    void BindingSendPart() {}
-    template<typename S = source_type ,FairMQ::tools::enable_if_has_BindSendPart<S> = 0 >
+    template<typename S = source_type,FairMQ::tools::enable_if_has_BindSendPart<S> = 0>
     void BindingSendPart()
     {
         source_type::BindSendPart(std::bind(&base_GenericSampler::SendPart,this,std::placeholders::_1) );
     }
 
-    template<typename S = source_type ,FairMQ::tools::enable_if_hasNot_BindGetSocketNumber<S> = 0 >
+    template<typename S = source_type,FairMQ::tools::enable_if_hasNot_BindGetSocketNumber<S> = 0>
-    void BindingGetSocketNumber(){}
+    void BindingGetSocketNumber() {}
-    template<typename S = source_type ,FairMQ::tools::enable_if_has_BindGetSocketNumber<S> = 0 >
+    template<typename S = source_type,FairMQ::tools::enable_if_has_BindGetSocketNumber<S> = 0>
     void BindingGetSocketNumber()
     {
         source_type::BindGetSocketNumber(std::bind(&base_GenericSampler::GetSocketNumber,this) );
     }
 
-    template<typename S = source_type ,FairMQ::tools::enable_if_hasNot_BindGetCurrentIndex<S> = 0 >
+    template<typename S = source_type,FairMQ::tools::enable_if_hasNot_BindGetCurrentIndex<S> = 0>
-    void BindingGetCurrentIndex(){}
+    void BindingGetCurrentIndex() {}
-    template<typename S = source_type ,FairMQ::tools::enable_if_has_BindGetCurrentIndex<S> = 0 >
+    template<typename S = source_type,FairMQ::tools::enable_if_has_BindGetCurrentIndex<S> = 0>
     void BindingGetCurrentIndex()
     {
         source_type::BindGetCurrentIndex(std::bind(&base_GenericSampler::GetCurrentIndex,this) );
@@ -193,4 +187,3 @@ class base_GenericSampler : public FairMQDevice, public T, public U
 #include "GenericSampler.tpl"
 
 #endif /* GENERICSAMPLER_H */
-

fairmq/devices/GenericSampler.tpl

@@ -38,9 +38,6 @@ void base_GenericSampler<T,U,K,L>::InitTask()
     fNumEvents = source_type::GetNumberOfEvent();
 }
 
-
-
-
 template <typename T, typename U, typename K, typename L>
 void base_GenericSampler<T,U,K,L>::Run()
 {
@@ -56,19 +53,19 @@ void base_GenericSampler<T,U,K,L>::Run()
     {
         for (fCurrentIdx = 0; fCurrentIdx < fNumEvents; fCurrentIdx++)
         {
-            for(auto& p : fChannels[fOutChanName])
+            for (auto& p : fChannels[fOutChanName])
             {
-                FairMQMessage* msg = fTransportFactory->CreateMessage();
+                std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
-                serialization_type::SetMessage(msg);
+                serialization_type::SetMessage(msg.get());
                 source_type::SetIndex(fCurrentIdx);
                 ExecuteTasks();
                 p.Send(serialization_type::SerializeMsg(source_type::GetOutData()));
-                if (msg)
-                    msg->CloseMessage();
                 sentMsgs++;
 
-                if(fChannels[fOutChanName].size()>1)
+                if (fChannels[fOutChanName].size() > 1)
+                {
                     fCurrentIdx++;
+                }
 
                 // Optional event rate limiting
                 // --fEventCounter;
@@ -77,11 +74,15 @@ void base_GenericSampler<T,U,K,L>::Run()
                 // }
 
                 if (!CheckCurrentState(RUNNING))
+                {
                     break;
+                }
             }
             // if more than one socket, remove the last incrementation
-            if(fChannels[fOutChanName].size()>1)
+            if (fChannels[fOutChanName].size() > 1)
+            {
                     fCurrentIdx--;
+            }
         }
     }
     while (CheckCurrentState(RUNNING) && fContinuous);
@@ -96,14 +97,12 @@ template <typename T, typename U, typename K, typename L>
 void base_GenericSampler<T,U,K,L>::SendPart(int socketIdx)
 {
     fCurrentIdx++;
-    if(fCurrentIdx<fNumEvents)
+    if (fCurrentIdx < fNumEvents)
     {
-        FairMQMessage* msg = fTransportFactory->CreateMessage();
+        std::unique_ptr<FairMQMessage> msg(fTransportFactory->CreateMessage());
-        serialization_type::SetMessage(msg);
+        serialization_type::SetMessage(msg.get());
         source_type::SetIndex(fCurrentIdx);
         fChannels[fOutChanName].at(socketIdx).Send(serialization_type::SerializeMsg(source_type::GetOutData()), "snd-more");
-        if (msg)
-            msg->CloseMessage();
     }
 }
 
@@ -129,7 +128,7 @@ void base_GenericSampler<T,U,K,L>::SetContinuous(bool flag)
 template <typename T, typename U, typename K, typename L>
 void base_GenericSampler<T,U,K,L>::ResetEventCounter()
 {
-    while (GetCurrentState() == RUNNING)
+    while (CheckCurrentState(RUNNING))
     {
         try
         {
@@ -197,8 +196,6 @@ std::string base_GenericSampler<T,U,K,L>::GetProperty(const int key, const std::
     }
 }
 
-
-
 template<typename T, typename U>
 using GenericSampler = base_GenericSampler<T,U,int,std::function<void()> >;
 typedef std::map<int, std::function<void()> > SamplerTasksMap;

fairmq/run/runBenchmarkSampler.cxx

@@ -67,12 +67,11 @@ int main(int argc, char** argv)
         LOG(INFO) << "PID: " << getpid();
 
 #ifdef NANOMSG
-        FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN();
+        sampler.SetTransport(new FairMQTransportFactoryNN());
 #else
-        FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ();
+        sampler.SetTransport(new FairMQTransportFactoryZMQ());
 #endif
 
-        sampler.SetTransport(transportFactory);
 
         sampler.SetProperty(FairMQBenchmarkSampler::Id, id);
         sampler.SetProperty(FairMQBenchmarkSampler::EventSize, eventSize);

fairmq/zeromq/FairMQMessageZMQ.cxx

@@ -95,6 +95,24 @@ void FairMQMessageZMQ::SetMessage(void* data, size_t size)
 }
 
 void FairMQMessageZMQ::Copy(FairMQMessage* msg)
+{
+    // DEPRECATED: Use Copy(const unique_ptr<FairMQMessage>&)
+
+    // Shares the message buffer between msg and this fMessage.
+    if (zmq_msg_copy(&fMessage, (zmq_msg_t*)msg->GetMessage()) != 0)
+    {
+        LOG(ERROR) << "failed copying message, reason: " << zmq_strerror(errno);
+    }
+
+    // Alternatively, following code does a hard copy of the message, which allows to modify the original after making a copy, without affecting the new msg.
+
+    // CloseMessage();
+    // size_t size = msg->GetSize();
+    // zmq_msg_init_size(&fMessage, size);
+    // memcpy(zmq_msg_data(&fMessage), msg->GetData(), size);
+}
+
+void FairMQMessageZMQ::Copy(const unique_ptr<FairMQMessage>& msg)
 {
     // Shares the message buffer between msg and this fMessage.
     if (zmq_msg_copy(&fMessage, (zmq_msg_t*)msg->GetMessage()) != 0)

fairmq/zeromq/FairMQMessageZMQ.h

@@ -40,6 +40,7 @@ class FairMQMessageZMQ : public FairMQMessage
 
     virtual void CloseMessage();
     virtual void Copy(FairMQMessage* msg);
+    virtual void Copy(const std::unique_ptr<FairMQMessage>& msg);
 
     static void CleanUp(void* data, void* hint);