observer.hpp

#pragma once
 
#include "shared/constants.h"
#include "software/multithreading/thread_safe_buffer.hpp"
 
template <typename T, typename Clock = std::chrono::steady_clock>
class Observer
{
   public:
    Observer(size_t buffer_size = DEFAULT_BUFFER_SIZE, bool log_buffer_full = true);
 
    virtual void receiveValue(T val);
 
    virtual double getDataReceivedPerSecond() final;
 
    virtual ~Observer() = default;
 
    static constexpr size_t TIME_BUFFER_SIZE = 5;
 
   protected:
    virtual std::optional<T> popMostRecentlyReceivedValue(Duration max_wait_time) final;
 
    virtual std::optional<T> popLeastRecentlyReceivedValue(Duration max_wait_time) final;
 
    static constexpr size_t DEFAULT_BUFFER_SIZE = 1;
 
   private:
    ThreadSafeBuffer<T> buffer;
    boost::circular_buffer<std::chrono::milliseconds> receive_time_buffer;
};
 
template <typename T, typename Clock>
Observer<T, Clock>::Observer(size_t buffer_size, bool log_buffer_full)
    : buffer(buffer_size, log_buffer_full), receive_time_buffer(TIME_BUFFER_SIZE)
{
}
 
template <typename T, typename Clock>
void Observer<T, Clock>::receiveValue(T val)
{
    receive_time_buffer.push_back(std::chrono::duration_cast<std::chrono::milliseconds>(
        Clock::now().time_since_epoch()));
    buffer.push(std::move(val));
}
 
template <typename T, typename Clock>
std::optional<T> Observer<T, Clock>::popMostRecentlyReceivedValue(Duration max_wait_time)
{
    return buffer.popMostRecentlyAddedValue(max_wait_time);
}
 
template <typename T, typename Clock>
std::optional<T> Observer<T, Clock>::popLeastRecentlyReceivedValue(Duration max_wait_time)
{
    return buffer.popLeastRecentlyAddedValue(max_wait_time);
}
 
template <typename T, typename Clock>
double Observer<T, Clock>::getDataReceivedPerSecond()
{
    if (receive_time_buffer.empty())
    {
        return 0;
    }
    else
    {
        double time_s =
            static_cast<double>(
                (receive_time_buffer.back() - receive_time_buffer.front()).count()) *
            SECONDS_PER_MILLISECOND;
 
        double rate = static_cast<double>(receive_time_buffer.size() - 1) / time_s;
        return std::max(0.0, rate);
    }
}