qt6-migration-pipeline/external/tmsif/include/ddkit/access/Reader.hpp

#pragma once

#include "ddkit/access/DdsWrap.hpp"
#include "ddkit/access/Dispatcher.hpp"
#include "ddkit/access/Topic.hpp"
#include "ddkit/access/Listener.hpp"
#include "ddkit/access/ConnectState.hpp"

namespace ddkit {
namespace access {

#if defined(DDS_VENDOR_CONNEXT)
#define DDKIT_USE_KEY_VALUE ON
#ifdef _WIN32
#define DDKIT_COND_TYPE_UNUSED
#else
#define DDKIT_COND_TYPE_UNUSED dds::core::cond::Condition
#endif
#elif defined(DDS_VENDOR_CYCLONE)
#define DDKIT_COND_TYPE_UNUSED dds::core::cond::Condition
#elif defined(DDS_VENDOR_OPENSPLICE)
#define DDKIT_COND_TYPE_UNUSED dds::core::cond::Condition
#else
#error "DDS_VENDOR not set"
#endif


template <typename T>
class Reader final {
public:
    Reader(
        std::shared_ptr<access::Dispatcher> dispatcher,
        std::shared_ptr<Topic<T>> topic,
        dds::sub::qos::DataReaderQos dataReaderQos,
        dds::sub::qos::SubscriberQos subQos,
        std::function<void(T const&)> funcWrite,
        std::function<void(T const&)> funcDispose)
        : m_dispatcher(dispatcher)
        , m_connectState(std::make_shared<ConnectState>(topic->topicName(), topic->topicNameModule(), topic->topicNameTopic()))
        , m_listener(std::make_shared<ListenerRead<T> >(m_connectState))
        , m_ddsReader(dataReader(topic, m_listener.get(), dataReaderQos, subQos))
        , m_dataState(dds::sub::status::DataState::any())
        , m_shutdownRequest(false)
        , m_funcWrite(funcWrite)
        , m_funcDispose(funcDispose)
        , m_func([this](DDKIT_COND_TYPE_UNUSED) {

            DDKIT_LOG_TRACE;

            const dds::sub::LoanedSamples<T> samples = m_ddsReader.select().state(m_dataState).take();
            for (const auto& sample : samples) {
                if (m_shutdownRequest) return; // fast exit

                const auto& sampleInfo = sample.info();
#ifdef DDKIT_USE_KEY_VALUE
                const auto& instanceHandle = sampleInfo.instance_handle();
#endif
                const auto& dataState = sampleInfo.state();
                const auto& instanceState = dataState.instance_state();

                using namespace dds::sub::status;
                ddkit::log::trace() << "Sample alive:" << (instanceState == InstanceState::alive());
                ddkit::log::trace() << "Sample disposed:" << (instanceState == InstanceState::not_alive_disposed());
                ddkit::log::trace() << "Sample no_writers:" << (instanceState == InstanceState::not_alive_no_writers());

                // How to get partition of sending writer
                // only connext and cyclone > 0.10
                // auto pubData = dds::sub::matched_publication_data(m_ddsReader, sampleInfo.publication_handle());
                // for (auto& n : pubData.partition()->name()) {
                //     std::cout << "Partition: " << n << std::endl;
                // }

                if (sampleInfo.valid() && instanceState == InstanceState::alive()) {
                    const T& t = sample.data();

                    // callback
                    if (m_funcWrite) {
                        ddkit::log::debug() << "received: " << t << ". Callbacking user ...";
                        m_funcWrite(t);
                        ddkit::log::trace() << "... usercallback called.";
                    }
                } else {

                    if (m_funcDispose) {
#ifdef DDKIT_USE_KEY_VALUE
                        // - 'key_value()' works perfectly for connext
                        //   (and for cyclone+opensplice while explicitly disposing)
                        // - no (positive) effect on cross-vendor (see top of file)
                        // - so it is best to stay with native sample.data() for cyclone+opensplice but
                        //   use simple key_value for connext
                        T t;
                        m_ddsReader.key_value(t, instanceHandle);
#else
                        // - vendors, that are able to fetch the key even on dispose-calls via sample.data()
                        //   (they fill in default-ctor-data in the other attrs, because only the key has
                        //    to be transferred via DDS for dispose - same definition on our API - fine).
                        const T& t = sample.data();
#endif
                        ddkit::log::debug() << "received for disposing: " << t << ". Callbacking user ...";
                        m_funcDispose(t);
                        ddkit::log::trace() << "... usercallback called.";
                    }
                }
            }
        })
        , m_readCondition(m_ddsReader, m_dataState, m_func) {
        DDKIT_LOG_TRACE;
        m_dispatcher->addReadCondition(m_readCondition);
    }

    ~Reader() {
        DDKIT_LOG_TRACE;

        unload();
        m_ddsReader.close();
    }

    void unload() {
        if (!m_shutdownRequest) {
            m_shutdownRequest = true;
            m_dispatcher->removeReadCondition(m_readCondition);
        }
    }

    IConnectCheckPtr getConnectCheck() {
        return m_connectState;
    }

private:
    std::shared_ptr<Dispatcher> m_dispatcher;
    ConnectStatePtr m_connectState;
    std::shared_ptr<ListenerRead<T>> m_listener;
    dds::sub::DataReader<T> m_ddsReader;
    dds::sub::status::DataState m_dataState;

    std::atomic<bool> m_shutdownRequest;
    std::function<void(T const&)> m_funcWrite;
    std::function<void(T const&)> m_funcDispose;
    std::function<void(DDKIT_COND_TYPE_UNUSED)> m_func;
    dds::sub::cond::ReadCondition m_readCondition;

    /// Note on reader-listeners: see writer-listeners in Writer.hpp
    static dds::sub::DataReader<T> dataReader(
        std::shared_ptr<Topic<T>> topic,
        ListenerRead<T>* listener,
        dds::sub::qos::DataReaderQos dataReaderQos,
        dds::sub::qos::SubscriberQos subQos) {

        DDKIT_LOG_TRACE;

        dds::sub::DataReader<T> ddsReader = dds::sub::DataReader<T>(
            dds::sub::Subscriber(topic->domainParticipant(), subQos),
            topic->ddsTopic(),
            dataReaderQos,
            listener,
            dds::core::status::StatusMask::all());

        return ddsReader;
    }

    // no copying
    Reader(Reader const&) = delete;
    Reader(Reader&&)      = delete;
    void operator=(Reader const&) = delete;
};

#undef DDKIT_USE_KEY_VALUE
#undef DDKIT_COND_TYPE_UNUSED

} // namespace access
} // namespace ddkit