DataIO.h

Go to the documentation of this file.

// $Id: DataIO.h 4575 2020-08-04 12:44:29Z 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_DataIO
#define DABC_DataIO
 
#ifndef DABC_Worker
#include "dabc/Worker.h"
#endif
 
#ifndef DABC_Url
#include "dabc/Url.h"
#endif
 
namespace dabc {
 
   class Buffer;
   class InputTransport;
 
   enum DataInputCodes {
      di_ValidSize     = 0xFFFFFFF0,   // last valid size for buffer
      di_None          = 0xFFFFFFF1,   // invalid return code
      di_Repeat        = 0xFFFFFFF2,   // no data in input, try as soon as possible
      di_RepeatTimeOut = 0xFFFFFFF3,   // no data in input, try with timeout
      di_EndOfStream   = 0xFFFFFFF4,   // no more data in input is expected, object can be destroyed
      di_Ok            = 0xFFFFFFF5,   // normal code
      di_CallBack      = 0xFFFFFFF6,   // data source want to work via callback
      di_Error         = 0xFFFFFFF7,   // error
      di_SkipBuffer    = 0xFFFFFFF8,   // when doing complete, buffer cannot be filled
      di_DfltBufSize   = 0xFFFFFFF9,   // default buffer size means that any size provided by memory pool will be accepted
      di_NeedMoreBuf   = 0xFFFFFFFA,   // when input needs more buffer for it work, DataInput can initiate callback at any time
      di_HasEnoughBuf  = 0xFFFFFFFB,   // input indicates that enough buffers are accumulated and transport should wait for callback
      di_MoreBufReady  = 0xFFFFFFFC,   // when input has more bufs ready for complete
      di_QueueBufReady = 0xFFFFFFFD    // when data input can provide next buffer
   };
 
   class DataInput {
      friend class InputTransport;
 
      protected:
 
         virtual WorkerAddon* Read_GetAddon() { return 0; }
 
      public:
 
         virtual ~DataInput() {}
 
         virtual bool Read_Init(const WorkerRef& wrk, const Command& cmd) { return true; }
 
 
         virtual unsigned Read_Size() { return di_EndOfStream; }
 
         virtual unsigned Read_Start(Buffer& buf) { return di_Ok; }
 
         virtual unsigned Read_Complete(Buffer& buf) { return di_EndOfStream; }
 
         virtual double Read_Timeout() { return 0.1; }
 
 
         virtual bool Read_Stat(dabc::Command cmd) { return false; }
 
         Buffer ReadBuffer();
 
   };
 
   // _________________________________________________________________
 
   enum DataOutputCodes {
       do_Ok,               // operation is ok
       do_Skip,             // if returned, buffer must be skipped
       do_Repeat,           // repeat operation as soon as possible
       do_RepeatTimeOut,    // repeat operation after timeout
       do_CallBack,         // operation will be called back
       do_Error,            // operation error, object must be cleaned
       do_Close             // output is closed, one should destroy it
   };
 
 
   class DataOutput {
 
      friend class OutputTransport;
 
      protected:
         std::string         fInfoName;     // parameter name for info settings
 
         DataOutput(const dabc::Url& url);
 
         virtual WorkerAddon* Write_GetAddon() { return 0; }
 
         void ShowInfo(int lvl, const std::string &info);
 
      public:
 
         virtual ~DataOutput() {}
 
         void SetInfoParName(const std::string &name);
 
         virtual std::string ProvideInfo() { return std::string(); }
 
         virtual bool Write_Init() { return true; }
 
         virtual unsigned Write_Check() { return do_Ok; }
 
         virtual unsigned Write_Buffer(Buffer& buf) { return do_Ok; }
 
         virtual unsigned Write_Complete() { return do_Ok; }
 
         virtual double Write_Timeout() { return 0.1; }
 
         virtual void Write_Flush() {}
 
         bool WriteBuffer(Buffer& buf);
 
         virtual bool Write_Stat(dabc::Command cmd) { return false; }
 
         virtual bool Write_Retry() { return false; }
 
         virtual bool Write_Restart(dabc::Command cmd) { return false; }
 
   };
 
   // ===========================================================
 
   class FileInterface;
 
   class FileInput : public DataInput {
      protected:
         std::string          fFileName;
         dabc::Reference      fFilesList;
         dabc::FileInterface* fIO;
         std::string          fCurrentName;
         bool                 fLoop; 
         double               fReduce; 
 
         bool InitFilesList();
         bool TakeNextFileName();
         const std::string &CurrentFileName() const { return fCurrentName; }
         void ClearCurrentFileName() { fCurrentName.clear(); }
 
         FileInput(const dabc::Url& url);
 
      public:
 
         virtual ~FileInput();
 
         void SetIO(dabc::FileInterface* io);
 
         virtual bool Read_Init(const WorkerRef& wrk, const Command& cmd);
 
         virtual bool Read_Stat(dabc::Command cmd);
   };
 
   // ============================================================================
 
   class FileOutput : public DataOutput {
      protected:
 
         std::string          fFileName;
         int                  fSizeLimitMB;
         std::string          fFileExtens;
 
         dabc::FileInterface* fIO;
 
         int                  fCurrentFileNumber;
         std::string          fCurrentFileName;
         long                 fCurrentFileSize;
 
         long                 fTotalFileSize;
         long                 fTotalNumBufs;
         long                 fTotalNumEvents;
 
         void ProduceNewFileName();
         const std::string &CurrentFileName() const { return fCurrentFileName; }
 
         bool CheckBufferForNextFile(unsigned sz);
 
         void AccountBuffer(unsigned sz, int numev = 0);
 
         FileOutput(const dabc::Url& url, const std::string &ext = "");
 
         std::string ProduceFileName(const std::string &suffix);
 
         int GetSizeLimitMB() const { return fSizeLimitMB; }
 
      public:
 
         virtual ~FileOutput();
 
         void SetIO(dabc::FileInterface* io);
 
         virtual std::string ProvideInfo();
 
         virtual bool Write_Init();
 
         virtual bool Write_Stat(dabc::Command cmd);
   };
 
}
 
#endif