Port.h

Go to the documentation of this file.

// $Id: Port.h 4694 2021-02-23 13:54:37Z 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_Port
#define DABC_Port
 
#ifndef DABC_ModuleItem
#include "dabc/ModuleItem.h"
#endif
 
#ifndef DABC_LocalTransport
#include "dabc/LocalTransport.h"
#endif
 
#ifndef DABC_ConnectionRequest
#include "dabc/ConnectionRequest.h"
#endif
 
namespace dabc {
 
   class PortRef;
   class Port;
   class Module;
   class ModuleAsync;
   class ModuleSync;
   class ConnTimer;
 
   class Port : public ModuleItem {
 
      friend class Module;
      friend class ModuleAsync;
      friend class ModuleSync;
      friend class PortRef;
      friend class LocalTransport;
      friend class ConnTimer;
 
      public:
         // TODO: provide meaningful names
         enum EventsProducing {
            SignalNone = 0,   // port will not produce any events
            SignalConfirm,    // next event can be produced when previous event is received, confirmed and
                              // event consumer performed next operation
            SignalOperation,  // next event can be produced when event consumer performs next operation
            SignalEvery       // every operation will produce event
         };
 
      protected:
         unsigned           fQueueCapacity;     
         Parameter          fRate;              
         EventsProducing    fSignal;            
         LocalTransportRef  fQueue;             
         std::string        fBindName;          
         std::string        fRateName;          
         unsigned           fMaxLoopLength;     
         double             fReconnectPeriod;   
         int                fReconnectLimit;    
         bool               fDoingReconnect;    
         std::string        fOnError;           
 
 
         virtual void ObjectCleanup();
 
         virtual void DoStart();
         virtual void DoStop();
         virtual void DoCleanup();
 
         Port(int kind, Reference parent,
                  const std::string &name,
                  unsigned queuesize);
         virtual ~Port();
 
         void SetQueue(Reference& ref);
 
         inline void ConfirmEvent()
         { if (SignalingKind() == SignalConfirm) fQueue.ConfirmEvent(GetType()==mitOutPort); }
 
         int GetMaxLoopLength();
 
         void ReadPortConfiguration();
 
         virtual unsigned NumStartEvents() { return 0; }
 
         ConnectionRequest GetConnReq(bool force = false);
 
         void RemoveConnReq();
 
         unsigned QueueCapacity() const;
 
         void Disconnect(bool witherr = false) { fQueue.Disconnect(IsInput(), witherr); }
 
         void SetBindName(const std::string &name);
 
         std::string GetBindName() const;
 
         bool IsConnected() const { return fQueue.IsConnected(); }
 
         void ConfigureReconnect(double period = -1, int numtry = -1)
         {
            fReconnectPeriod = period;
            fReconnectLimit = numtry;
            if ((fReconnectPeriod <= 0) && (fReconnectLimit >= 0)) fReconnectPeriod = 1.; else
            // if ((fReconnectPeriod > 0) && (fReconnectLimit == -1)) fReconnectLimit = 10; else
            if (fReconnectPeriod <= 0) SetDoingReconnect(false);
         }
 
         bool TryNextReconnect(bool caused_by_error, bool can_disconnect = true);
 
         double GetReconnectPeriod() const { return fReconnectPeriod; }
 
         bool IsDoingReconnect() const { return fDoingReconnect; }
 
         void SetDoingReconnect(bool on = true) { fDoingReconnect = on; }
 
         void SetRateMeter(const Parameter& ref);
 
         void SetMaxLoopLength(unsigned cnt) { fMaxLoopLength = cnt; }
 
         bool SetSignaling(EventsProducing kind);
 
         EventsProducing SignalingKind() const { return fSignal; }
 
         bool SubmitCommandToTransport(Command cmd);
 
         void ConfigureOnError(const std::string &action = "") { fOnError = action; }
 
      public:
 
         virtual const char* ClassName() const { return "Port"; }
 
         virtual bool IsInput() const { return false; }
         virtual bool IsOutput() const { return false; }
   };
 
   // __________________________________________________________________________
 
 
   class PortRef : public ModuleItemRef {
      DABC_REFERENCE(PortRef, ModuleItemRef, Port)
 
      
      bool IsInput() const { return GetObject() ? GetObject()->IsInput() : false; }
 
      bool IsOutput() const { return GetObject() ? GetObject()->IsOutput() : false; }
 
      unsigned QueueCapacity() const { return GetObject() ? GetObject()->QueueCapacity() : 0; }
 
      int GetSignalingKind();
 
      bool IsConnected();
 
      bool Disconnect(bool witherr = false);
 
      PortRef GetBindPort();
 
      void ConfigureOnError(const std::string &action = "")
         { if (GetObject()) GetObject()->ConfigureOnError(action); }
 
      void ConfigureReconnect(double period = 1., int numtry = -1)
         { if (GetObject()) GetObject()->ConfigureReconnect(period, numtry); }
 
      ConnectionRequest MakeConnReq(const std::string &url, bool isserver = false);
   };
 
 
   // =====================================================================================
 
   class InputPort : public Port {
 
      friend class Module;
      friend class ModuleAsync;
      friend class ModuleSync;
 
      private:
         InputPort(Reference parent,
                   const std::string &name,
                   unsigned queuesize);
 
      protected:
 
         virtual ~InputPort();
 
         virtual unsigned NumStartEvents();
 
         inline unsigned NumCanRecv() { return fQueue.Size(); }
 
         inline BufferSize_t TotalSizeCanRecv() { return fQueue.TotalBuffersSize(); }
 
         bool QueueFull() { return fQueue.Full(); }
 
         Buffer Item(unsigned indx) { return fQueue.Item(indx); }
 
         inline bool CanRecv() const { return fQueue.CanRecv(); }
 
         //Buffer Recv() { Buffer buf; fQueue.Recv(buf); fRate.SetDouble(buf.GetTotalSize()/1024./1024.); return buf; }
 
         Buffer Recv();
 
         void SignalWhenFull() { fQueue.SignalWhenFull(); }
 
         bool SkipBuffers(unsigned cnt=1);
 
      public:
 
         virtual const char* ClassName() const { return "InputPort"; }
 
         virtual bool IsInput() const { return true; }
   };
 
 
   // =======================================================================================
 
   class OutputPort : public Port {
 
      friend class Module;
      friend class ModuleAsync;
      friend class ModuleSync;
 
      private:
 
         bool   fSendallFlag; // flag, used by SendToAllOutputs to mark, which output must be used for sending
 
         OutputPort(Reference parent,
                   const std::string &name,
                   unsigned queuesize);
 
      protected:
 
         virtual ~OutputPort();
 
         virtual unsigned NumStartEvents();
 
         unsigned NumCanSend() const { return fQueue.NumCanSend(); }
 
         bool CanSend() const { return fQueue.CanSend(); }
 
//         bool Send(dabc::Buffer& buf) { fRate.SetDouble(buf.GetTotalSize()/1024./1024.); return fQueue.Send(buf); }
 
         bool Send(dabc::Buffer& buf);
 
      public:
 
         virtual const char* ClassName() const { return "OutputPort"; }
 
         virtual bool IsOutput() const { return true; }
 
   };
 
   // =======================================================================================
 
   class PoolHandle : public Port {
 
      friend class Module;
      friend class ModuleAsync;
      friend class ModuleSync;
 
      private:
 
         Reference     fPool;
 
         PoolHandle(Reference parent,
                    Reference pool,
                    const std::string &name,
                    unsigned queuesize);
 
      protected:
 
 
         virtual ~PoolHandle();
 
         virtual void DoCleanup()
         {
            fPool.Release();
            Port::DoCleanup();
         }
 
         virtual void ObjectCleanup()
         {
            fPool.Release();
            Port::ObjectCleanup();
         }
 
         virtual unsigned NumStartEvents();
 
         // inline MemoryPool* Pool() const { return fPool(); }
 
         inline unsigned NumCanTake() const { return fQueue.Size(); }
 
         inline Buffer Item(unsigned indx) { return fQueue.Item(indx); }
 
         inline bool CanTakeBuffer() const
         {
            return (QueueCapacity()==0) ? true : fQueue.CanRecv();
         }
 
         Buffer TakeBuffer(BufferSize_t size = 0);
 
         unsigned NumRequestedBuffer() const
         {
            return fQueue.Size();
         }
 
         Buffer TakeRequestedBuffer()
         {
            Buffer buf; fQueue.Recv(buf); return buf;
         }
 
      public:
 
         virtual const char* ClassName() const { return "PoolHandle"; }
 
         virtual bool IsInput() const { return true; }
   };
 
}
 
#endif