FIX: fixed empty bug

One feature added to EventManager has been disconnecting
a participant at runtime. Unfortunatly, then the
manager of this participant is set to nullptr using
the setManager method. But this method also subscribes to the
shutdown event afterwords. This commit checks if the manager
is nullptr and ignore subscribing in that case.

.drone.yml

@@ -0,0 +1,48 @@
+kind: pipeline
+type: kubernetes
+name: build-amd64
+platform:
+  arch: amd64
+node_selector:
+  kubernetes.io/arch: amd64
+
+steps:
+  - name: submodules
+    image: alpine/git
+    commands:
+      - git submodule update --init --recursive
+
+  - name: build-amd64
+    image: debian:bookworm-slim
+    commands:
+        - apt-get update
+        - apt-get -qy install gcc-12 cmake make build-essential 
+        - mkdir build
+        - cd build; cmake ..
+        - make -j 4 
+        - make test
+
+--- 
+kind: pipeline
+type: kubernetes
+name: build-arm64
+platform:
+  arch: arm64
+node_selector:
+  kubernetes.io/arch: arm64
+
+steps:
+  - name: submodules
+    image: alpine/git
+    commands:
+      - git submodule update --init --recursive
+
+  - name: build-arm64
+    image: debian:bookworm-slim
+    commands:
+        - apt-get update
+        - apt-get -qy install gcc-12 cmake make build-essential
+        - mkdir build
+        - cd build; cmake ..
+        - make -j 4 
+        - make test

CMakeLists.txt

@@ -42,17 +42,17 @@ SET(EVENTMANAGER_SOURCES
   src/EventManager/Manager.cpp
 )
 
-add_library(objlib OBJECT ${EVENTMANAGER_SOURCES})
+add_library(em-objlib OBJECT ${EVENTMANAGER_SOURCES})
-set_property(TARGET objlib PROPERTY POSITION_INDEPENDENT_CODE 1)
+set_property(TARGET em-objlib PROPERTY POSITION_INDEPENDENT_CODE 1)
-target_include_directories(objlib
+target_include_directories(em-objlib
   PUBLIC
           include
   PRIVATE
           src
 )
-target_link_libraries(objlib PUBLIC Threads::Threads)
+target_link_libraries(em-objlib PUBLIC Threads::Threads)
 
-add_library(eventmanager SHARED $<TARGET_OBJECTS:objlib>)
+add_library(eventmanager SHARED $<TARGET_OBJECTS:em-objlib>)
 target_include_directories(eventmanager
   PUBLIC
           include
@@ -61,7 +61,7 @@ target_include_directories(eventmanager
 )
 target_link_libraries(eventmanager PUBLIC Threads::Threads)
 
-add_library(eventmanager-static STATIC $<TARGET_OBJECTS:objlib>)
+add_library(eventmanager-static STATIC $<TARGET_OBJECTS:em-objlib>)
 target_include_directories(eventmanager-static
   PUBLIC
           include
@@ -70,6 +70,7 @@ target_include_directories(eventmanager-static
 )
 target_link_libraries(eventmanager-static PUBLIC Threads::Threads)
 
+IF(${EM_TESTS})
 #
 # add tests as executable
 #
@@ -80,3 +81,5 @@ catch_discover_tests(test_event)
 add_executable(test_basic tests/test_basic.cpp)
 target_link_libraries(test_basic Catch2::Catch2 eventmanager-static)
 catch_discover_tests(test_basic)
+
+ENDIF()

README.md

@@ -10,7 +10,7 @@ A small Event System written as C++20 library.
 
 ## License
 
-GPLv3
+MPLv2
 
 ## Rest
 T.b.d.

include/EventManager/Event.hpp

@@ -22,7 +22,13 @@
 
    /// Eventtype to notify all participants that a shutdown is immanent
    const static std::uint32_t EVENT_TYPE_SHUTDOWN = 0;
-
+   /**
+     * @class Event
+     * @brief An Event is the element in the system that triggers actions from participants
+     *
+     * Derive own events from this class to sned e.g. also a payload to subscribing 
+     * participants. 
+     */
    class Event
    {
     private:
@@ -36,7 +42,7 @@
       std::uint64_t responseId_;
 
       /// identifies if this event is a response to another event
-      std::atomic<bool> isResponse_;
+      bool isResponse_;
 
       /// emitter of the event
       std::shared_ptr<EventManager::Participant> emitter_;
@@ -45,17 +51,17 @@
       /**
       * @brief constructor for creating a simple event
       *
-      * @param type - what kinf of event is this
+      * @param type - what kind of event is this
       */
       Event(std::uint32_t type);
 
       /**
-      * @brief Constructor to create a response Event
+      * @brief Constructor to create a response event
       */
       Event(std::uint32_t type, const EventManager::Event &event);
 
       /**
-      * @brief Constructor to create a response Event
+      * @brief Constructor to create a response event
       */
       Event(std::uint32_t type, const std::shared_ptr<EventManager::Event> event);
 

include/EventManager/Manager.hpp

@@ -25,10 +25,42 @@
  namespace EventManager
  {
 
+    /**
+      * @brief command types
+      */
+    enum class commandType :uint16_t
+    {
+      // command to connect a participant to the manager
+      CONNECT,
+      // command to disconnect a participant from the manager
+      DISCONNECT,
+      // command to schedule start scheduling a participant
+      ENABLE_SCHEDULING,
+      // command to disable scheduling for a participant (not schedule it anymore)
+      DISABLE_SCHEDULING
+    }; // commandType
+
    // forward declaration of EventManager::Participant
    class Participant;
 
-   class Manager
+   /**
+     * @class Manager
+     *
+     * If you use the manager it has to be a shared pointer. Otherwise you will 
+     * get a mem error.
+     *
+     * To add participants to the manager call class function connect.
+     * Calling start method will start the manager.
+     *
+     * Depending on your concept you can first connect all participants to 
+     * the manager and then start it. Or you can start in first and then 
+     * connect the participants. In the first example all participant will
+     * be started at the same time (when calling start from the manager.)
+     * In the second example they will be started when they are connected.
+     * That means if you have one starting event that all participants need
+     * to receive you would choose example 1.
+     */
+   class Manager : public std::enable_shared_from_this<Manager>
    {
      /// the thread the event manager is transmitting events in
      std::unique_ptr<std::thread> mainThread_;
@@ -51,6 +83,9 @@
      /// map holding all the event type and plugin combinations
      std::map<std::uint32_t, std::list<std::shared_ptr<EventManager::Participant>>> eventMap_;
     
+     /// mutex to protect the eventMap
+     std::mutex mutexEventMap_;
+
      /// queue for incomng events
      std::queue<std::shared_ptr<EventManager::Event>> eventQueue_;
 
@@ -63,9 +98,24 @@
      /// list of all plugins requiring scheduling
      std::list<std::shared_ptr<EventManager::Participant>> schedulingParticipants_;
 
-     /// mutex to protect schedulingPlugins_
+     /// mutex to protect list schedulingParticipants_
      std::mutex mutexSchedulingParticipants_;
 
+     /// list of all participants connected
+     std::list<std::shared_ptr<EventManager::Participant>> participants_;
+
+     /// mutex to protect participants_
+     std::mutex mutexParticipants_;
+
+     /// the id for the next participant connecting
+     std::uint32_t nextParticipantID_;
+
+     /// queue for different command requests like e.g. connecting a participant to the manager
+     std::queue<std::pair<EventManager::commandType,std::shared_ptr<EventManager::Participant>>> commandQueue_;
+
+     /// mutex to protect the commandQueue_
+     std::mutex mutexCommandQueue_;
+
          /*
           * all private methods
           */
@@ -85,6 +135,46 @@
      */
      void processEvent(const std::shared_ptr<EventManager::Event> event);
 
+     /**
+     * @brief processes the commands in the commandQueue depending on their type 
+     */
+     void processCommands_();
+     
+     /**
+     * @brief adds the queued participants to the list of connected participants
+     *
+     * The connectionQueue_ contains the participants that should be connected to
+     * the manager. All connected participants are stored in the list participants_.
+     * This class function adds the queued participants to the list participants_
+     * and removes them from the queue.
+     */
+     void processConnect_( std::shared_ptr<EventManager::Participant> participant );
+     
+     /**
+     * @brief removes the given participant form the list of connected participants
+     *
+     * The participants_ list contains the participants that are connected to
+     * the manager. This class function removes the queued participants from the list 
+     * participants_.
+     */
+     void processDisconnect_( std::shared_ptr<EventManager::Participant> participant );
+
+     /**
+     * @brief processes the command to enable scheduling for the given participant
+     * 
+     * This class function adds the given participant to the schedulingParticipants_ list
+     * where the participants are scheduled by calling their schedule class function.
+     */
+     void processEnableScheduling_( std::shared_ptr<EventManager::Participant> participant );
+
+     /**
+     * @brief disables the scheduling of the requested participants
+     *
+     * Removes the participants from the schedulingParticipants_ list that
+     * should not be scheduled anymore.
+     */
+     void processDisableScheduling_( std::shared_ptr<EventManager::Participant> participant );
+
      /**
      * @brief start the main thread
      */
@@ -110,14 +200,16 @@
       /**
       * @brief The constructor for the event manager
       *
-      * Just initializes all attributes to its starting value
+      * Just initializes all attributes to their starting values
       */
       Manager() : mainThread_(nullptr), isMainThreadRunning_(false),
                        stopMainThread_(false), schedulingThread_(nullptr),
-                       isSchedulingThreadRunning_(false), stopSchedulingThread_(false){}
+                       isSchedulingThreadRunning_(false), stopSchedulingThread_(false),
+                       nextParticipantID_(1){}
 
 
       ~Manager();
+
       /**
       * @brief start the event manager
       */
@@ -185,6 +277,23 @@
       */
       void schedule(std::shared_ptr<EventManager::Participant> plugin);
 
+      /**
+       * @brief remove a participant from scheduling
+       * 
+       * @param participant  - the participant to remove
+       */
+      void unschedule(std::shared_ptr<EventManager::Participant> participant);
+
+      /**
+        * @brief method to connect a particpant to the manager
+        */
+      void connect(std::shared_ptr<EventManager::Participant> participant);
+
+      /**
+      * @brief method to disconnect a particpant from the manager
+      */
+      void disconnect(std::shared_ptr<EventManager::Participant> participant);
+
    }; // class Manager
 
  }; //namespace EventManager

include/EventManager/Participant.hpp

@@ -27,10 +27,15 @@
 
    /**
    * @brief The entity participating in the event system.
+   *
+   * If you want the participant to be scheduled from the manager call 
+   * _enableScheduling class function.
    */
    class Participant : public std::enable_shared_from_this<Participant>
    {
     private:
+      /// a unique id for this participant, helpful for debugging
+      std::uint32_t id_;
 
       /// pointer to the event manager
       std::shared_ptr<EventManager::Manager> manager_;
@@ -64,6 +69,14 @@
       */
       virtual void schedule_() { throw std::runtime_error(std::string(__PRETTY_FUNCTION__) + " not implemented");}
 
+      /**
+      * @brief method called by the EventManager::Manager on connecting this particpant
+      *
+      * This method need to be implemented by each child class of EventManager::Participant
+      * Its the best location for subscribing to events and enable scheduling if required.
+      */
+      virtual void init_() {throw std::runtime_error(std::string(__PRETTY_FUNCTION__) + " not implemented");}
+
 
     protected:
 
@@ -92,6 +105,13 @@
       */
       void _waitForEvent();
 
+      /**
+      * @brief wait for a new event with timeout
+      *
+      * @return true  - new event available and queue locked
+      * @return false - no new event, queue not locked, timeout reached
+      */
+      bool _waitForEvent(std::uint32_t timeoutMS);
 
       /**
       * @brief This method subscribes the participant to an event type
@@ -122,18 +142,69 @@
       */
       void _enableScheduling();
 
+      /**
+       * @brief disable scheduling of this particpant through the EventManager::Manager
+       */
+      void _disableScheduling();
+
       /**
       * @brief check if the participant is scheduled by event manager
       */
       bool isScheduledByManager() const {return isScheduledByManager_;}
 
+      /**
+      * @brief connect a new participant through another participant
+      */
+      void connect(std::shared_ptr<EventManager::Participant> participant);
+
+      /**
+      * @brief disconnect a participant through another participant
+      */
+      void disconnect(std::shared_ptr<EventManager::Participant> participant);
+
+      /**
+      * @brief disconnect this participant from the event manager 
+      */
+      void disconnect();
+
     public:
       /**
       * @brief Constructor setting the participant up for use
       */
       Participant();
 
-      void setManager(std::shared_ptr<EventManager::Manager> manager) { manager_=manager;_subscribe(EVENT_TYPE_SHUTDOWN);}
+      /**
+       * @brief Method to set the Manager object
+       *
+       * This method is in general only used by the EventManager::Manager!
+       * Only use this method if you really know what you are doing!
+       *
+       * @param manager - the manager to set
+       */
+      void setManager(std::shared_ptr<EventManager::Manager> manager) 
+        {   manager_=manager; 
+            if (manager_!=nullptr)
+            {
+              _subscribe(EVENT_TYPE_SHUTDOWN);
+            }
+        }
+
+      /**
+       * @brief Method to set the unique id of the participant
+       * 
+       * This method is in general only used by the EventManager::Manager!
+       * Only use this method if you really know what you are doing!
+       *
+       * @param id - the id to set
+       */
+      void setID(const std::uint32_t id) { id_=id;}
+
+      /**
+       * @brief Method to return the unique id of the participant
+       * 
+       * @return std::uint32_t 
+       */
+      std::uint32_t getID() const { return id_;}
 
       /**
       * @brief Method called by the EventManager::Manager to schedule the particpant
@@ -141,6 +212,12 @@
       */
       void schedule() {schedule_();};
 
+      /**
+      * @brief method called by the EventManager::Manager when connecting the particpant
+      *
+      * method calls virtual _init method which has to be implemented by each particpants child class
+      */
+      void init() {init_();}
 
       /**
       * @brief emit an event to this participant

src/EventManager/Manager.cpp

@@ -11,6 +11,8 @@
  #include <EventManager/Manager.hpp>
  #include <EventManager/Participant.hpp>
  #include <iostream>
+ #include <algorithm>
+#include <mutex>
 
  void EventManager::Manager::startMain_()
  {
@@ -86,8 +88,11 @@
      throw std::runtime_error("can not stop main thread");
    }
 
+    if (mainThread_->joinable())
+    {
       mainThread_->join();
     }
+ }
 
  void EventManager::Manager::stopScheduling_()
  {
@@ -105,8 +110,11 @@
      throw std::runtime_error("can not stop scheduling thread");
    }
 
+    if (schedulingThread_->joinable())
+    {
       schedulingThread_->join();
     }
+}
 
  void EventManager::Manager::start()
  {
@@ -196,6 +204,8 @@
        }
      }
      mutexSchedulingParticipants_.unlock();
+
+     processCommands_();
      std::this_thread::sleep_for( std::chrono::milliseconds(100) );
    }
 
@@ -206,6 +216,7 @@
 
  void EventManager::Manager::subscribe(std::uint32_t type, std::shared_ptr<EventManager::Participant> participant)
  {
+   std::lock_guard<std::mutex> lockGuard(mutexEventMap_);
 
    // check if participant is already registered
    auto it = eventMap_.find(type);
@@ -230,6 +241,7 @@
 
  void EventManager::Manager::unsubscribe(std::uint32_t type, std::shared_ptr<EventManager::Participant> participant)
  {
+   std::lock_guard<std::mutex> lockGuard(mutexEventMap_);
    
    auto it = eventMap_.find(type);
 
@@ -283,6 +295,11 @@
  {
    bool isEmpty=true;
   
+   while(!commandQueue_.empty())
+   {
+     std::this_thread::sleep_for(std::chrono::milliseconds(100));
+   }
+
    for (auto it = eventMap_.begin(); it != eventMap_.end(); ++it)
    {
      if ( !(*it).second.empty())
@@ -315,8 +332,84 @@
 
 void EventManager::Manager::schedule(std::shared_ptr<EventManager::Participant> participant)
 {
-   std::lock_guard<std::mutex> guard(mutexSchedulingParticipants_);
+  std::lock_guard<std::mutex> guard( mutexCommandQueue_ );
+  commandQueue_.push( std::make_pair( EventManager::commandType::ENABLE_SCHEDULING, participant) );
+}
 
+
+void EventManager::Manager::unschedule(std::shared_ptr<EventManager::Participant> participant )
+{
+  std::lock_guard<std::mutex> guard( mutexCommandQueue_ );
+  commandQueue_.push( std::make_pair( EventManager::commandType::DISABLE_SCHEDULING, participant) );
+}
+
+
+void EventManager::Manager::processCommands_()
+{
+  std::unique_lock<std::mutex> lk( mutexCommandQueue_ );
+  while( commandQueue_.empty() == false )
+  {
+    auto pair = commandQueue_.front();
+    commandQueue_.pop();
+    lk.unlock();
+    switch( pair.first )
+    {
+      case EventManager::commandType::CONNECT:
+        processConnect_( pair.second );
+        break;
+      case EventManager::commandType::DISCONNECT:
+        processDisconnect_( pair.second );
+        break;
+      case EventManager::commandType::ENABLE_SCHEDULING:
+        processEnableScheduling_( pair.second );
+        break;
+      case EventManager::commandType::DISABLE_SCHEDULING:
+        processDisableScheduling_( pair.second );
+        break;
+    }
+    lk.lock();
+  }
+  lk.unlock();
+}
+
+
+void EventManager::Manager::processConnect_( std::shared_ptr<EventManager::Participant> participant )
+{
+  std::lock_guard<std::mutex> guard(mutexParticipants_);
+  auto it = std::find( participants_.begin(), participants_.end(), participant );
+  if( it == participants_.end() )
+  {
+    participant->setManager(shared_from_this());
+    participant->setID(nextParticipantID_);
+    // we can set and increment here because this critical section is secured by a mutex
+    nextParticipantID_++;
+    participants_.push_back(participant);
+    participant->init();
+  }
+}
+
+
+void EventManager::Manager::processDisconnect_( std::shared_ptr<EventManager::Participant> participant )
+{
+  // before the participant gets disconnected it has to be unscheduled
+  processDisableScheduling_( participant ); 
+
+  // unsubscribe plugin from all events
+  unsubscribe(participant);
+
+  std::lock_guard<std::mutex> guard(mutexParticipants_);
+  auto it = std::find( participants_.begin(), participants_.end(), participant );
+  if( it != participants_.end() )
+  {
+    participant->setManager(nullptr);
+    participants_.erase( it );
+  }
+}
+
+
+void EventManager::Manager::processEnableScheduling_( std::shared_ptr<EventManager::Participant> participant )
+{
+  std::lock_guard<std::mutex> guard(mutexSchedulingParticipants_);
   auto it = std::find( schedulingParticipants_.begin(), schedulingParticipants_.end(), participant );
 
   if( it == schedulingParticipants_.end() )
@@ -324,3 +417,29 @@
     schedulingParticipants_.push_back( participant );
   }
 }
+
+
+void EventManager::Manager::processDisableScheduling_( std::shared_ptr<EventManager::Participant> participant )
+{
+  std::lock_guard<std::mutex> guard( mutexSchedulingParticipants_ );
+  auto it = std::find( schedulingParticipants_.begin(), schedulingParticipants_.end(), participant );
+
+  if( it != schedulingParticipants_.end() ) 
+  {
+    schedulingParticipants_.erase(it);
+  }
+}
+
+
+void EventManager::Manager::connect( std::shared_ptr<EventManager::Participant> participant )
+{
+  std::lock_guard<std::mutex> guard( mutexCommandQueue_ );
+  commandQueue_.push( std::make_pair( EventManager::commandType::CONNECT, participant) );
+}
+
+
+void EventManager::Manager::disconnect( std::shared_ptr<EventManager::Participant> participant )
+{ 
+  std::lock_guard<std::mutex> guard( mutexCommandQueue_ );
+  commandQueue_.push( std::make_pair( EventManager::commandType::DISCONNECT, participant) );
+}

src/EventManager/Participant.cpp

@@ -11,6 +11,9 @@
  #include <EventManager/Participant.hpp>
  #include <EventManager/Manager.hpp>
  #include <iostream>
+ #include <chrono>
+ using namespace std::chrono_literals;
+
 
 EventManager::Participant::Participant() : manager_(nullptr),
                                   isScheduledByManager_(false), isQueueLocked_(false)
@@ -18,6 +21,35 @@ EventManager::Participant::Participant() : manager_(nullptr),
 
 }
 
+void EventManager::Participant::connect(std::shared_ptr<EventManager::Participant> participant)
+{
+  if (manager_ == nullptr)
+  {
+    throw std::runtime_error("no event manager set yet");
+  }
+  manager_->connect(participant);
+}
+
+void EventManager::Participant::disconnect(std::shared_ptr<EventManager::Participant> participant)
+{
+  if (manager_ == nullptr)
+  {
+    throw std::runtime_error("no event manager set yet");
+  }
+  manager_->disconnect(participant);
+}
+
+
+void EventManager::Participant::disconnect()
+{
+  if (manager_ == nullptr)
+  {
+    throw std::runtime_error("no event manager set yet");
+  }
+  manager_->disconnect(shared_from_this());
+}
+
+
 void EventManager::Participant::emit(std::shared_ptr<EventManager::Event> event)
 {
   {
@@ -72,6 +104,20 @@ void EventManager::Participant::_waitForEvent()
   isQueueLocked_=true;
 }
 
+bool EventManager::Participant::_waitForEvent(std::uint32_t timeoutMS)
+{
+  std::unique_lock<std::mutex> lock(mutexEventQueue_);
+  if (newEventInQueue_.wait_for(lock,timeoutMS*1ms)==std::cv_status::no_timeout)
+  {
+    isQueueLocked_=true;
+    return true;
+  }
+
+  return false;
+
+}
+
+
 void EventManager::Participant::_enableScheduling()
 {
   if (manager_ == nullptr)
@@ -82,6 +128,14 @@ void EventManager::Participant::_enableScheduling()
   isScheduledByManager_=true;
 }
 
+void EventManager::Participant::_disableScheduling() {
+  if (manager_ == nullptr) {
+    throw std::runtime_error("no event manager set yet");
+  }
+  manager_->unschedule(shared_from_this());
+  isScheduledByManager_ = false;
+}
+
 void EventManager::Participant::_subscribe(std::uint32_t type)
 {
   if (manager_ == nullptr)

tests/test_basic.cpp

@@ -33,6 +33,11 @@ class myParticipant : public EventManager::Participant
       _unsubscribe();
     }
 
+    void init_() {
+      _subscribe(eventType_);
+      _enableScheduling();
+    }
+
     void schedule_() {
       std::shared_ptr<EventManager::Event> event = nullptr;
 
@@ -67,10 +72,7 @@ class myParticipant : public EventManager::Participant
 
     bool eventReceived() const { return receivedEvent_;}
 
-    void init() {
+
-      _subscribe(eventType_);
-      _enableScheduling();
-    }
 };
 
 
@@ -85,14 +87,18 @@ SCENARIO("Basic Usage of EventManager", "[Manager]")
       manager = std::make_shared<EventManager::Manager>();
     }());
 
+    REQUIRE_NOTHROW([&]()
+    {
+      manager->start();
+    }());
+
     REQUIRE(manager->empty() == true);
 
     std::shared_ptr<myParticipant> participant0;
     REQUIRE_NOTHROW([&]()
     {
       participant0 = std::make_shared<myParticipant>(0,TEST_EVENT0);
-      participant0->setManager(manager);
+      manager->connect(participant0);
-      participant0->init();
     }());
 
     REQUIRE(manager->empty() == false);
@@ -101,17 +107,11 @@ SCENARIO("Basic Usage of EventManager", "[Manager]")
     REQUIRE_NOTHROW([&]()
     {
       participant1 = std::make_shared<myParticipant>(1,TEST_EVENT1);
-      participant1->setManager(manager);
+      manager->connect(participant1);
-      participant1->init();
     }());
 
     REQUIRE(manager->empty() == false);
 
-    REQUIRE_NOTHROW([&]()
-    {
-      manager->start();
-    }());
-
     REQUIRE(manager->isRunning() == true);
 
     WHEN("emitting shutdown event")