hbufferedwriter.h

#ifndef __HBUFFEREDWRITER_H
#define __HBUFFEREDWRITER_H
 
#include <thread>
#include <mutex>
#include <condition_variable>
 
#include "hwriter.h"
#include "hwriterconsumer.h"
 
#define DEFAULT_BLOCKS_RESERVED 1000
 
template <class T>
class HBufferedWriter : public HWriter<T>, public HWriterConsumer<T>
{
    private:
 
        size_t _blocksize;
        int _blocksReserved;
        HWriter<T>* _writer;
        std::vector<T*> _buffer;
        int _blocks;
 
        bool _enabled;
 
        std::thread* _drain;
        bool _isDraining;
 
        std::mutex _drainMutex;
        std::condition_variable _drainLock;
 
        std::mutex _readWriteMutex;
 
    public:
 
        HBufferedWriter(HWriterConsumer<T>* consumer, size_t blocksize, int reserved = DEFAULT_BLOCKS_RESERVED, bool enabled = true):
                _blocksize(blocksize),
                _blocksReserved(reserved),
                _blocks(reserved),
                _isDraining(false),
                _enabled(enabled)
        {
            Init();
 
            consumer->SetWriter(this->Writer());
        }
 
        HBufferedWriter(HWriter<T>* writer, size_t blocksize, int reserved = DEFAULT_BLOCKS_RESERVED, bool enabled = true):
                _blocksize(blocksize),
                _blocksReserved(reserved),
                _blocks(reserved),
                _isDraining(false),
                _enabled(enabled),
                _writer(writer)
        {
            Init();
        }
 
        ~HBufferedWriter()
        {
            StopDrain();
 
            HLog("Deleting remaining items from the buffer");
            while( _buffer.size() > 0 ) {
                delete _buffer.at(0);
                _buffer.erase( _buffer.begin(), _buffer.begin() + 1 );
            }
            HLog("Reducing the size of the buffer");
            _buffer.shrink_to_fit();
            HLog("Cleanup completed, ready to delete this object");
        }
 
        int Write(T* src, size_t blocksize);
 
        bool Start()
        {
            HLog("Start() called");
            if( _isDraining ) {
                throw new HWriterIOException("Drain thread is running. Writer must not be started again without being stopped first");
            }
 
            HLog("Starting downstream writer");
            bool started = _writer->Start();
 
            HLog("Starting drain thread");
            StartDrain();
 
            HLog("Drain thread is running");
            return started;
        }
 
        bool Stop()
        {
            HLog("Stop() called");
            StopDrain();
 
            HLog("Drain thread is stopped, stopping downstream writer");
            return _writer->Stop();
        }
 
        bool Command(HCommand* command)
        {
            return _writer->Command(command);
        }
 
        void SetWriter(HWriter<T>* writer)
        {
            _writer = writer;
        }
 
        void Flush()
        {
            while( _isDraining && _buffer.size() > 0 ) {
                usleep(100);
            }
        }
 
        int Reserved()
        {
            return _blocks;
        }
 
        int Used()
        {
            return _buffer.size();
        }
 
        bool Enabled()
        {
            return _enabled;
        }
 
    private:
 
        void Init() {
            _buffer.reserve( _blocks * sizeof(T*) );
        }
 
        void StartDrain() {
 
            // Start a new drain thread
            HLog("Starting drain thread");
            _isDraining = true;
            _drain = new std::thread( [this]()  {
                while( _isDraining ) {
                    Drain();
                }
                HLog("Drain thread exiting");
            } );
            HLog("Drain thread is running");
        }
 
        void StopDrain() {
 
            // Stop a running drain thread
            if( _isDraining && _drain != NULL ) {
 
                HLog("Signal halt to drain thread");
                _isDraining = false;
 
                HLog("Awake a waiting drain thread");
                _drainLock.notify_one();
 
                HLog("Joining drain thread");
                _drain->join();
                _drain = NULL;
                HLog("Drain thread joined");
            }
        }
 
        void Drain();
};
 
#endif