BufferImplementation.h

#ifndef _ots_BufferImplementation_h_
#define _ots_BufferImplementation_h_
 
#include "otsdaq/DataManager/CircularBufferBase.h"
#include "otsdaq/Macros/CoutMacros.h"
#include "otsdaq/Macros/StringMacros.h"
#include "otsdaq/MessageFacility/MessageFacility.h"
 
#include <atomic>
#include <iostream>
#include <map>
#include <string>
#include <vector>
 
namespace ots
{
template<class D, class H>
class BufferImplementation
{
    struct ConsumerStruct
    {
        ConsumerStruct()
            : priority_(CircularBufferBase::LowConsumerPriority)
            , readPointer_(0)
            , subBuffersStatus_(nullptr)
        {
        }
 
        CircularBufferBase::ConsumerPriority priority_;
        int                                  readPointer_;
        std::atomic_bool* subBuffersStatus_;  
    };
 
  public:
    BufferImplementation(const std::string& producerName       = "",
                         unsigned int       numberOfSubBuffers = 100);
    BufferImplementation(const BufferImplementation<D, H>& toCopy);
    BufferImplementation<D, H>& operator=(const BufferImplementation<D, H>& toCopy);
    virtual ~BufferImplementation(void);
 
    void init(void);
    void reset(void);
    void resetConsumerList(void);
    void registerConsumer(const std::string&                   name,
                          CircularBufferBase::ConsumerPriority priority);
    // void unregisterConsumer              (const std::string& name);
    int attachToEmptySubBuffer(D*& data, H*& header);
    int setWrittenSubBuffer(void);
    int write(const D& buffer, const H& header = H());
    int read(D& buffer, const std::string& consumer);
    int read(D& buffer, H& header, const std::string& consumer);
    int read(D*& buffer, H*& header, const std::string& consumer);
    int setReadSubBuffer(const std::string& consumer);  
 
    bool         isEmpty(void) const;
    unsigned int bufferSize(void) const { return numberOfSubBuffers_; }
    unsigned int numberOfWrittenBuffers(void) const;
 
    const std::map<std::string, ConsumerStruct>& getConsumers(void) const
    {
        return consumers_;
    };
 
    void dumpStatus(std::ostream* out = (std::ostream*)&(std::cout)) const;
 
  protected:
    const std::string mfSubject_;
 
  private:
    enum
    {
        ErrorBufferFull          = -1,
        ErrorBufferLocked        = -2,
        ErrorBufferNotAvailable  = -3,
        ErrorReadBufferOutOfSync = -4
    };
 
    const std::string producerName_;
    unsigned int      numberOfSubBuffers_;
    std::map<std::string, ConsumerStruct>
        consumers_;     
    int writePointer_;  
    std::atomic_bool* subBuffersStatus_;  
    std::vector<H>    headers_;           
    std::vector<D>    subBuffers_;        
    const bool        bufferFree_;
 
    unsigned int nextWritePointer(void);
    unsigned int nextReadPointer(const std::string& consumer);
    int          getFreeBufferIndex(void);  
    unsigned int getReadPointer(const std::string& consumer);
    void         setWritten(unsigned int subBuffer);
    void         setFree(unsigned int subBuffer, const std::string& consumer);
    std::atomic_bool& isFree(unsigned int subBuffer) const;
    std::atomic_bool& isFree(unsigned int subBuffer, const std::string& consumer) const;
 
    H&   getHeader(unsigned int subBuffer);
    D&   getSubBuffer(unsigned int subBuffer);
    void writeSubBuffer(unsigned int subBuffer, const D& buffer, const H& header);
};
#include "otsdaq/DataManager/BufferImplementation.icc"
 
}  // namespace ots
#endif