test(PluginServices): state change subscription thread-safety

test/plugin_services/_control.cxx

@@ -1,14 +1,18 @@
 /********************************************************************************
- *    Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
+ * Copyright (C) 2017-2023 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH  *
  *                                                                              *
  *              This software is distributed under the terms of the             *
  *              GNU Lesser General Public Licence (LGPL) version 3,             *
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
 #include "Fixture.h"
+#include <array>
 #include <condition_variable>
+#include <fairmq/Tools.h>
+#include <memory>
 #include <mutex>
+#include <string>
 
 namespace
 {
@@ -142,4 +146,27 @@ TEST_F(PluginServices, ControlStateTransitionConversions)
     EXPECT_NO_THROW(mServices.ToStr(DeviceStateTransition::ErrorFound));
 }
 
+TEST_F(PluginServices, SubscriptionThreadSafety)
+{
+    // obviously not a perfect test, but I could segfault fmq reliably with it (without the fix)
+
+    constexpr auto attempts = 1000;
+    constexpr auto subscribers = 5;
+
+    std::array<std::unique_ptr<std::thread>, subscribers> threads;
+    auto id = 0;
+    for (auto& thread : threads) {
+        thread = std::make_unique<std::thread>([&](){
+            auto const subscriber = fair::mq::tools::ToString("subscriber_", id);
+            for (auto i = 0; i < attempts; ++i) {
+                mServices.SubscribeToDeviceStateChange(subscriber, [](DeviceState){});
+                mServices.UnsubscribeFromDeviceStateChange(subscriber);
+            }
+        });
+        ++id;
+    }
+
+    for (auto& thread : threads) { thread->join(); }
+}
+
 } /* namespace */