NBEngine/engine/NBCore/include/NBCore/DataSink.hpp

#pragma once
#ifndef _NB_DATASINK
#define _NB_DATASINK

#include <atomic>
#include <thread>
#include <utility>

#include <NBCore/ThreadSafeQueue.hpp>

namespace nb {

template<typename DataType>
class DataSink {
    public:
    DataSink(const DataSink&) = delete;
    DataSink(DataSink&&) = delete;
    DataSink& operator=(const DataSink&) = delete;

    virtual bool isRunning() const noexcept {
        std::atomic_thread_fence(std::memory_order_acquire);
        return _running.load(std::memory_order_acquire);
    }
    virtual bool stop() noexcept = 0;
    virtual bool run() = 0;
    virtual bool in(const DataType&) = 0;
    
    protected:
    DataSink() {
        _running.store(false, std::memory_order_release);
    }

    std::atomic<bool> _running;
};

template<typename DataType, typename SinkTypes=DataSink<DataType>>
class MultiSink : public DataSink<DataType> {
    protected:
    using Base = DataSink<DataType>;
    using SinkPtr = std::shared_ptr<SinkTypes>;
    using Base::_running;
    std::vector<SinkPtr> _sinks;

    public:
    MultiSink(std::vector<SinkPtr> sinks={}) : _sinks(sinks) {}
    virtual void addSink(SinkPtr sink) {
        _sinks.push_back(sink);
    }
    virtual std::vector<SinkPtr>& getSinks() { return _sinks; }
    bool isRunning() const noexcept override {
        return Base::isRunning();
    }
    bool stop() noexcept override {
        _running.store(false, std::memory_order_release);
        for (auto& sink : _sinks) {
            sink->stop();
        }
        return isRunning();
    }
    bool run() override {
        _running.store(true, std::memory_order_release);
        for (auto& sink : _sinks) {
            sink->run();
        }
        return isRunning();
    }
    bool in(const DataType& data) override {
        if (isRunning()) {
            bool success = true;
            for (auto& sink : _sinks) {
                success &= sink->in(data);
            }
            return success;
        }
        return false;
    }
};

template<typename DataType, typename BufferType, typename ProcessorType>
class BufferedDataProcessor : public DataSink<DataType> {
    private:
    ProcessorType* const type_ptr = static_cast<ProcessorType*>(this);

    protected:
    using Base = DataSink<DataType>;
    using Base::_running;
    BufferType _buffer;
    virtual unsigned int count() const {
        return type_ptr->count();
    }
    virtual bool pop(std::shared_ptr<DataType> ret) {
        return type_ptr->pop(ret);
    }
    virtual void flush() {
        type_ptr->flush();
    }
    virtual void clear() {
        type_ptr->clear();
    }
    virtual bool process(const DataType& val) = 0;

    public:
    using Base::Base;
    virtual bool stop() noexcept override { return type_ptr->stop(); }
    virtual bool run() override { return type_ptr->run(); }
    virtual bool in(const DataType& val) override { return type_ptr->in(val); }

};

template<typename DataType, typename ProcessorType>
class MultithreadedDataProcessor 
: public BufferedDataProcessor<DataType, ThreadsafeQueue<DataType>, ProcessorType> {
    private:
    ProcessorType* const type_ptr = static_cast<ProcessorType*>(this);
    std::mutex _pause;

    protected:
    using Base = BufferedDataProcessor<DataType, ThreadsafeQueue<DataType>, ProcessorType>;
    using Base::Base;
    using Base::process;
    using Base::_running;
    std::shared_ptr<std::thread> _runningThread;
    virtual unsigned int count() const override {
        return type_ptr->_buffer.size();
    }
    virtual bool pop(std::shared_ptr<DataType> ret=nullptr) override {
        type_ptr->_buffer.pop(ret);
        return this->process(*ret);
    }
    virtual bool popBlock(std::shared_ptr<DataType> ret=nullptr) {
        ret = type_ptr->_buffer.popBlock();
        return this->process(*ret);
    }
    virtual void flush() override {
        while(type_ptr->count()) {
            type_ptr->pop();
        }
    }
    virtual void clear() override {
        type_ptr->_buffer.empty();
    }
    virtual std::unique_lock<std::mutex> pause() {
        return std::move(std::unique_lock<std::mutex>(_pause));
    }

    public:
    using Base::isRunning;
    virtual ~MultithreadedDataProcessor() { type_ptr->stop(); }

    bool run() override {
        if (!type_ptr->isRunning()) {
            _running.store(true, std::memory_order_release);
            _runningThread = std::make_shared<std::thread>([&]{
                while(type_ptr->isRunning()) {
                    std::lock_guard<std::mutex> lock(_pause);
                    type_ptr->flush();
                }
                type_ptr->flush();
                type_ptr->stop();
            });
        }
        return type_ptr->isRunning();
    }
    bool stop() noexcept override {
        if (type_ptr->isRunning()) {
            _running.store(false, std::memory_order_release);
            if (_runningThread) {
                _runningThread->join();
                _runningThread = nullptr;
            }
            type_ptr->flush();
        }
        return !type_ptr->isRunning();
    }
    bool in(const DataType& val) override {
        type_ptr->_buffer.push(val);
        return type_ptr->isRunning();
    }

};


} // namespace nb
#endif // _NB_DATASINK