sample_collectors.h

Go to the documentation of this file.

 
#pragma once
#include "samplereader.h"
 
#include <mutex>
#include <deque>
#include <vector>
#include <memory>
#include <functional>
 
namespace adtf
{
namespace streaming
{
namespace ant
{
    
namespace detail
{
 
class cExternalSampleReaderQueue: public ISampleReaderQueue
{
    private:
        cStreamItem m_oLastReadSampleStreamItem;
 
        std::recursive_mutex m_oQueueMutex;
        std::deque<cStreamItem> m_oItems;
 
        cExternalQueueSampleReader& m_oSampleReader;
 
    public:
        cExternalSampleReaderQueue(cExternalQueueSampleReader& oSampleReader);
        ~cExternalSampleReaderQueue();
 
        tResult GetSampleBeforeOrAt(tTimeStamp tmTime,
                                    adtf::ucom::iobject_ptr<const ISample>& pSample,
                                    adtf::ucom::iobject_ptr<const IStreamType>& pStreamType);
        tResult GetNextSampleAndType(adtf::ucom::iobject_ptr<const ISample>& pSample,
                                     adtf::ucom::iobject_ptr<const IStreamType>& pStreamType);
 
        tResult Push(const IStreamItem& oStreamItem, tTimeStamp tmTime) override;
        tResult Pop(IStreamItem& oStreamItem) override;
        void Clear() override;
 
        size_t IsEmpty();
 
        cExternalQueueSampleReader& GetReader();
};
 
}
    
class cLastSamplesCollector
{
    private:
        std::vector<cLastSampleReader*> m_oSampleReaders;
 
    public:
 
        tResult CollectFrom(cLastSampleReader& oSampleReader)
        {
            m_oSampleReaders.push_back(&oSampleReader);
            RETURN_NOERROR;
        }
 
        template<typename ... T>
        tResult GetLastSamples(adtf::ucom::iobject_ptr<const ISample> & pSample, T & ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
            if (m_oSampleReaders.size() > nSize)
            {
                RETURN_IF_FAILED(m_oSampleReaders[m_oSampleReaders.size() - nSize - 1]->GetLastSample(pSample));
            }
            else
            {
                RETURN_ERROR(ERR_INVALID_ARG);
            }
            return GetLastSamples(samples...);
        }
 
        tResult GetLastSamples()
        {
            RETURN_NOERROR;
        }
};
 
class cSampleSetCollector
{
    public:
        typedef std::function<tResult(cExternalQueueSampleReader&, const adtf::ucom::iobject_ptr<const IStreamType>& pStreamType)> tStreamTypeChangeCallback;
 
    private:
        tStreamTypeChangeCallback m_pTypeChangeCallback;
 
        struct tQueue
        {
            std::shared_ptr<detail::cExternalSampleReaderQueue> pSampleQueue;
            adtf::ucom::ant::object_ptr<const ISample> pFirstSample;
            adtf::ucom::ant::object_ptr<const ISample> pNextSample;
            adtf::ucom::ant::object_ptr<const IStreamType> pNextSampleType;
        };
 
        std::vector<tQueue> m_oQueues;
 
    public:
        cSampleSetCollector() = default;
 
        cSampleSetCollector(tStreamTypeChangeCallback pTypeChangeCallback);
 
        cSampleSetCollector(const cSampleSetCollector&) = delete;
        cSampleSetCollector(cSampleSetCollector&&) = default;
 
        cSampleSetCollector& operator=(const cSampleSetCollector&) = delete;
        cSampleSetCollector& operator=(cSampleSetCollector&&) = default;
 
 
        tResult CollectFrom(cExternalQueueSampleReader& oSampleReader);
 
        void ClearQueues();
 
        template<typename ...T>
        tResult GetNextSampleSet(T& ...samples)
        {
            if (m_oQueues.size() <= 0)
            {
                RETURN_ERROR(ERR_EMPTY);
            }
 
            constexpr size_t nSize = sizeof...(T);
            if (m_oQueues.size() < nSize)
            {
                RETURN_ERROR(ERR_INVALID_ARG);
            }
 
            RETURN_IF_FAILED(RetrieveSamples());
 
            return GetSamples(samples...);
        }
 
    private:
        tResult RetrieveSamples();
 
        template<typename ... T>
        tResult GetSamples(adtf::ucom::iobject_ptr<const ISample>& pSample, T & ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
 
            RETURN_IF_FAILED(pSample.Reset(m_oQueues[m_oQueues.size() - nSize - 1].pFirstSample));
 
            return GetSamples(samples...);
        }
 
        tResult GetSamples()
        {
            RETURN_NOERROR;
        }
};
 
class cTimeBasedSampleSetCollector
{
    public:
        typedef std::function<tResult(cExternalQueueSampleReader&, const adtf::ucom::iobject_ptr<const IStreamType>& pStreamType)> tStreamTypeChangeCallback;
 
    private:
        std::vector<std::shared_ptr<detail::cExternalSampleReaderQueue>> m_oQueues;
        tStreamTypeChangeCallback m_pTypeChangeCallback;
 
    public:
        cTimeBasedSampleSetCollector() = default;
 
        cTimeBasedSampleSetCollector(tStreamTypeChangeCallback pTypeChangeCallback);
 
        cTimeBasedSampleSetCollector(const cTimeBasedSampleSetCollector&) = delete;
        cTimeBasedSampleSetCollector(cTimeBasedSampleSetCollector&&) = default;
 
        cTimeBasedSampleSetCollector& operator=(const cTimeBasedSampleSetCollector&) = delete;
        cTimeBasedSampleSetCollector& operator=(cTimeBasedSampleSetCollector&&) = default;
 
        tResult CollectFrom(cExternalQueueSampleReader& oSampleReader);
 
        void ClearQueues();
 
        template<typename ... T>
        tResult GetSampleSetAt(tTimeStamp tmTime, T& ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
 
            if (m_oQueues.size() < nSize)
            {
                RETURN_ERROR(ERR_INVALID_ARG);
            }
 
            bool bSampleFound = false;
            RETURN_IF_FAILED(GetSampleSetWithCheckAt(tmTime, bSampleFound, samples...));
 
            if (!bSampleFound)
            {
                RETURN_ERROR(ERR_NOT_FOUND);
            }
 
            RETURN_NOERROR;
        }
 
    private:
        template<typename ... T>
        tResult GetSampleSetWithCheckAt(tTimeStamp tmTime, bool& bSampleFound, adtf::ucom::iobject_ptr<const ISample>& pSample, T & ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
 
            auto& pQueue = m_oQueues[m_oQueues.size() - nSize - 1];
            adtf::ucom::object_ptr<const IStreamType> pNewStreamType;
            bSampleFound |= IS_OK(pQueue->GetSampleBeforeOrAt(tmTime, pSample, pNewStreamType));
 
            if (pNewStreamType && m_pTypeChangeCallback)
            {
                RETURN_IF_FAILED(m_pTypeChangeCallback(pQueue->GetReader(), pNewStreamType));
            }
 
            return GetSampleSetWithCheckAt(tmTime, bSampleFound, samples...);
        }
 
        tResult GetSampleSetWithCheckAt(tTimeStamp /* tmTime */, bool& /* bSampleFound */)
        {
            RETURN_NOERROR;
        }
};
 
 
} //namespace ant
 
namespace flash
{
 
namespace detail
{
 
class cExternalSampleReaderQueue: public ISampleReaderQueue
{
    public:
        cExternalSampleReaderQueue(ISampleReader& oSampleReader);
        virtual ~cExternalSampleReaderQueue();
 
        tResult GetSampleBeforeOrAt(tTimeStamp tmTime,
                                    adtf::ucom::iobject_ptr<const ant::ISample>& pSample,
                                    adtf::ucom::iobject_ptr<const ant::IStreamType>& pStreamType);
        tResult GetNextSampleAndType(adtf::ucom::iobject_ptr<const ant::ISample>& pSample,
                                     adtf::ucom::iobject_ptr<const ant::IStreamType>& pStreamType);
 
        tResult Push(const IStreamItem& oStreamItem, tTimeStamp tmTime) override;
        tResult Pop(IStreamItem& oStreamItem) override;
        void Clear() override;
 
        size_t IsEmpty();
 
        ISampleReader& GetReader();
 
    private:
        cStreamItem m_oLastReadSampleStreamItem;
 
        std::recursive_mutex m_oQueueMutex;
        std::deque<cStreamItem> m_oItems;
 
        ISampleReader& m_oSampleReader;
};
 
template<typename T>
struct function_type_from_operator
{
    using type = void;
};
 
template<typename Ret, typename Class, typename... Args>
struct function_type_from_operator<Ret(Class::*)(Args...) const>
{
    using type = std::function<Ret(Args...)>;
};
 
template<typename F>
typename function_type_from_operator<decltype(&F::operator())>::type
function_from_lambda(F const &func)
{
    return func;
}
 
template <typename ReaderType>
std::function<tResult(ISampleReader&, const adtf::ucom::iobject_ptr<const IStreamType>&)>
get_type_change_callback_translator(std::function<tResult(ReaderType&, const adtf::ucom::iobject_ptr<const IStreamType>&)> fnCallback)
{
    return [=](ISampleReader& oReader, const adtf::ucom::iobject_ptr<const IStreamType>& pStreamType) -> tResult
    {
        auto pReader = dynamic_cast<ReaderType*>(&oReader);
        if (!pReader)
        {
            RETURN_ERROR_DESC(ERR_UNEXPECTED, "Reader is not of the expected class.");
        }
        return fnCallback(*pReader, pStreamType);
    };
}
 
template <typename Callable>
std::function<tResult(ISampleReader&, const adtf::ucom::iobject_ptr<const IStreamType>&)>
get_type_change_callback_translator(Callable&& fnCallback)
{
    return get_type_change_callback_translator(function_from_lambda(fnCallback));
}
 
}
 
class cLastSamplesCollector
{
    private:
        std::vector<ISampleReader*> m_oSampleReaders;
 
    public:
 
        tResult CollectFrom(ISampleReader& oSampleReader)
        {
            m_oSampleReaders.push_back(&oSampleReader);
            RETURN_NOERROR;
        }
 
        template<typename ... T>
        tResult GetLastSamples(adtf::ucom::iobject_ptr<const ant::ISample>& pSample, T & ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
            if (m_oSampleReaders.size() > nSize)
            {
                RETURN_IF_FAILED(m_oSampleReaders[m_oSampleReaders.size() - nSize - 1]->GetLastSample(pSample));
            }
            else
            {
                RETURN_ERROR(ERR_INVALID_ARG);
            }
            return GetLastSamples(samples...);
        }
 
        tResult GetLastSamples()
        {
            RETURN_NOERROR;
        }
};
 
class cSampleSetCollector
{
    public:
        typedef std::function<tResult(ISampleReader&, const adtf::ucom::iobject_ptr<const IStreamType>& pStreamType)> tStreamTypeChangeCallback;
 
    public:
        cSampleSetCollector() = default;
 
        template <typename Callback>
        cSampleSetCollector(Callback pTypeChangeCallback):
            cSampleSetCollector()
        {
            m_pTypeChangeCallback = detail::get_type_change_callback_translator(pTypeChangeCallback);
        }
 
        cSampleSetCollector(const cSampleSetCollector&) = delete;
        cSampleSetCollector(cSampleSetCollector&&) = default;
 
        cSampleSetCollector& operator=(const cSampleSetCollector&) = delete;
        cSampleSetCollector& operator=(cSampleSetCollector&&) = default;
 
        tResult CollectFrom(ISampleReader& oSampleReader);
        void ClearQueues();
 
        template<typename ...T>
        tResult GetNextSampleSet(T& ...samples)
        {
            if (m_oQueues.size() <= 0)
            {
                RETURN_ERROR(ERR_EMPTY);
            }
 
            constexpr size_t nSize = sizeof...(T);
            if (m_oQueues.size() < nSize)
            {
                RETURN_ERROR(ERR_INVALID_ARG);
            }
 
            RETURN_IF_FAILED(RetrieveSamples());
 
            return GetSamples(samples...);
        }
 
    private:
        tResult RetrieveSamples();
 
        template<typename ... T>
        tResult GetSamples(adtf::ucom::iobject_ptr<const ant::ISample>& pSample, T & ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
 
            RETURN_IF_FAILED(pSample.Reset(m_oQueues[m_oQueues.size() - nSize - 1].pFirstSample));
 
            return GetSamples(samples...);
        }
 
        tResult GetSamples()
        {
            RETURN_NOERROR;
        }
 
    private:
        tStreamTypeChangeCallback m_pTypeChangeCallback;
 
        struct tQueue
        {
            std::shared_ptr<detail::cExternalSampleReaderQueue> pSampleQueue;
            adtf::ucom::ant::object_ptr<const ant::ISample> pFirstSample;
            adtf::ucom::ant::object_ptr<const ant::ISample> pNextSample;
            adtf::ucom::ant::object_ptr<const ant::IStreamType> pNextSampleType;
        };
 
        std::vector<tQueue> m_oQueues;
};
 
class cTimeBasedSampleSetCollector
{
    public:
        typedef std::function<tResult(ISampleReader&, const adtf::ucom::iobject_ptr<const IStreamType>& pStreamType)> tStreamTypeChangeCallback;
 
    public:
        cTimeBasedSampleSetCollector() = default;
 
        template <typename Callback>
        cTimeBasedSampleSetCollector(Callback pTypeChangeCallback):
            cTimeBasedSampleSetCollector()
        {
            m_pTypeChangeCallback = detail::get_type_change_callback_translator(pTypeChangeCallback);
        }
 
        cTimeBasedSampleSetCollector(const cTimeBasedSampleSetCollector&) = delete;
        cTimeBasedSampleSetCollector(cTimeBasedSampleSetCollector&&) = default;
 
        cTimeBasedSampleSetCollector& operator=(const cTimeBasedSampleSetCollector&) = delete;
        cTimeBasedSampleSetCollector& operator=(cTimeBasedSampleSetCollector&&) = default;
 
        tResult CollectFrom(ISampleReader& oSampleReader);
        void ClearQueues();
 
        template<typename ... T>
        tResult GetSampleSetAt(tTimeStamp tmTime, T& ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
 
            if (m_oQueues.size() < nSize)
            {
                RETURN_ERROR(ERR_INVALID_ARG);
            }
 
            bool bSampleFound = false;
            RETURN_IF_FAILED(GetSampleSetWithCheckAt(tmTime, bSampleFound, samples...));
 
            if (!bSampleFound)
            {
                RETURN_ERROR(ERR_NOT_FOUND);
            }
 
            RETURN_NOERROR;
        }
 
    private:
        template<typename ... T>
        tResult GetSampleSetWithCheckAt(tTimeStamp tmTime, bool& bSampleFound, adtf::ucom::iobject_ptr<const ant::ISample>& pSample, T & ... samples)
        {
            constexpr size_t nSize = sizeof...(T);
 
            auto& pQueue = m_oQueues[m_oQueues.size() - nSize - 1];
            adtf::ucom::object_ptr<const IStreamType> pNewStreamType;
            bSampleFound |= IS_OK(pQueue->GetSampleBeforeOrAt(tmTime, pSample, pNewStreamType));
 
            if (pNewStreamType && m_pTypeChangeCallback)
            {
                RETURN_IF_FAILED(m_pTypeChangeCallback(pQueue->GetReader(), pNewStreamType));
            }
 
            return GetSampleSetWithCheckAt(tmTime, bSampleFound, samples...);
        }
 
        tResult GetSampleSetWithCheckAt(tTimeStamp /* tmTime */, bool& /* bSampleFound */)
        {
            RETURN_NOERROR;
        }
 
    private:
        std::vector<std::shared_ptr<detail::cExternalSampleReaderQueue>> m_oQueues;
        tStreamTypeChangeCallback m_pTypeChangeCallback;
};
 
 
}
 
using flash::cLastSamplesCollector;
using flash::cSampleSetCollector;
using flash::cTimeBasedSampleSetCollector;
 
} //namespace streaming
} //namespace adtf