observablePublisher.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2020-2025 European Southern Observatory (ESO)
//
// SPDX-License-Identifier: LGPL-3.0-only
 
#ifndef HLCC_OLDBMUX_CIIPUBLISHER_HPP_
#define HLCC_OLDBMUX_CIIPUBLISHER_HPP_
 
#include <string>
#include <functional>
#include <memory>
#include <variant>
 
#include <mal/Cii.hpp>
#include <mal/utility/LoadMal.hpp>
#include <mal/ps/qos/QoS.hpp>
#include <mal/ps/Publisher.hpp>
#include <mal/utility/Uri.hpp>
 
#include <rad/oldbInterface.hpp>
#include <taiclock/taiClock.hpp>
 
#include <ciiLogManager.hpp>
 
namespace hlcc::oldbmux {
 
struct PublishingTimeElapsed {
    std::chrono::nanoseconds malpub_elapsed;
    std::chrono::nanoseconds observers_elapsed;
};
 
template<typename T>
class ObservablePublisher {
 
using ObsCallback              = std::function<void (const T&)>;
using ObsCallbackWithTimestamp = std::function<void (const T&, const taiclock::TaiClock::time_point&)>;
using ObsCallbackWithDpQuality = std::function<void (const T&, const elt::oldb::CiiOldbDpQuality&)>;
using ObsCallbackWithTimestampAndDpQuality =
    std::function<void (const T&, const taiclock::TaiClock::time_point&, const elt::oldb::CiiOldbDpQuality&)>;
using ObsVariant = std::variant<ObsCallback, ObsCallbackWithTimestamp, ObsCallbackWithDpQuality, ObsCallbackWithTimestampAndDpQuality>;
 
public:
 
    explicit ObservablePublisher(std::unique_ptr<::elt::mal::ps::Publisher<T>> malpub, bool use_malpub = true);
 
    // Accepts the callback functions without user timestamp (internal timestamping)
    void AddObserver(ObsCallback observer_function);
 
    // Accepts the callback functions with user-provided timestamp parameter
    void AddObserver(ObsCallbackWithTimestamp observer_function);
 
    // Accepts the callback functions with Oldb datapoint quality
    void AddObserver(ObsCallbackWithDpQuality observer_function);
 
    // Accepts the callback functions with user-provided timestamp parameter and Oldb datapoint quality
    void AddObserver(ObsCallbackWithTimestampAndDpQuality observer_function);
 
    std::shared_ptr<T> createDataEntity();
 
    PublishingTimeElapsed Publish (const T& data,
                                   const std::chrono::milliseconds& timeout,
                                   const taiclock::TaiClock::time_point& timestamp = taiclock::TaiClock::now(),
                                   const elt::oldb::CiiOldbDpQuality& dp_quality = elt::oldb::CiiOldbDpQuality::OK) const;
 
private:
    const log4cplus::Logger logger = elt::log::CiiLogManager::GetLogger();
    const std::unique_ptr<::elt::mal::ps::Publisher<T>> m_malpub;
    const bool m_use_malpub;
    // Vector that will hold the observer functions in a variant due to both possibilities
    // with and without timestamp parameter
    std::vector<ObsVariant> m_observers;
};
 
template<typename T>
ObservablePublisher<T>::ObservablePublisher(std::unique_ptr<::elt::mal::ps::Publisher<T>> malpub, bool use_malpub)
: m_malpub {std::move(malpub)}, m_use_malpub {use_malpub} {
}
 
template<typename T>
void ObservablePublisher<T>::AddObserver(ObsCallback observer_function) {
    m_observers.push_back(observer_function);
}
 
template<typename T>
void ObservablePublisher<T>::AddObserver(ObsCallbackWithTimestamp observer_function) {
    m_observers.push_back(observer_function);
}
 
template<typename T>
void ObservablePublisher<T>::AddObserver(ObsCallbackWithDpQuality observer_function) {
    m_observers.push_back(observer_function);
}
 
template<typename T>
void ObservablePublisher<T>::AddObserver(ObsCallbackWithTimestampAndDpQuality observer_function) {
    m_observers.push_back(observer_function);
}
 
template<typename T>
std::shared_ptr<T> ObservablePublisher<T>::createDataEntity() {
    return m_malpub->createDataEntity();
}
 
template<typename T>
PublishingTimeElapsed ObservablePublisher<T>::Publish(const T& data,
                                                      const std::chrono::milliseconds& timeout,
                                                      const taiclock::TaiClock::time_point& timestamp,
                                                      const elt::oldb::CiiOldbDpQuality& dp_quality) const {
    auto t_0 = std::chrono::steady_clock::now();
    if (m_malpub != nullptr && m_use_malpub) {
       logger.log(log4cplus::DEBUG_LOG_LEVEL, "publishing over MAL");
       try {
           m_malpub->publish(data, timeout);
       } catch (std::exception& exc) {
          std::string cause = exc.what();
          logger.log(log4cplus::ERROR_LOG_LEVEL, "Ignoring: publishing to MAL failed, due to: " + cause);
       }
    }
 
    auto t_1 = std::chrono::steady_clock::now();
    logger.log(log4cplus::DEBUG_LOG_LEVEL, "notifying observers");
    for (ObsVariant obs_func : m_observers) {
        try {
            // Check if variant contains callback function without timestamp
            if (std::holds_alternative<ObsCallback>(obs_func)) {
                auto func = std::get<ObsCallback>(obs_func);
                func(data);
            }
            // Check if variant contains callback function with timestamp
            else if (std::holds_alternative<ObsCallbackWithTimestamp>(obs_func)) {
                auto func = std::get<ObsCallbackWithTimestamp>(obs_func);
                func(data, timestamp);
            }
            // Check if variant contains callback function with Oldb dp quality
            else if (std::holds_alternative<ObsCallbackWithDpQuality>(obs_func)) {
                auto func = std::get<ObsCallbackWithDpQuality>(obs_func);
                func(data, dp_quality);
            }
            // Check if variant contains callback function with timestamp & Oldb dp quality
            else if (std::holds_alternative<ObsCallbackWithTimestampAndDpQuality>(obs_func)) {
                auto func = std::get<ObsCallbackWithTimestampAndDpQuality>(obs_func);
                func(data, timestamp, dp_quality);
            }
            // If variant is empty (should not be possible) log an error
            else {
                logger.log(log4cplus::ERROR_LOG_LEVEL, "Observable function variant holds NO valid"
                                                       "alternative.");
            }
        } catch (std::exception& exc) {
           std::string cause = exc.what();
           logger.log(log4cplus::DEBUG_LOG_LEVEL, "Ignoring: callback to observer failed, due to: " + cause);
        }
    }
    logger.log(log4cplus::DEBUG_LOG_LEVEL, "publish done");
 
    return PublishingTimeElapsed {(t_1 - t_0), (std::chrono::steady_clock::now() - t_1)};
}
 
}
// namespace
 
#endif