StreamProc.h

#ifndef BASE_STREAMPROC_H
#define BASE_STREAMPROC_H
 
#include <string>
 
#include "base/Processor.h"
 
namespace base {
 
   class Event;
 
   class StreamProc : public Processor {
      friend class ProcMgr;
 
      public:
 
         enum SyncKind {
            sync_None,         
            sync_Inter,        
            sync_Left,         
            sync_Right         
         };
 
 
      protected:
 
         typedef RecordsQueue<base::Buffer, false> BuffersQueue;
 
         typedef RecordsQueue<base::SyncMarker, false> SyncMarksQueue;
 
         BuffersQueue fQueue;                     
 
         unsigned        fQueueScanIndex;         
         unsigned        fQueueScanIndexTm;       
 
         AnalysisKind    fAnalysisKind;           
 
         SyncKind        fSynchronisationKind;    
         SyncMarksQueue  fSyncs;                  
         unsigned        fSyncScanIndex;          
         bool            fSyncFlag;               
 
         LocalMarkersQueue  fLocalMarks;          
         double          fTriggerAcceptMaring;    
         GlobalTime_t    fLastLocalTriggerTm;     
 
         GlobalMarksQueue fGlobalMarks;           
 
         unsigned fGlobalTrigScanIndex;           
         unsigned fGlobalTrigRightIndex;          
 
         bool fTimeSorting;                       
 
         base::H1handle fTriggerTm;  
         base::H1handle fMultipl;    
 
         base::C1handle fTriggerWindow;   
 
         static unsigned fMarksQueueCapacity;   
         static unsigned fBufsQueueCapacity;   
 
         StreamProc(const char* name = "", unsigned brdid = DummyBrdId, bool basehist = true);
 
         void SetSynchronisationKind(SyncKind kind = sync_Inter) { fSynchronisationKind = kind; }
 
         void AddSyncMarker(SyncMarker& marker);
 
         bool AddTriggerMarker(LocalTimeMarker& marker, double tm_range = 0.);
 
         GlobalTime_t LocalToGlobalTime(GlobalTime_t localtm, unsigned* sync_index = 0);
 
         bool IsSyncIndexWithInterpolation(unsigned indx) const
         { return (indx>0) && (indx<numReadySyncs()); }
 
         virtual bool doTriggerSelection() const { return false; }
 
         virtual GlobalTime_t ProvidePotentialFlushTime(GlobalTime_t last_marker);
 
         bool VerifyFlushTime(const base::GlobalTime_t& flush_time);
 
         virtual double MaximumDisorderTm() const { return 0.; }
 
         unsigned TestHitTime(const base::GlobalTime_t& hittime, bool normal_hit, bool can_close_event = true);
 
         // TODO: this is another place for future improvement
         // one can preallocate number of subevents with place ready for some messages
         // than one can use these events instead of creating them on the fly
 
         template<class EventClass, class MessageClass>
         void AddMessage(unsigned indx, EventClass* ev, const MessageClass& msg)
         {
            if (!ev) {
               ev = new EventClass;
               fGlobalMarks.item(indx).subev = ev;
            }
            ev->AddMsg(msg);
         }
 
         bool eraseSyncAt(unsigned indx);
 
         bool eraseFirstSyncs(unsigned sync_num);
 
      public:
 
         virtual ~StreamProc();
 
         void SetTimeSorting(bool on) { fTimeSorting = on; }
         bool IsTimeSorting() const { return fTimeSorting; }
 
         void SetTriggerMargin(double margin = 0.) { fTriggerAcceptMaring = margin; }
 
         void CreateTriggerHist(unsigned multipl = 40, unsigned nbins = 2500, double left = -1e-6, double right = 4e-6);
 
         virtual void SetTriggerWindow(double left, double right)
         { ChangeC1(fTriggerWindow, left, right); }
 
         void SetRawScanOnly() { fAnalysisKind = kind_RawOnly; }
         bool IsRawScanOnly() const { return fAnalysisKind == kind_RawOnly; }
 
         bool IsRawAnalysis() const { return fAnalysisKind <= kind_Raw; }
         bool IsTriggeredAnalysis() const { return fAnalysisKind == kind_Triggered; }
         bool IsStreamAnalysis() const { return fAnalysisKind == kind_Stream; }
 
         bool IsSynchronisationRequired() const { return fSynchronisationKind != sync_None; }
 
         unsigned minNumSyncRequired() const {
            switch (fSynchronisationKind) {
               case sync_None:  return 0;
               case sync_Inter: return 2;
               case sync_Left:  return 1;
               case sync_Right: return 1;
            }
            return 0;
         }
 
         virtual bool AddNextBuffer(const Buffer& buf);
 
         virtual bool ScanNewBuffers();
 
         virtual bool ScanNewBuffersTm();
 
         virtual void SkipAllData();
 
         virtual bool SkipBuffers(unsigned cnt);
 
         unsigned numSyncs() const { return fSyncs.size(); }
         unsigned numReadySyncs() const { return fSyncScanIndex; }
         SyncMarker& getSync(unsigned n) { return fSyncs.item(n); }
         unsigned findSyncWithId(unsigned syncid) const;
 
         virtual bool CollectTriggers(GlobalMarksQueue& queue);
 
         virtual bool DistributeTriggers(const GlobalMarksQueue& queue);
 
         virtual bool ScanDataForNewTriggers();
 
         unsigned NumReadySubevents() const { return fGlobalTrigScanIndex; }
 
         virtual bool AppendSubevent(base::Event* evt);
 
         virtual bool FirstBufferScan(const base::Buffer&) { return false; }
 
         virtual bool SecondBufferScan(const base::Buffer&) { return false; }
 
         virtual void Store(Event*) {}
 
         virtual void ResetStore() {}
 
         static void SetMarksQueueCapacity(unsigned sz) { fMarksQueueCapacity = sz; }
         static void SetBufsQueueCapacity(unsigned sz) { fBufsQueueCapacity = sz; }
 
   };
 
}
 
#endif