DPL: Unblock the event loop when undergoing state transitions

Author: ktf

Framework/Core/src/DataProcessingDevice.cxx

@@ -262,6 +262,21 @@ void DataProcessingDevice::Init()
       mState.inputChannelInfos[ci].state = InputChannelState::Pull;
     }
   }
+  /// This should post a message on the queue...
+  SubscribeToNewTransition("dpl", [loop = mState.loop](fair::mq::Transition t) {
+    if (loop) {
+      uv_async_t handle;
+      int res = uv_async_init(loop, &handle, nullptr);
+      if (res < 0) {
+        LOG(ERROR) << "Unable to initialise subscription";
+      }
+      res = uv_async_send(&handle);
+      if (res < 0) {
+        LOG(ERROR) << "Unable to notify subscription";
+      }
+      LOG(debug) << "State transition requested";
+    }
+  });
 }
 
 void on_signal_callback(uv_signal_t* handle, int signum)
@@ -438,6 +453,11 @@ bool DataProcessingDevice::ConditionalRun()
       shouldNotWait = true;
     }
     uv_run(mState.loop, shouldNotWait ? UV_RUN_NOWAIT : UV_RUN_ONCE);
+
+    // A new state was requested, we exit.
+    if (NewStatePending()) {
+      return false;
+    }
   }
 
   // Notify on the main thread the new region callbacks, making sure