[occ] Try to make plugin unsubscribe from FairMQ at right time

AliceO2Group · Nov 10, 2023 · 9a20ea9 · 9a20ea9
1 parent 3e8163f
commit 9a20ea9
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 4 deletions.
diff --git a/occ/plugin/OccFMQCommon.cxx b/occ/plugin/OccFMQCommon.cxx
@@ -148,7 +148,9 @@ std::tuple<OccLite::nopb::TransitionResponse, ::grpc::Status> doTransition(fair:
 
     m_pluginServices->SubscribeToDeviceStateChange(id, onDeviceStateChange);
     DEFER({
-        m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        if ( !newStates.empty() && newStates.back() != "EXITING") {
+            m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        }
     });
 
     try {
@@ -220,6 +222,7 @@ std::tuple<OccLite::nopb::TransitionResponse, ::grpc::Status> doTransition(fair:
     }
 
     if (newStates.back() == "EXITING") {
+        m_pluginServices->UnsubscribeFromDeviceStateChange(id);
         m_pluginServices->ReleaseDeviceControl(FMQ_CONTROLLER_NAME);
         OLOG(debug) << "releasing device control";
     }

diff --git a/occ/plugin/OccLiteServer.cxx b/occ/plugin/OccLiteServer.cxx
@@ -122,10 +122,12 @@ OccLite::Service::EventStream(::grpc::ServerContext* context, const OccLite::nop
     std::mutex writer_mu;
     std::condition_variable finished;
     std::mutex finished_mu;
+    std::string last_known_state;
 
     auto onDeviceStateChange = [&](fair::mq::PluginServices::DeviceState reachedState) {
         std::lock_guard<std::mutex> lock(writer_mu);
         auto state = fair::mq::PluginServices::ToStr(reachedState);
+        last_known_state = state;
 
         OLOG(debug) << "[EventStream] new state: " << state;
 
@@ -147,7 +149,9 @@ OccLite::Service::EventStream(::grpc::ServerContext* context, const OccLite::nop
 
     m_pluginServices->SubscribeToDeviceStateChange(id, onDeviceStateChange);
     DEFER({
-        m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        if (last_known_state == "EXITING") {
+            m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        }
     });
 
     {

diff --git a/occ/plugin/OccPluginServer.cxx b/occ/plugin/OccPluginServer.cxx
@@ -58,10 +58,12 @@ OccPluginServer::EventStream(grpc::ServerContext* context,
     std::mutex writer_mu;
     std::condition_variable finished;
     std::mutex finished_mu;
+    std::string last_known_state;
 
     auto onDeviceStateChange = [&](fair::mq::PluginServices::DeviceState reachedState) {
         std::lock_guard<std::mutex> lock(writer_mu);
         auto state = fair::mq::PluginServices::ToStr(reachedState);
+        last_known_state = state
 
         OLOG(debug) << "[EventStream] new state: " << state;
 
@@ -83,7 +85,9 @@ OccPluginServer::EventStream(grpc::ServerContext* context,
 
     m_pluginServices->SubscribeToDeviceStateChange(id, onDeviceStateChange);
     DEFER({
-        m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        if (last_known_state == "EXITING") {
+            m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        }
     });
 
     {
@@ -106,10 +110,12 @@ OccPluginServer::StateStream(grpc::ServerContext* context,
     std::mutex writer_mu;
     std::condition_variable finished;
     std::mutex finished_mu;
+    std::string last_known_state;
 
     auto onDeviceStateChange = [&](fair::mq::PluginServices::DeviceState reachedState) {
         std::lock_guard<std::mutex> lock(writer_mu);
         auto state = fair::mq::PluginServices::ToStr(reachedState);
+        last_known_state = state
         pb::StateType sType = isIntermediateFMQState(state) ? pb::STATE_INTERMEDIATE : pb::STATE_STABLE;
 
         pb::StateStreamReply response;
@@ -132,7 +138,9 @@ OccPluginServer::StateStream(grpc::ServerContext* context,
 
     m_pluginServices->SubscribeToDeviceStateChange(id, onDeviceStateChange);
     DEFER({
-        m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        if (last_known_state == "EXITING") {
+            m_pluginServices->UnsubscribeFromDeviceStateChange(id);
+        }
     });
 
     {