add callback on state change

fairmq/FairMQDevice.cxx

@@ -228,6 +228,11 @@ void FairMQDevice::InitWrapper()
         }
     }
 
+    if (!fStateChangeCallback.empty())
+    {
+        fStateChangeCallback(INITIALIZING_DEVICE);
+    }
+
     // Bind channels. Here one run is enough, because bind settings should be available locally
     // If necessary this could be handled in the same way as the connecting channels
     AttachChannels(uninitializedBindingChannels);
@@ -414,6 +419,11 @@ bool FairMQDevice::BindEndpoint(FairMQSocket& socket, string& endpoint)
 
 void FairMQDevice::InitTaskWrapper()
 {
+    if (!fStateChangeCallback.empty())
+    {
+        fStateChangeCallback(INITIALIZING_TASK);
+    }
+
     InitTask();
 
     ChangeState(internal_READY);
@@ -472,30 +482,13 @@ void FairMQDevice::PrintChannel(const string& name)
     }
 }
 
-void FairMQDevice::OnData(const string& channelName, InputMsgCallback callback)
-{
-    fDataCallbacks = true;
-    fMsgInputs.insert(make_pair(channelName, callback));
-
-    if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
-    {
-        fInputChannelKeys.push_back(channelName);
-    }
-}
-
-void FairMQDevice::OnData(const string& channelName, InputMultipartCallback callback)
-{
-    fDataCallbacks = true;
-    fMultipartInputs.insert(make_pair(channelName, callback));
-
-    if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
-    {
-        fInputChannelKeys.push_back(channelName);
-    }
-}
-
 void FairMQDevice::RunWrapper()
 {
+    if (!fStateChangeCallback.empty())
+    {
+        fStateChangeCallback(RUNNING);
+    }
+
     LOG(INFO) << "DEVICE: Running...";
 
     // start the rate logger thread
@@ -765,6 +758,16 @@ void FairMQDevice::PostRun()
 {
 }
 
+void FairMQDevice::PauseWrapper()
+{
+    if (!fStateChangeCallback.empty())
+    {
+        fStateChangeCallback(PAUSED);
+    }
+
+    Pause();
+}
+
 void FairMQDevice::Pause()
 {
     while (CheckCurrentState(PAUSED))
@@ -1195,6 +1198,11 @@ void FairMQDevice::Unblock()
 
 void FairMQDevice::ResetTaskWrapper()
 {
+    if (!fStateChangeCallback.empty())
+    {
+        fStateChangeCallback(RESETTING_TASK);
+    }
+
     ResetTask();
 
     ChangeState(internal_DEVICE_READY);
@@ -1206,6 +1214,11 @@ void FairMQDevice::ResetTask()
 
 void FairMQDevice::ResetWrapper()
 {
+    if (!fStateChangeCallback.empty())
+    {
+        fStateChangeCallback(RESETTING_DEVICE);
+    }
+
     Reset();
 
     ChangeState(internal_IDLE);

fairmq/FairMQDevice.h

@@ -356,7 +356,17 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
         }
     }
 
-    void OnData(const std::string& channelName, InputMsgCallback);
+    void OnData(const std::string& channelName, InputMsgCallback callback)
+    {
+        fDataCallbacks = true;
+        fMsgInputs.insert(make_pair(channelName, callback));
+
+        if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
+        {
+            fInputChannelKeys.push_back(channelName);
+        }
+    }
+
 
     template<class T>
     void OnData(const std::string& channelName, bool (T::* memberFunction)(FairMQParts& parts, int index))
@@ -373,7 +383,16 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
         }
     }
 
-    void OnData(const std::string& channelName, InputMultipartCallback);
+    void OnData(const std::string& channelName, InputMultipartCallback callback)
+    {
+        fDataCallbacks = true;
+        fMultipartInputs.insert(make_pair(channelName, callback));
+
+        if (find(fInputChannelKeys.begin(), fInputChannelKeys.end(), channelName) == fInputChannelKeys.end())
+        {
+            fInputChannelKeys.push_back(channelName);
+        }
+    }
 
     bool Terminated();
 
@@ -487,6 +506,8 @@ class FairMQDevice : public FairMQStateMachine, public FairMQConfigurable
     void InitTaskWrapper();
     /// Handles the Run() method
     void RunWrapper();
+    /// Handles the Pause() method
+    void PauseWrapper();
     /// Handles the ResetTask() method
     void ResetTaskWrapper();
     /// Handles the Reset() method

fairmq/FairMQStateMachine.cxx

@@ -225,3 +225,9 @@ bool FairMQStateMachine::WaitForEndOfStateForMs(std::string event, int durationI
 {
     return WaitForEndOfStateForMs(GetEventNumber(event), durationInMs);
 }
+
+void FairMQStateMachine::OnStateChange(std::function<void(const State)> callback)
+{
+    fStateChangeCallback.connect(callback);
+}
+

fairmq/FairMQStateMachine.h

@@ -35,6 +35,8 @@
 #include <boost/msm/front/state_machine_def.hpp>
 #include <boost/msm/front/functor_row.hpp>
 
+#include <boost/signals2.hpp> // signal/slot for onStateChange callbacks
+
 #include "FairMQLogger.h"
 
 namespace msm = boost::msm;
@@ -82,6 +84,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         , fWorkAvailable(false)
         , fState()
         , fChangeStateMutex()
+        , fStateChangeCallback()
         {}
 
     // Destructor
@@ -129,6 +132,10 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         {
             LOG(STATE) << "Entering IDLE state";
             fsm.fState = IDLE;
+            if (!fsm.fStateChangeCallback.empty())
+            {
+                fsm.fStateChangeCallback(IDLE);
+            }
         }
     };
 
@@ -158,6 +165,10 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         {
             LOG(STATE) << "Entering DEVICE READY state";
             fsm.fState = DEVICE_READY;
+            if (!fsm.fStateChangeCallback.empty())
+            {
+                fsm.fStateChangeCallback(DEVICE_READY);
+            }
         }
     };
 
@@ -181,6 +192,10 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         {
             LOG(STATE) << "Entering READY state";
             fsm.fState = READY;
+            if (!fsm.fStateChangeCallback.empty())
+            {
+                fsm.fStateChangeCallback(READY);
+            }
         }
     };
 
@@ -218,7 +233,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
                 fsm.fWorkDoneCondition.wait(lock);
             }
             fsm.fWorkAvailable = true;
-            fsm.fWork = std::bind(&FairMQFSM_::Pause, &fsm);
+            fsm.fWork = std::bind(&FairMQFSM_::PauseWrapper, &fsm);
             fsm.fWorkAvailableCondition.notify_one();
         }
     };
@@ -249,6 +264,10 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         {
             LOG(STATE) << "Entering READY state";
             fsm.fState = READY;
+            if (!fsm.fStateChangeCallback.empty())
+            {
+                fsm.fStateChangeCallback(READY);
+            }
 
             fsm.Unblock();
             std::unique_lock<std::mutex> lock(fsm.fWorkMutex);
@@ -266,6 +285,10 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         {
             LOG(STATE) << "RUNNING state finished without an external event, entering READY state";
             fsm.fState = READY;
+            if (!fsm.fStateChangeCallback.empty())
+            {
+                fsm.fStateChangeCallback(READY);
+            }
         }
     };
 
@@ -314,6 +337,10 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         {
             LOG(STATE) << "Entering EXITING state";
             fsm.fState = EXITING;
+            if (!fsm.fStateChangeCallback.empty())
+            {
+                fsm.fStateChangeCallback(EXITING);
+            }
 
             // terminate worker thread
             {
@@ -338,8 +365,11 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
         void operator()(EVT const&, FSM& fsm, SourceState&, TargetState&)
         {
             LOG(STATE) << "Entering ERROR state";
-
             fsm.fState = ERROR;
+            if (!fsm.fStateChangeCallback.empty())
+            {
+                fsm.fStateChangeCallback(ERROR);
+            }
         }
     };
 
@@ -350,6 +380,7 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
     virtual void InitTask() {}
     virtual void RunWrapper() {}
     virtual void Run() {}
+    virtual void PauseWrapper() {}
     virtual void Pause() {}
     virtual void ResetWrapper() {}
     virtual void Reset() {}
@@ -535,6 +566,8 @@ struct FairMQFSM_ : public msmf::state_machine_def<FairMQFSM_>
   protected:
     std::atomic<State> fState;
     std::mutex fChangeStateMutex;
+
+    boost::signals2::signal<void(const State)> fStateChangeCallback;
 };
 
 // reactivate the warning for non-virtual destructor
@@ -582,6 +615,8 @@ class FairMQStateMachine : public FairMQFSM::FairMQFSM
 
     bool WaitForEndOfStateForMs(int state, int durationInMs);
     bool WaitForEndOfStateForMs(std::string state, int durationInMs);
+
+    void OnStateChange(std::function<void(const State)> callback);
 };
 
 #endif /* FAIRMQSTATEMACHINE_H_ */

fairmq/options/FairMQProgOptions.cxx

@@ -351,8 +351,8 @@ void FairMQProgOptions::InitOptionDescription()
     else
     {
         fMQOptionsInCmd.add_options()
-            ("id",                     po::value<string>(),                                     "Device ID (required argument)")
+            ("id",                     po::value<string>(),                                     "Device ID (required argument).")
-            ("io-threads",             po::value<int   >()->default_value(1),                   "Number of I/O threads")
+            ("io-threads",             po::value<int   >()->default_value(1),                   "Number of I/O threads.")
             ("transport",              po::value<string>()->default_value("zeromq"),            "Transport ('zeromq'/'nanomsg').")
             ("config",                 po::value<string>()->default_value("static"),            "Config source ('static'/<config library filename>).")
             ("control",                po::value<string>()->default_value("interactive"),       "States control ('interactive'/'static'/<control library filename>).")