ZMQSubscriptionManager.h

Go to the documentation of this file.

#ifndef CHIMERATK_DOOCS_BACKEND_ZMQSUBSCRIPTIONMANAGER_H
#define CHIMERATK_DOOCS_BACKEND_ZMQSUBSCRIPTIONMANAGER_H
 
#include <ChimeraTK/Exception.h>
 
#include <boost/shared_ptr.hpp>
 
#include <eq_fct.h>
#include <pthread.h>
 
#include <map>
#include <mutex>
#include <string>
#include <vector>
 
namespace ChimeraTK {
 
  class DoocsBackendRegisterAccessorBase;
  class DoocsBackend;
 
  namespace DoocsBackendNamespace {
 
    class ZMQSubscriptionManager {
     public:
      static ZMQSubscriptionManager& getInstance() {
        static ZMQSubscriptionManager manager;
        return manager;
      }
 
      void subscribe(const std::string& path, DoocsBackendRegisterAccessorBase* accessor);
 
      void unsubscribe(const std::string& path, DoocsBackendRegisterAccessorBase* accessor);
 
      void activateAllListeners(DoocsBackend* backend);
 
      void deactivateAllListeners(DoocsBackend* backend);
 
      void deactivateAllListenersAndPushException(DoocsBackend* backend);
 
     private:
      ZMQSubscriptionManager();
      ~ZMQSubscriptionManager();
 
      // Activate ZeroMQ subscription.
      // Caller need to own subscriptionMap_mutex already.
      void activateSubscription(const std::string& path);
 
      // Deactivate ZeroMQ subscription. Caller must not own subscriptionMap_mutex or any listeners_mutex.
      void deactivateSubscription(const std::string& path);
 
      // Poll initial value via RPC call and push it into the queues
      void pollInitialValue(const std::string& path, const std::list<DoocsBackendRegisterAccessorBase*>& accessors);
 
      // Push error to listener
      static void pushError(DoocsBackendRegisterAccessorBase* listener, const std::string& message);
 
      bool dmsgStartCalled{false};
      std::mutex dmsgStartCalled_mutex;
 
      struct Subscription {
        Subscription() : zqmThreadId(ZMQSubscriptionManager::pthread_t_invalid) {}
 
        std::vector<DoocsBackendRegisterAccessorBase*> listeners;
 
        std::mutex listeners_mutex;
 
        pthread_t zqmThreadId;
 
        bool active{false};
 
        bool started{false};
        std::condition_variable startedCv{};
 
        bool gotInitialValue{false};
      };
 
      static pthread_t pthread_t_invalid;
 
      std::map<std::string, Subscription> subscriptionMap;
 
      std::mutex subscriptionMap_mutex;
 
      static void zmq_callback(void* self, doocs::EqData* data, dmsg_info_t* info);
    };
 
  } // namespace DoocsBackendNamespace
 
} // namespace ChimeraTK
 
#endif // CHIMERATK_DOOCS_BACKEND_ZMQSUBSCRIPTIONMANAGER_H