ciiOldbDataPointAsync.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2020-2025 European Southern Observatory (ESO)
//
// SPDX-License-Identifier: LGPL-3.0-only
 
/*
 * ciiOldbDataPointAsync.hpp
 *
 *  Created on: Jul 2, 2021
 *      Author: hsommer
 */
#define BOOST_THREAD_PROVIDES_FUTURE
#define BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION
#define BOOST_THREAD_PROVIDES_FUTURE_WHEN_ALL_WHEN_ANY // see https://www.javaer101.com/en/article/84361855.html
 
#ifndef HLCC_OLDBMUX_CIIOLDBDATAPOINTASYNC_HPP
#define HLCC_OLDBMUX_CIIOLDBDATAPOINTASYNC_HPP
 
#include <hlcc/oldbmux/circularBufferConcurrent.hpp>
 
#include "ciiOldbDataPoint.hpp"
#include "ciiOldbDpValue.hpp"
#include "ciiOldbDpQuality.hpp"
 
#include <log4cplus/logger.h>
#include <rad/logger.hpp>
 
#include <boost/thread/future.hpp>
#include <boost/asio/thread_pool.hpp>
#include <boost/asio/post.hpp>
 
#include <atomic>
#include <mutex>
 
 
namespace hlcc::oldbmux {

template<typename T>
class CiiOldbDataPointAsync {
 
public:
 
     // See declaration below.
    struct OldbData;
    struct OldbDataWithPromise;

    CiiOldbDataPointAsync(
            std::string name, std::shared_ptr<elt::oldb::CiiOldbDataPoint<T>> delegate,
            boost::asio::thread_pool& async_exec, const log4cplus::Logger& logger,
            std::size_t buffer_capacity = 1 );
 
    virtual ~CiiOldbDataPointAsync();
 
    std::shared_ptr<elt::oldb::CiiOldbDpValue<T>> ReadValue(
            bool check_bad_quality = true);
 
    boost::future<typename CiiOldbDataPointAsync<T>::OldbData> WriteValue(
            const T& value, 
            int64_t timestamp = elt::oldb::CiiOldbUtil::Now(), 
            elt::oldb::CiiOldbDpQuality quality = elt::oldb::CiiOldbDpQuality::OK,
            bool is_disable_publishing = false);
 
    boost::future<typename CiiOldbDataPointAsync<T>::OldbData> SetQuality(
            elt::oldb::CiiOldbDpQuality quality,
            bool is_disable_publishing = false);
 
private:
    
    boost::future<typename CiiOldbDataPointAsync<T>::OldbData> WriteAsync(
            OldbData&& new_data, 
            bool is_disable_publishing);

    void WriteBufferToOldb();
 

    std::condition_variable m_sync_dtor_cv;

    int m_sync_dtor_n_writing;

    std::mutex m_sync_dtor_mutex;
 
 
    const log4cplus::Logger logger;

    const std::string name;

    const std::shared_ptr<elt::oldb::CiiOldbDataPoint<T>> delegate;

    CircularBufferConcurrent<OldbDataWithPromise> buffer;

    boost::asio::thread_pool& async_exec;

    std::atomic<bool> is_async_processing;
};
 
 

template<typename T>
struct CiiOldbDataPointAsync<T>::OldbData {
 
public:

    OldbData(elt::oldb::CiiOldbDpQuality quality)
    : m_value {std::nullopt},
      m_timestamp {std::nullopt},
      m_quality {quality}
      { }

    OldbData(std::optional<T> value, std::optional<int64_t> timestamp, std::optional<elt::oldb::CiiOldbDpQuality> quality)
    : m_value {value},
      m_timestamp {timestamp},
      m_quality {quality}
      { }

    virtual ~OldbData() = default;
 
    OldbData(const OldbData&) = delete;
    OldbData& operator=(const OldbData&) = delete;
 
    OldbData(OldbData&&) = default;
    OldbData& operator=(OldbData&&) = default;
 
    std::optional<T> GetValue() const {
        return m_value;
    }
    void SetValue(T value) {
        m_value = value;
    }
    std::optional<int64_t> GetTimestamp() const {
        return m_timestamp;
    }
    void SetTimestamp(int64_t timestamp) {
        m_timestamp = timestamp;
    }
    std::optional<elt::oldb::CiiOldbDpQuality> GetQuality() const {
        return m_quality;
    }
private:
    std::optional<T> m_value;

    std::optional<int64_t> m_timestamp;

    std::optional<elt::oldb::CiiOldbDpQuality> m_quality;
};
 

template<typename T>
struct CiiOldbDataPointAsync<T>::OldbDataWithPromise : public OldbData {
 
public:
    OldbDataWithPromise(std::optional<T> value, std::optional<int64_t> timestamp, std::optional<elt::oldb::CiiOldbDpQuality> quality, boost::promise<OldbData>&& promise)
    : OldbData {value, timestamp, quality},
      m_promise {std::move(promise)}
      { }
 
    virtual ~OldbDataWithPromise() = default;
 
    OldbDataWithPromise(const OldbDataWithPromise&) = delete;
    OldbDataWithPromise& operator=(const OldbDataWithPromise&) = delete;
 
    // creates memberwise moves incl base class.
    OldbDataWithPromise(OldbDataWithPromise&& other) = default;
 
    OldbDataWithPromise& operator=(OldbDataWithPromise&& other) = default;
 
    boost::promise<CiiOldbDataPointAsync<T>::OldbData>* GetPromise(){
        return &m_promise;
    }
 
private:
    boost::promise<CiiOldbDataPointAsync<T>::OldbData> m_promise;
};
 
}  // namespace hlcc::oldbmux
 
 
// include template class definition from separate header file.
#include <hlcc/oldbmux/ciiOldbDataPointAsync.ipp>
 
#endif /* HLCC_OLDBMUX_CIIOLDBDATAPOINTASYNC_HPP */