pubBase.hpp

Go to the documentation of this file.

 
#ifndef CCF_COMMON_PUB_BASE_HPP_H_
#define CCF_COMMON_PUB_BASE_HPP_H_
 
#include <ccf/common/base.hpp>
#include <ccf/common/dataFrame.hpp>
#include <ccf/common/setup.hpp>
#include <ccf/common/db.hpp>
#include <ccf/common/appBase.hpp>
 
#include <Recif.hpp>
 
namespace ccf::common {
 
  // @{
  const std::string DB_NODE_PUB_ID                 = "id";
  const std::string DB_NODE_PUB_NAME               = "name";
  const std::string DB_NODE_ADAPTER                = "adapter";
  const std::string DB_NODE_PUB_START              = "start";
  const std::string DB_NODE_PUB_STATUS             = "status";
  const std::string DB_NODE_PUB_FRAMES_HANDLED     = "frames_handled";
  const std::string DB_NODE_PUB_VOLUME_HANDLED     = "volume_handled";
  const std::string DB_NODE_PUB_NB_OF_FILES_STORED = "nb_of_files_stored";
  const std::string DB_NODE_PUB_FILES_STORED       = "files_stored";
  // @}
 
  class PublisherStatus: public Base {
  public:
 
    PublisherStatus(const std::string& pub_id,
                    const std::string obj_id = "ccf::common::PublisherStatus");
 
    PublisherStatus(const PublisherStatus& source);
 
    PublisherStatus();
 
    virtual ~PublisherStatus();
 
    void Reset();
 
    const std::string& GetPubId() const;
 
    void SetStatus(const PubStatus status);
 
    PubStatus GetStatus() const;
 
    double GetStartTime() const;
 
    void IncrFramesHandled();
 
    int64_t GetFramesHandled() const;
 
    void IncrVolumeHandled(const uint32_t vol);
 
    int64_t GetVolumeHandled() const;
 
    void AddFileStored(const std::string& new_file);
 
    uint64_t GetNbOfFilesStored() const;
 
    void GetFilesStored(std::vector<std::string>& files_stored) const;
 
    void SetFramesHandled(const int64_t nb_of_frames_handled);
 
    void SetVolumeHandled(const int64_t volume_handled);
 
    void SetFilesStored(const std::vector<std::string>& files_stored);
 
    std::string ToString() const;
 
    PublisherStatus& operator = (const PublisherStatus& source);
 
  protected:
 
  private:
    void _Copy(const PublisherStatus& source);
 
    std::string                         m_obj_id;
    std::string                         m_pub_id;
    PubStatus                           m_pub_status;
    double                              m_start;
    int64_t                             m_frames_handled;
    int64_t                             m_volume_handled;
    std::map<std::string, std::string>  m_files_stored;
  };
 
  class PubBase: public Base {
  public:
 
    static const std::string& GenId(const std::string& proc_thread_name,
                                    const std::string& pub_thread_name);
 
    template <class TYPE>
    static void AddPubFactoryObj(TYPE& pub_factory_obj) {
      LOG4CPLUS_TRACE_METHOD(Logger(), __PRETTY_FUNCTION__);
 
      if (pub_factory_obj.GetClassName() == "") {
        CCFTHROW("Must define a classname for a Data Publisher Plug-In");
      }
 
      BEGIN_CRIT_SEC("ccf::common::PubBase") {
        if (s_pub_factory_objects.find(pub_factory_obj.GetClassName()) !=
          s_pub_factory_objects.end()) {
            CCFTHROW(fmt::format("Factory object with classname: {} already exists",
                                 pub_factory_obj.GetClassName()));
        }
        CCFINFO(Logger(), "Registering Data Pub factory object. Classname: " <<
          pub_factory_obj.GetClassName());
        s_pub_factory_objects[pub_factory_obj.GetClassName()] = std::make_shared<TYPE>();
      } END_CRIT_SEC();
 
    }
 
    static const std::map<std::string, std::shared_ptr<PubBase>>& GetPubFactoryObjs();
 
    // static void CreatePubObj(const std::string& class_name,
    //                          const std::string& proc_thread_name,
    //                          const std::string& pub_thread_name,
    //                          std::shared_ptr<PubBase>& new_object);
    static void CreatePubObj(const std::string& class_name,
                             const std::string& proc_thread_name,
                             const std::string& pub_thread_name,
                             PubBase** new_object);
 
    // static void GetPubObj(const std::string& proc_thread_name,
    //                       const std::string& pub_thread_name,
    //                       std::shared_ptr<PubBase>& pub_obj,
    //                       const bool initialise = true);
    static void GetPubObj(const std::string& proc_thread_name,
                          const std::string& pub_thread_name,
                          PubBase** pub_obj,
                          const bool initialise = true);
 
    //static void GetPubObjs(std::vector<std::shared_ptr<PubBase>>& pub_objs);
    static void GetPubObjs(std::vector<PubBase*>& pub_objs);
 
    // static bool HasPubObj(const std::string& proc_thread_name,
    //                       const std::string& pub_thread_name,
    //                       std::shared_ptr<PubBase>& pub_obj);
    static bool HasPubObj(const std::string& proc_thread_name,
                          const std::string& pub_thread_name,
                          PubBase** pub_obj);
 
    PubBase(const std::string& proc_thread_name,
            const std::string& pub_thread_name);
    PubBase();
    virtual ~PubBase();
 
    void Initialise();
    bool GetInitialised() const;
 
    void Dismantle();
 
    void SetEnabled(const bool enabled);
    bool GetEnabled() const;
 
    const std::string& GetProcThreadName() const;
    const std::string& GetPubId() const;
    const std::string& GetPubThreadName() const;
 
    void Publish(DataFrame& frame,
                 const bool force);
 
    PubType GetPublisherType() const;
 
    void Activate();
    virtual void ActivateUser();
    void Deactivate();
    virtual void DeactivateUser();
    void SetActivated(const bool activated);
    bool GetActivated() const;
    virtual void CheckForDeactivationUser();
 
    PublisherStatus& GetPubStatus();
 
    void EstimateExecution(double& duration,
                           int64_t& nb_of_frames);
 
    void GetPubStatus(PublisherStatus& pub_stat) const;
 
    virtual void InitialiseUser();
 
    virtual void DismantleUser();
 
    virtual void PublishUser(DataFrame& frame);
 
    // virtual void CreateObjectUser(const std::string& proc_thread_name,
    //                               const std::string& pub_thread_name,
    //                               std::shared_ptr<PubBase>& new_object);
 
    virtual void CreateObjectUser(const std::string& proc_thread_name,
                                  const std::string& pub_thread_name,
                                  PubBase** new_object);
 
 
    virtual void EstimateExecutionUser(double& duration,
                                       int64_t& nb_of_frames);
 
    void ResetDbStatus() const;
    void UpdateDbStatus() const;
 
    // TODO: Add:
    // virtual void UpdateDbStatusUser().
    
    std::string ToString() const;
 
    // TODO: Add + invoke when a setup command is received.
    // void HandleSetup();
    // void HandleSetupUser();
 
 
  protected:
    PubType       m_publisher_type;
    bool          m_initialised;
    bool          m_setup_enabled;
    bool          m_stat_active;
 
  private:
    static std::map<std::string, std::shared_ptr<PubBase>>  s_pub_factory_objects;
    //static std::map<std::string, std::shared_ptr<PubBase>>  s_pub_objects;
    static std::map<std::string, PubBase*>  s_pub_objects;
 
    std::string        m_pub_id;
    std::string        m_proc_thread_name;
    std::string        m_pub_thread_name;
    double             m_delay;
    std::string        m_db_path;
 
    std::shared_ptr<PublisherStatus>   m_pub_status;
  };
 
}
 
#endif // CCF_COMMON_PUB_BASE_HPP_H_