Publisher.h

Go to the documentation of this file.

// $Id: Publisher.h 4472 2020-04-15 13:33:18Z linev $
 
/************************************************************
 * The Data Acquisition Backbone Core (DABC)                *
 ************************************************************
 * Copyright (C) 2009 -                                     *
 * GSI Helmholtzzentrum fuer Schwerionenforschung GmbH      *
 * Planckstr. 1, 64291 Darmstadt, Germany                   *
 * Contact:  http://dabc.gsi.de                             *
 ************************************************************
 * This software can be used under the GPL license          *
 * agreements as stated in LICENSE.txt file                 *
 * which is part of the distribution.                       *
 ************************************************************/
 
#ifndef DABC_Publisher
#define DABC_Publisher
 
#ifndef DABC_Worker
#include "dabc/Worker.h"
#endif
 
namespace dabc {
 
   class PublisherRef;
   class HierarchyStore;
 
   class CmdPublisher : public Command {
      DABC_COMMAND(CmdPublisher, "CmdPublisher");
   };
 
   class CmdGetBinary : public Command {
      DABC_COMMAND(CmdGetBinary, "CmdGetBinary");
 
      CmdGetBinary(const std::string &path, const std::string &kind, const std::string &query) :
         Command(CmdName())
      {
         SetStr("Item", path);
         SetStr("Kind", kind);
         SetStr("Query", query);
      }
   };
 
   class CmdHierarchyExec : public Command {
      DABC_COMMAND(CmdHierarchyExec, "CmdHierarchyExec");
 
      CmdHierarchyExec(const std::string &path) :
         Command(CmdName())
      {
         SetStr("Item", path);
      }
   };
 
   class CmdGetNamesList : public Command {
      DABC_COMMAND(CmdGetNamesList, "CmdGetNamesList");
 
      CmdGetNamesList(const std::string &kind, const std::string &path, const std::string &query) :
         Command(CmdName())
      {
         SetStr("textkind", kind);
         SetStr("path", path);
         SetStr("query", query);
      }
 
      void AddHeader(const std::string &name, const std::string &value, bool withquotes = true)
      {
         int num = GetInt("NumHdrs");
         SetStr(dabc::format("OptHdrName%d", num), name);
         SetStr(dabc::format("OptHdrValue%d", num), withquotes ? std::string("\"") + value + "\"" : value);
         SetInt("NumHdrs", num+1);
      }
 
      static void SetResNamesList(dabc::Command &cmd, Hierarchy& res);
 
      static Hierarchy GetResNamesList(dabc::Command &cmd);
   };
 
 
   class CmdSubscriber : public Command {
      DABC_COMMAND(CmdSubscriber, "CmdSubscriber");
 
      void SetEntry(const Hierarchy& e) { SetRef("entry", e); }
      Hierarchy GetEntry() { return GetRef("entry"); }
   };
 
   struct PublisherEntry {
      unsigned id;          // unique id in the worker
      std::string path;     // absolute path in hierarchy
      std::string worker;   // worker which is manages hierarchy
      std::string fulladdr;  // address, used to identify producer of the hierarchy branch
      void* hier;           // local hierarchy pointer, one not allowed to use it from publisher
      uint64_t version;     // last requested version
      uint64_t lastglvers;  // last version, used to build global
      bool local;           // is entry from local node
      bool mgrsubitem;      // if true, belongs to manager hierarchy
      int  errcnt;          // counter of consequent errors
      bool waiting_publisher; // indicate if next request is submitted
      Hierarchy rem;        // remote hierarchy
      HierarchyStore* store; // store object for the registered hierarchy
 
      PublisherEntry() :
         id(0), path(), worker(), fulladdr(), hier(0),
         version(0), lastglvers(0), local(true), mgrsubitem(false),
         errcnt(0), waiting_publisher(false), rem(), store(0) {}
 
      // implement copy constructor to avoid any extra copies which are not necessary
      PublisherEntry(const PublisherEntry& src) :
         id(src.id), path(), worker(), fulladdr(), hier(0),
         version(0), lastglvers(0), local(true), mgrsubitem(false),
         errcnt(0), waiting_publisher(false), rem(), store(0) {}
 
      ~PublisherEntry();
 
      // implement assign operator to avoid any extra copies which are not necessary
      PublisherEntry& operator=(const PublisherEntry& src)
      {
         id = src.id;
         return *this;
      }
   };
 
   struct SubscriberEntry {
      unsigned id;          // unique id in the worker
      std::string path;     // absolute path in hierarchy
      std::string worker;   // worker which is manages hierarchy
      bool local;           // is entry from local node
      int hlimit;           // history limit -1 - DNS, 0 - no history
      Hierarchy fEntry;     // entry which is provided to the worker as result
      uint64_t version;     // last version provided to subscriber
      bool waiting_worker;  // when enabled, we are waiting while worker process entry
 
      SubscriberEntry() :
         id(0), path(), worker(), local(true),
         hlimit(0), fEntry(), version(0), waiting_worker(false) {}
   };
 
   class Publisher : public dabc::Worker {
 
      friend class PublisherRef;
 
      protected:
 
         Hierarchy fGlobal;  
 
         Hierarchy fLocal;   
 
         uint64_t fLastLocalVers; 
 
         typedef std::list<PublisherEntry> PublishersList;
 
         typedef std::list<SubscriberEntry> SubscribersList;
 
         PublishersList fPublishers;
 
         SubscribersList fSubscribers;
 
         unsigned fCnt;            
 
         std::string fMgrPath;     
         Hierarchy   fMgrHiearchy; 
 
         std::string fStoreDir;   
         std::string fStoreSel;   
         int         fFileLimit;  
         int         fTimeLimit;  
         double      fStorePeriod; 
 
         virtual void OnThreadAssigned();
 
         virtual double ProcessTimeout(double last_diff);
 
         virtual int ExecuteCommand(Command cmd);
 
         virtual bool ReplyCommand(Command cmd);
 
         dabc::Command CreateExeCmd(const std::string &path, const std::string &query, dabc::Command tgt = nullptr);
 
         void CheckDnsSubscribers();
 
         void InvalidateGlobal();
 
         bool ApplyEntryDiff(unsigned recid, dabc::Buffer& buf, uint64_t version, bool witherror = false);
 
         bool DoStorage() const { return !fStoreDir.empty(); }
 
         Hierarchy GetWorkItem(const std::string &path, bool *islocal = nullptr);
 
         bool RedirectCommand(dabc::Command cmd, const std::string &itemname);
 
         bool IdentifyItem(bool asproducer, const std::string &itemname, bool &islocal, std::string &producer_name, std::string &request_name);
 
      public:
 
         Publisher(const std::string &name, dabc::Command cmd = nullptr);
         virtual ~Publisher();
 
         static const char* DfltName() { return "/publ"; }
 
         virtual const char* ClassName() const { return "Publisher"; }
   };
 
   // ==============================================================
 
   class PublisherRef : public WorkerRef {
      DABC_REFERENCE(PublisherRef, WorkerRef, Publisher)
 
      bool Register(const std::string &path, const std::string &workername, void* hierarchy)
      {  return OwnCommand(1, path, workername, hierarchy); }
 
      bool Unregister(const std::string &path, const std::string &workername, void* hierarchy)
      {  return OwnCommand(2, path, workername, hierarchy); }
 
      bool Subscribe(const std::string &path, const std::string &workername)
      {  return OwnCommand(3, path, workername); }
 
      bool Unsubscribe(const std::string &path, const std::string &workername)
      {  return OwnCommand(4, path, workername); }
 
      bool RemoveWorker(const std::string &workername, bool sync = true)
      {  return OwnCommand(sync ? 5 : -5, "", workername); }
 
      bool AddRemote(const std::string &remnode, const std::string &workername)
      {  return OwnCommand(6, remnode, workername); }
 
      std::string UserInterfaceKind(const char *uri, std::string &path, std::string &fname);
 
      int NeedAuth(const std::string &path);
 
      Buffer GetBinary(const std::string &fullname, const std::string &kind, const std::string &query, double tmout = 5.);
 
      Hierarchy GetItem(const std::string &fullname, const std::string &query = "", double tmout = 5.);
 
      Command ExeCmd(const std::string &fullname, const std::string &query);
 
      protected:
         bool OwnCommand(int id, const std::string &path, const std::string &workername, void *hier = nullptr);
   };
 
}
 
#endif