signal_listening_intf.h

Go to the documentation of this file.

 
#pragma once
 
#include "signal_registry_intf.h"
 
#include <adtfbase/exception_handling.h>
#include <unordered_set>
#include <vector>
#include <chrono>
#include <optional>
 
namespace adtf
{
namespace services
{
namespace ant
{
 
class ISignalListening: public adtf::ucom::IObject
{
    public:
        ADTF_IID(ISignalListening, "signal_listening.ant.services.adtf.iid");
 
    protected:
        ~ISignalListening() = default;
 
    public:
        class ISignalsListener
        {
            public:
                virtual void SignalAdded(const ISignalRegistry::tSignalAttributes& sSignalAttributes) = 0;
 
                virtual void SignalRemoved(const ISignalRegistry::tSignalAttributes& sSignalAttributes) = 0;
        };
 
        class ISignalListener
        {
            public:
                virtual void SignalUpdated(ISignalRegistry::tSignalID nSignalID, const ISignalRegistry::tSignalValue& sValue) = 0;
        };
 
    public:
        virtual tResult RegisterSignalsListener(ISignalsListener& oListener) = 0;
 
        virtual tResult UnregisterSignalsListener(ISignalsListener& oListener) = 0;
 
        virtual tResult RequestSignalUpdates(ISignalRegistry::tSignalID nSignalID, ISignalListener& oListener) = 0;
 
        virtual tResult CancelSignalUpdates(ISignalRegistry::tSignalID nSignalID, ISignalListener& oListener) = 0;
};
 
}
 
namespace flash
{
 
class ISignalListening: public ant::ISignalListening
{
    public:
        ADTF_IID(ISignalListening, "signal_listening.flash.services.adtf.iid");
 
    protected:
        ~ISignalListening() = default;
 
    public:
        class ISignalListener
        {
            public:
                virtual void SignalUpdated(ISignalRegistry::tSignalID nSignalID, const adtf::services::flash::ISignalRegistry::tSignalValueNs& sValue)
                {
                    return SignalUpdated(nSignalID, adtf::services::ant::ISignalRegistry::tSignalValue{base::flash::duration_cast<tTimeStamp>(sValue.nTimeStamp),
                                         sValue.f64Value,
                                         sValue.strTextValue});
                }
                virtual void SignalUpdated(ISignalRegistry::tSignalID /*nSignalID*/, const adtf::services::ant::ISignalRegistry::tSignalValue& /*sValue*/)
                {}
        };
 
    public:
        using ant::ISignalListening::RequestSignalUpdates;
        using ant::ISignalListening::CancelSignalUpdates;
 
        virtual tResult RequestSignalUpdates(ISignalRegistry::tSignalID nSignalID, ISignalListener& oListener) = 0;
 
        virtual tResult CancelSignalUpdates(ISignalRegistry::tSignalID nSignalID, ISignalListener& oListener) = 0;
};
 
}
 
using flash::ISignalListening;
 
namespace testing
{
 
namespace ant
{
 
struct cTestSignalValueNs : public ISignalRegistry::tSignalValueNs
{
    cTestSignalValueNs() = default;
    cTestSignalValueNs(const cTestSignalValueNs&) = default;
    cTestSignalValueNs& operator=(const cTestSignalValueNs&) = default;
    cTestSignalValueNs(cTestSignalValueNs&&) = default;
    cTestSignalValueNs& operator=(cTestSignalValueNs&&) = default;
    cTestSignalValueNs(const ISignalRegistry::tSignalValueNs& oSignalValue):
        ISignalRegistry::tSignalValueNs(oSignalValue)
    {
        if (oSignalValue.strTextValue)
        {
            m_strTextValueCopy = oSignalValue.strTextValue;
        }
    }
    std::string m_strTextValueCopy;
};
 
class cTestListener: public ISignalListening::ISignalsListener,
                     public ISignalListening::ISignalListener
{
    public:
        using tSignal = std::pair<ISignalRegistry::tSignalAttributes, std::vector<cTestSignalValueNs>>;
        using tSignals = std::map<adtf::util::cString, tSignal>;
 
    public:
        cTestListener()
        {
            m_oUpdateCount = 0;
            THROW_IF_FAILED(_runtime->GetObject(m_pSignalListening));
            THROW_IF_FAILED(m_pSignalListening->RegisterSignalsListener(*this));
        }
 
        virtual ~cTestListener()
        {
            m_pSignalListening->UnregisterSignalsListener(*this);
 
            for (const auto& strSignal: m_oRequestedSignals)
            {
                const auto itSignal = m_oSignals.find(strSignal.c_str());
                if (itSignal != m_oSignals.end())
                {
                    m_pSignalListening->CancelSignalUpdates(itSignal->second.first.nSignalID, *this);
                }
            }
        }
 
        virtual void SignalAdded(const ISignalRegistry::tSignalAttributes& sSignalAttributes) override
        {
            {
                std::scoped_lock oSignalsGuard(m_oSignalsMutex);
 
                m_oIdMap[sSignalAttributes.nSignalID] = m_oSignals.emplace(adtf::util::cString(sSignalAttributes.strName), tSignal{sSignalAttributes, {}}).first;
 
                if (m_oRequestedSignals.find(sSignalAttributes.strName) != m_oRequestedSignals.end())
                {
                    THROW_IF_FAILED(m_pSignalListening->RequestSignalUpdates(sSignalAttributes.nSignalID, *this));
                }
            }
 
            m_oSignalsChanged.notify_all();
        }
 
        virtual void SignalRemoved(const ISignalRegistry::tSignalAttributes& sSignalAttributes) override
        {
            {
                std::scoped_lock oGuard(m_oSignalsMutex);
                m_oSignals.erase(sSignalAttributes.strName);
            }
 
            m_oSignalsChanged.notify_all();
        }
 
        virtual void SignalUpdated(ISignalRegistry::tSignalID nSignalID, const ISignalRegistry::tSignalValueNs& sValue) override
        {
            std::scoped_lock oGuard(m_oSignalsMutex);
            ++m_oUpdateCount;
            m_oIdMap.at(nSignalID)->second.second.push_back(cTestSignalValueNs(sValue));
        }
 
        const tSignals& GetSignals() const
        {
            std::scoped_lock oSignalsGuard(m_oSignalsMutex);
            return m_oSignals;
        }
 
        tSignals GetCurrentSignals()
        {
            std::scoped_lock oSignalsGuard(m_oSignalsMutex);
            return m_oSignals;
        }
 
        size_t GetUpdateCount(const char* strSignal)
        {
            std::scoped_lock oSignalsGuard(m_oSignalsMutex);
            return m_oSignals.at(strSignal).second.size();
        }
 
        void RequestUpdates(const char* strSignal)
        {;
            std::optional<ISignalRegistry::tSignalID> nId;
 
            {
                std::scoped_lock oSignalsGuard(m_oSignalsMutex);
                m_oRequestedSignals.insert(strSignal);
 
                const auto itSignal = m_oSignals.find(strSignal);
                if (itSignal != m_oSignals.end())
                {
                    nId = itSignal->second.first.nSignalID;
                }
            }
 
            if (nId)
            {
                THROW_IF_FAILED(m_pSignalListening->RequestSignalUpdates(*nId, *this));
            }
        }
 
        void CancelUpdates(const char* strSignal)
        {
            std::optional<ISignalRegistry::tSignalID> nId;
 
            {
                std::scoped_lock oSignalsGuard(m_oSignalsMutex);
                m_oRequestedSignals.erase(strSignal);
                const auto itSignal = m_oSignals.find(strSignal);
                if (itSignal != m_oSignals.end())
                {
                    nId = itSignal->second.first.nSignalID;
                }
            }
 
            if (nId)
            {
                m_pSignalListening->CancelSignalUpdates(*nId, *this);
            }
        }
 
        bool WaitForUpdates(std::chrono::seconds nMaxSeconds, size_t nUpdateCountEachSignal)
        {
            auto oBeginTimePoint = std::chrono::steady_clock::now();
            while (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - oBeginTimePoint) < nMaxSeconds)
            {
                const size_t oCurrentCount = m_oUpdateCount;
                if (oCurrentCount < nUpdateCountEachSignal * m_oSignals.size())
                {
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                }
                else
                {
                    return true;
                }
            }
            const size_t oLastCount = m_oUpdateCount;
            return (oLastCount >= nUpdateCountEachSignal * m_oSignals.size());
        }
 
        bool WaitForSignal(const char* strName, std::chrono::nanoseconds tmTimeout)
        {
            std::unique_lock oLock(m_oSignalsMutex);
            return m_oSignalsChanged.wait_for(oLock, tmTimeout, [&] { return m_oSignals.count(strName) > 0 ;});
        }
 
    private:
        adtf::ucom::object_ptr<ISignalListening> m_pSignalListening;
        mutable std::recursive_mutex m_oSignalsMutex;
        std::condition_variable_any m_oSignalsChanged;
        tSignals m_oSignals;
        std::unordered_map<ISignalRegistry::tSignalID, tSignals::iterator> m_oIdMap;
        std::unordered_set<std::string> m_oRequestedSignals;
        std::atomic<size_t> m_oUpdateCount;
};
 
}
 
using ant::cTestListener;
using ant::cTestSignalValueNs;
 
}
 
}
 
}