133 lines
3.0 KiB
C++
133 lines
3.0 KiB
C++
#pragma once
|
|
#ifndef _NB_DATASINK
|
|
#define _NB_DATASINK
|
|
|
|
#include <thread>
|
|
|
|
#include "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 {
|
|
return _running;
|
|
}
|
|
virtual bool stop() noexcept = 0;
|
|
virtual bool run() = 0;
|
|
virtual bool in(const DataType&) = 0;
|
|
|
|
protected:
|
|
DataSink() = default;
|
|
|
|
std::atomic<bool> _running;
|
|
};
|
|
|
|
template<typename DataType, typename BufferType, typename ProcessorType>
|
|
class BufferedDataProcessor : public DataSink<DataType> {
|
|
using Base = DataSink<DataType>;
|
|
public:
|
|
using Base::Base;
|
|
|
|
bool stop() noexcept { return type_ptr->stop(); }
|
|
bool run() { return type_ptr->run(); }
|
|
bool in(const DataType& val) { return type_ptr->in(val); }
|
|
|
|
protected:
|
|
unsigned int count() const {
|
|
return type_ptr->count();
|
|
}
|
|
void push(const DataType& val) {
|
|
type_ptr->push(val);
|
|
}
|
|
DataType pop() {
|
|
return type_ptr->pop();
|
|
}
|
|
void flush() {
|
|
type_ptr->flush();
|
|
}
|
|
void clear() {
|
|
type_ptr->clear();
|
|
}
|
|
bool process(const DataType& val) {
|
|
return type_ptr->process(val);
|
|
}
|
|
|
|
using Base::_running;
|
|
|
|
BufferType _buffer;
|
|
|
|
private:
|
|
ProcessorType* const type_ptr = static_cast<ProcessorType*>(this);
|
|
|
|
};
|
|
|
|
template<typename DataType, typename ProcessorType>
|
|
class MultithreadedDataProcessor
|
|
: public BufferedDataProcessor<DataType, ThreadsafeQueue<DataType>, ProcessorType> {
|
|
using Base = BufferedDataProcessor<DataType, ThreadsafeQueue<DataType>, ProcessorType>;
|
|
public:
|
|
~MultithreadedDataProcessor() { type_ptr->stop(); }
|
|
|
|
bool isRunning() const noexcept override {
|
|
return this->_running && (_runningThread!=nullptr);
|
|
}
|
|
bool run() {
|
|
if (!type_ptr->isRunning()) {
|
|
this->_running = true;
|
|
_runningThread = std::make_shared<std::thread>([&]{
|
|
while(type_ptr->isRunning()) {
|
|
flush();
|
|
}
|
|
});
|
|
}
|
|
return this->isRunning();
|
|
}
|
|
bool stop() noexcept override {
|
|
if (type_ptr->isRunning()) {
|
|
this->_running = false;
|
|
_runningThread->join();
|
|
_runningThread = nullptr;
|
|
}
|
|
return !type_ptr->isRunning();
|
|
}
|
|
|
|
protected:
|
|
using Base::Base;
|
|
|
|
unsigned int count() const {
|
|
return this->_buffer.size();
|
|
}
|
|
void push(const DataType& val) {
|
|
this->_buffer.push(val);
|
|
};
|
|
DataType pop() {
|
|
std::shared_ptr<DataType> event;
|
|
this->_buffer.pop(event);
|
|
this->process(*event);
|
|
return *event;
|
|
}
|
|
void flush() {
|
|
while(type_ptr->count()) {
|
|
pop();
|
|
}
|
|
}
|
|
void clear() {
|
|
this->_buffer.empty();
|
|
}
|
|
|
|
std::shared_ptr<std::thread> _runningThread;
|
|
|
|
private:
|
|
ProcessorType* type_ptr = static_cast<ProcessorType*>(this);
|
|
|
|
};
|
|
|
|
|
|
} // namespace nb
|
|
#endif // _NB_DATASINK
|