samplereader.h

Go to the documentation of this file.

 
#pragma once
#include "samplestreamer_intf.h"
#include "streamtype_intf.h"
#include "sample_intf.h"
#include "sample.h"
#include "streamtype.h"
#include "streammetatypeanonymous.h"
#include "streammetatypeplain.h"
#include "pin.h"
#include "sampleinstream_intf.h"
#include "samplestream_intf.h"
#include "streamitem_intf.h"
#include "samplestreamaccess_intf.h"
#include "streamingrequests_intf.h"
 
#include <adtf_utils.h>
#include <adtfucom3/adtf_ucom3.h>
#include <adtfbase/string_intf.h>
 
namespace adtf
{
namespace streaming
{
namespace ant
{
 
template <typename INTERFACE, typename PINTYPE>
class sample_streamer : public INTERFACE
{
    public:
        typedef PINTYPE pin_type;
protected:
    adtf_util::cString                                              m_strName;
    ucom::object_ptr<const IStreamType>                       m_pStreamType;
    ucom::object_ptr<PINTYPE>                                 m_poPin;
 
    sample_streamer(const sample_streamer&) = delete;
    sample_streamer(sample_streamer&&) = delete;
    sample_streamer& operator=(const sample_streamer&) = delete;
    sample_streamer& operator=(sample_streamer&&) = delete;
 
public:
    sample_streamer() = default;
    sample_streamer(const char* strName,
        const ucom::ant::iobject_ptr<const IStreamType>& pStreamType) : m_strName(strName), m_pStreamType(pStreamType)
    {
    }
    virtual ~sample_streamer()
    {
        ResetPin();
    }
 
public:
 
    void SetName(const char* strName)
    {
        m_strName = strName;
    }
 
    void ResetPin(ucom::ant::object_ptr<PINTYPE>& pPin)
    {
        if (m_poPin)
        {
            ResetPin();
        }
        if (pPin)
        {
            pPin->RegisterStreamer(*this);
            m_poPin = pPin;
        }
        else
        {
            ResetPin();
        }
    }
    void ResetPin()
    {
        if (m_poPin)
        {
            m_poPin->UnregisterStreamer(*this);
            m_poPin = nullptr;
        }
    }
    tResult GetName(base::ant::IString&& strName) const
    {
        return strName.Set(m_strName);
    }
 
    tResult SetType(const ucom::ant::iobject_ptr<const IStreamType>& pStreamType)
    {
        if (pStreamType.Get())
        {
            m_pStreamType = ucom::make_object_ptr<cStreamType>(*pStreamType.Get());
        }
        else
        {
            m_pStreamType = nullptr;
        }
        RETURN_NOERROR;
    }
 
public: //implements ISampleStreamer
    tResult GetType(ucom::ant::iobject_ptr<const IStreamType>& pStreamType) const override
    {
        RETURN_IF_FAILED(pStreamType.Reset(m_pStreamType));
        RETURN_NOERROR;
    }
 
    tResult EndStreaming() override { RETURN_NOERROR; }
};
 
class cSampleReader : public sample_streamer<ISampleReader, cInPin>,
                      public ISampleReaderQueue,
                      public base::ant::runnable<base::ant::IRunnable::RUN_PUSH, ISampleStream::IPushReadEventSink>
{
private:
    ucom::object_ptr<ISampleInStream> m_pInStream;
    std::mutex m_oReadLock;
    typedef sample_streamer<ISampleReader, cInPin> base_type;
    std::function<tResult(const ucom::iobject_ptr<const IStreamType>& pStreamType)> m_fnAcceptTypeCallback;
    std::function<tResult(tResult oStreamError)> m_fnStreamErrorCallback;
 
    typedef base::ant::runnable<base::ant::IRunnable::RUN_PUSH, ISampleStream::IPushReadEventSink> runnable_type;
 
    bool m_bValidType = true;
    ucom::object_ptr<const IStreamType> m_pLastReadStreamType;
 
    ISampleStreamAccess::tMode m_eAccessMode;
 
public:
    cSampleReader() :
        runnable_type(ADTF_RUN_FUNCTION(Push)),
        m_eAccessMode(ISampleStreamAccess::PushRead)
    {
    }
 
    cSampleReader(ISampleStreamAccess::tMode eAccessMode):
        runnable_type(ADTF_RUN_FUNCTION(Push)),
        m_eAccessMode(eAccessMode)
    {
    }
 
    virtual tResult AcceptType(const ucom::ant::iobject_ptr<const IStreamType>& pStreamType);
 
public:
    tResult BeginStreaming(ISampleStream& oSampleStream) override
    {
        EndStreaming();
 
        // async mode is not working yet.
        if (m_eAccessMode != ISampleStreamAccess::tMode::PushRead)
        {
            RETURN_ERROR_DESC(ERR_NOT_SUPPORTED, "Currently only PushRead mode is supported.");
        }
 
        {
            std::lock_guard<std::mutex> _sync(m_oReadLock);
 
            ISampleStream::IPushReadEventSink* pSink = this;
            //@TODO : Something for TYPE Checking !!!
            RETURN_IF_FAILED(oSampleStream.Open(m_strName,
                                                m_pInStream,
                                                m_pStreamType,
                                                pSink,
                                                m_eAccessMode,
                                                0));
        }
 
        RETURN_NOERROR;
    }
 
    tResult EndStreaming() override
    {
        {
            std::lock_guard<std::mutex> _sync(m_oReadLock);
            //this will automatically close
            m_pInStream.Reset();
            Clear();
        }
        RETURN_NOERROR;
    }
 
    void SetAcceptTypeCallback(const std::function<
        tResult(const ucom::iobject_ptr<const IStreamType>& pStreamType)> & fnAcceptTypeCallback )
    {
        m_fnAcceptTypeCallback = fnAcceptTypeCallback;
    }
 
    void SetStreamErrorCallback(const std::function<tResult(tResult oStreamError)> & fnStreamErrorCallback)
    {
        m_fnStreamErrorCallback = fnStreamErrorCallback;
    }
 
public:
    tResult SetStreamError(const tResult& oError)
    {
        std::lock_guard<std::mutex> _sync(m_oReadLock);
        if (m_pInStream)
        {
            //report the error !!
            return m_pInStream->SetStreamError(oError);
        }
        RETURN_ERROR(ERR_NOT_CONNECTED);
    }
 
    void GetSampleInStream(ucom::ant::iobject_ptr<ISampleInStream>& pSampleInStream)
    {
        pSampleInStream.Reset(m_pInStream);
    }
 
    void GetLastType(ucom::ant::iobject_ptr<const IStreamType>& pType)
    {
        if (m_pLastReadStreamType)
        {
            pType.Reset(m_pLastReadStreamType);
        }
        else
        {
            pType.Reset(m_pStreamType);
        }
    }
 
    virtual tResult GetNextSample(ucom::ant::iobject_ptr<const ISample>& pSample)
    {
        return ReadNextSample(pSample);
    }
 
protected:
    tResult Push(tTimeStamp tmTimeofActivation)
    {
        //implementing Push
        //this will be done sync
        #ifdef _PIPES_DEBUG_LOG
                LOG_DUMP("Run Push Event in Reader");
        #endif
        if (!m_pInStream)
        {
            RETURN_ERROR(ERR_NOT_CONNECTED);
        }
 
        while (true)
        {
            cStreamItem oItem;
            {
                //very short LOCK !!
                std::lock_guard<std::mutex> _sync(m_oReadLock);
                if (IS_FAILED(m_pInStream->Read(oItem, tTimeStamp(0))))
                {
                    break;
                }
            }
 
            RETURN_IF_FAILED(Push(oItem, tmTimeofActivation));
        }
 
 
        RETURN_NOERROR;
    }
 
protected:
    virtual tResult ReadNextSample(ucom::ant::iobject_ptr<const ISample>& pSample)
    {
        if (!m_pInStream)
        {
            return ERR_NOT_CONNECTED;
        }
 
        while (true)
        {
            cStreamItem oItem;
            if (m_pInStream->GetMode() == ISampleStreamAccess::AsyncQueue)
            {
                RETURN_IF_FAILED(m_pInStream->Read(oItem, tTimeStamp(0)));
            }
            else
            {
                RETURN_IF_FAILED(Pop(oItem));
            }
 
            if (IS_OK(ProcessStreamItem(oItem)))
            {
                if (oItem.GetType() == IStreamItem::tType::Sample)
                {
                    return oItem.GetSample(pSample);
                }
            }
        }
    }
 
    tResult ProcessStreamItem(const IStreamItem& oStreamItem)
    {
        switch (oStreamItem.GetType())
        {
            case IStreamItem::tType::StreamType:
            {
                ucom::object_ptr<const IStreamType> pHelper;
                RETURN_IF_FAILED(oStreamItem.GetStreamType(pHelper));
                tResult nError = AcceptType(pHelper);
                if (IS_FAILED(nError))
                {
                    m_bValidType = false;
                    RETURN_IF_FAILED(HandleStreamError(ADTF_BASE_COMPOSED_RESULT(nError, "Stream type is not accepted by reader.")));
                }
                else
                {
                    m_bValidType = true;
                }
                break;
            }
            default:
            {
                if (!m_bValidType)
                {
                    RETURN_ERROR(ERR_INVALID_TYPE);
                }
                break;
            }
        }
 
        RETURN_NOERROR;
    }
 
public:
    void Reset() override
    {
        m_pLastReadStreamType.Reset();
        Clear();
    }
 
public:
    using ISampleReaderQueue::Push;
    using ISampleReaderQueue::Clear;
    using ISampleReaderQueue::Pop;
 
protected:
    tResult HandleStreamError(tResult nError)
    {
        if (m_fnStreamErrorCallback)
        {
            return m_fnStreamErrorCallback(nError);
        }
        else
        {
            return SetStreamError(nError);
        }
    }
};
 
 
class cExternalQueueSampleReader : public cSampleReader
{
private:
    std::set<ISampleReaderQueue*>  m_lstExternalQueues;
 
public:
    using cSampleReader::cSampleReader;
 
    tResult Push(const IStreamItem& oStreamItem, tTimeStamp tsTime) override
    {
        for (auto pQueue : m_lstExternalQueues)
        {
            RETURN_IF_FAILED(pQueue->Push(oStreamItem, tsTime));
        }
 
        RETURN_NOERROR;
    };
 
    void Clear() override
    {
        for (auto pQueue: m_lstExternalQueues)
        {
            pQueue->Clear();
        }
    }
 
    tResult Pop(IStreamItem& /* oStreamItem */) override
    {
        RETURN_ERROR(ERR_EMPTY);
    }
 
    tResult ReadAllAvailableItems()
    {
        return cSampleReader::Push(0);
    }
 
public:
    void RegisterExternalQueue(ISampleReaderQueue * pExternalBuffer)
    {
        m_lstExternalQueues.insert(pExternalBuffer);
    }
 
    void UnregisterExternalQueue(ISampleReaderQueue * pExternalBuffer)
    {
        m_lstExternalQueues.erase(pExternalBuffer);
    }
 
};
 
class cLastSampleReader : public cSampleReader
{
protected:
    ucom::object_ptr<const ISample> m_pLastValidSample;
 
public:
 
    using cSampleReader::cSampleReader;
 
    tResult GetLastSample(ucom::ant::iobject_ptr<const ISample>& pSample)
    {
        ucom::ant::object_ptr<const ISample> pDummy;
        while (IS_OK(ReadNextSample(pDummy)));
 
        if (!m_pLastValidSample)
        {
            RETURN_ERROR(ERR_EMPTY);
        }
 
        return pSample.Reset(m_pLastValidSample);
    }
 
    void Reset() override
    {
        m_pLastValidSample = nullptr;
        cSampleReader::Reset();
    }
 
protected:
 
    tResult ReadNextSample(ucom::ant::iobject_ptr<const ISample>& pSample) override
    {
        RETURN_IF_FAILED(cSampleReader::ReadNextSample(pSample));
        m_pLastValidSample.Reset(pSample);
        RETURN_NOERROR;
    }
 
};
 
class cDynamicSampleReaderQueue:
    public ISampleReaderQueue,
    private adtf_util::lock_free_queue<cStreamItem>
{
    typedef adtf_util::lock_free_queue<cStreamItem> base_type;
 
public: // implements ISampleReaderQueue
    tResult Push(const IStreamItem& oStreamItem, tTimeStamp /*  tmTime */) override
    {
        return base_type::Push(oStreamItem);
    }
 
    void Clear() override
    {
        cStreamItem oItem;
        while (IS_OK(base_type::Pop(&oItem)));
    }
 
    tResult Pop(IStreamItem& oStreamItem) override
    {
        cStreamItem oQueueItem;
        auto oResult = base_type::Pop(&oQueueItem);
        // this is a performance optimization in order to prevent the annotation
        // in the RETURN_IF_FAILED macro below.
        if (oResult == ERR_EMPTY)
        {
            return oResult;
        }
        RETURN_IF_FAILED(oResult);
        return oQueueItem.CopyTo(oStreamItem);
    }
    
};
 
 
class cLimitedSampleReaderQueue:
    public ISampleReaderQueue
{
    private:
        std::mutex m_oQueueMutex;
        std::deque<cStreamItem> m_oItems;
        cStreamItem m_oLastType;
        size_t m_nSampleCount = 0;
 
    public:
        tResult Push(const IStreamItem& oStreamItem, tTimeStamp /* tmTime */) override
        {
            std::lock_guard<std::mutex> oGuard(m_oQueueMutex);
            m_oItems.emplace_back(oStreamItem);
            if (oStreamItem.GetType() == IStreamItem::tType::Sample)
            {
                ++m_nSampleCount;
            }
            CheckQueue(m_oItems);
            RETURN_NOERROR;
        }
 
        void Clear() override
        {
            std::lock_guard<std::mutex> oGuard(m_oQueueMutex);
            m_oItems.clear();
            m_oLastType = cStreamItem();
            m_nSampleCount = 0;
        }
 
        tResult Pop(IStreamItem& oItem)
        {
            std::lock_guard<std::mutex> oGuard(m_oQueueMutex);
            if (m_oLastType.GetType() == IStreamItem::tType::StreamType)
            {
                RETURN_IF_FAILED(m_oLastType.CopyTo(oItem));
                m_oLastType = cStreamItem();
            }
            else
            {
                if (m_oItems.empty())
                {
                    RETURN_ERROR(ERR_EMPTY);
                }
 
                RETURN_IF_FAILED(m_oItems.front().CopyTo(oItem));
                if (m_oItems.front().GetType() == IStreamItem::tType::Sample)
                {
                    --m_nSampleCount;
                }
                m_oItems.pop_front();
            }
 
            RETURN_NOERROR;
        }
 
    protected:
        virtual void CheckQueue(const std::deque<cStreamItem>& oItems) = 0;
        void PopFront()
        {
            if (m_oItems.front().GetType() == IStreamItem::tType::StreamType)
            {
                m_oLastType = m_oItems.front();
            }
            else
            {
                --m_nSampleCount;
            }
            m_oItems.pop_front();
        }
 
        size_t GetSampleCount()
        {
            return m_nSampleCount;
        }
};
 
template<size_t MaxSize>
class size_limited_sample_reader_queue:
    public cLimitedSampleReaderQueue
{
    protected:
        void CheckQueue(const std::deque<cStreamItem>& oItems) override
        {
            while (!oItems.empty() &&
                   (GetSampleCount() > MaxSize ||
                    oItems.front().GetType() == IStreamItem::tType::StreamType))
            {
                PopFront();
            }
        }
};
 
template<tTimeStamp TimeRange>
class time_limited_sample_reader_queue:
    public cLimitedSampleReaderQueue
{
    protected:
        void CheckQueue(const std::deque<cStreamItem>& oItems) override
        {
            if (oItems.back().GetType() == IStreamItem::tType::Sample)
            {
                ucom::ant::object_ptr<const ISample> pLastSample;
                if (IS_OK(oItems.back().GetSample(pLastSample)))
                {
                    tTimeStamp nLastTime = pLastSample->GetTime();
                    while (oItems.size() > 1)
                    {
                        auto& oItem = oItems.front();
                        if (oItem.GetType() == IStreamItem::tType::Sample)
                        {
                            ucom::ant::object_ptr<const ISample> pSample;
                            if (IS_OK(oItem.GetSample(pSample)))
                            {
                                if (nLastTime - pSample->GetTime() < TimeRange)
                                {
                                    break;
                                }
                            }
 
                        }
 
                        PopFront();
                    }
                }
            }
        }
};
 
template<typename INTERNAL_QUEUE,
         bool STORE_LAST_SAMPLE = true,
         ISampleStreamAccess::tMode ACCESS_MODE = ISampleStreamAccess::tMode::PushRead>
class sample_reader : public std::conditional<STORE_LAST_SAMPLE, cLastSampleReader, cSampleReader>::type
{
protected:
    INTERNAL_QUEUE                                                  m_oInternalQueue;
 
    typedef typename std::conditional<STORE_LAST_SAMPLE, cLastSampleReader, cSampleReader>::type base_class;
public:
 
    sample_reader(): base_class(ACCESS_MODE)
    {
    }
 
    tResult Push(const IStreamItem& oItem, tTimeStamp tsTime) override
    {
        return m_oInternalQueue.Push(oItem, tsTime);
    };
 
    void Clear() override
    {
        m_oInternalQueue.Clear();
    };
 
    tResult Pop(IStreamItem& oItem) override
    {
        return m_oInternalQueue.Pop(oItem);
    }
};
 
using cDynamicSampleReader = sample_reader<cDynamicSampleReaderQueue>;
 
template<tTimeStamp TIME_RANGE, bool STORELASTSAMPLE = true>
using time_limited_sample_reader = sample_reader<time_limited_sample_reader_queue<TIME_RANGE>, STORELASTSAMPLE>;
 
template<size_t MAX_SIZE, bool STORELASTSAMPLE=true>
using size_limited_sample_reader = sample_reader<size_limited_sample_reader_queue<MAX_SIZE>, STORELASTSAMPLE>;
 
using cSingleSampleReader = size_limited_sample_reader<1>;
 
inline cSampleReader& operator>>(cSampleReader& oSampleReader, ucom::ant::iobject_ptr<const ISample>& pSample)
{
    ucom::ant::object_ptr<const ISample> pEmpty;
    pSample.Reset(pEmpty);
    oSampleReader.GetNextSample(pSample);
    return oSampleReader;
}
 
template<typename DATATYPE>
cSampleReader& operator>>(cSampleReader& oSampleReader, sample_data<DATATYPE>& oSampleData)
{
    ucom::object_ptr<const ISample> pSample;
    if (IS_OK(oSampleReader.GetNextSample(pSample)))
    {
        oSampleData.Reset(pSample);
    }
    else
    {
        oSampleData.Reset();
    }
    return oSampleReader;
}
inline cSampleReader& operator>>(cSampleReader& oSampleReader, ucom::ant::iobject_ptr<const IStreamType>& pType)
{
    oSampleReader.GetLastType(pType);
    return oSampleReader;
}
inline cSampleReader& operator>>(cSampleReader& oSampleReader, cSampleReader& (*pStreamfunc)(cSampleReader&))
{
    return pStreamfunc(oSampleReader);
}
 
inline tResult make_sample_reader(cSampleReader & oReader,
                                  const char* strNameOfReader,
                                  const ucom::iobject_ptr<const IStreamType>& pStreamType)
{
    oReader.SetName(strNameOfReader);
    return oReader.SetType(pStreamType);
}
 
class cExternelQueueSampleReader : public cExternalQueueSampleReader
{
public:
    using cExternalQueueSampleReader::cExternalQueueSampleReader;
 
    ADTF3_DEPRECATED("The class 'cExternelQueueSampleReader' is deprecated. Please use 'cExternalQueueSampleReader' instead.")
    cExternelQueueSampleReader()
        : cExternalQueueSampleReader()
    {
 
    }
};
 
} //namespace ant
 
namespace flash
{
 
class cSampleReader: public ISampleReader
{
    public:
        cSampleReader(ant::ISampleReaderQueue& oQueue,
                      ant::ISampleStreamAccess::tMode eAccessMode,
                      bool bStoreLastSample);
 
        ~cSampleReader() override;
 
        void SetName(const char* strName) override;
        tResult GetName(base::ant::IString&& strName) override;
 
        tResult SetType(const ucom::ant::iobject_ptr<const ant::IStreamType>& pStreamType) override;
        tResult GetType(ucom::ant::iobject_ptr<const ant::IStreamType>& pStreamType) const override;
 
        tResult BeginStreaming(ISampleStream& oSampleStream) override;
        tResult EndStreaming() override;
 
        tResult SetStreamerPin(const ucom::ant::iobject_ptr<IStreamerPin>& pStreamerPin) override;
 
        void SetAcceptTypeCallback(const std::function<tResult(const ucom::ant::iobject_ptr<const ant::IStreamType>& pStreamType)> & fnAcceptTypeCallback);
 
        void SetStreamErrorCallback(const std::function<tResult(tResult oStreamError)> & fnStreamErrorCallback);
        tResult SetStreamError(tResult oError) override;
 
        tResult GetLastType(ucom::ant::iobject_ptr<const ant::IStreamType>& pType) override;
        tResult GetNextSample(ucom::ant::iobject_ptr<const ant::ISample>& pSample) override;
 
        tResult GetLastSample(ucom::ant::iobject_ptr<const ant::ISample>& pSample) override;
        void Reset();
 
        tResult ReadAllAvailableItems() override;
 
        tResult RequestSamples(ucom::ant::iobject_ptr<hollow::IStreamingRequest>& pRequest,
                               uint32_t nSubStreamId,
                               const base::ant::IProperties* pRequestProperties = nullptr);
        void SetSynchronousTypeUpdateCallback(const std::function<tResult(const ucom::ant::iobject_ptr<const ant::IStreamType>& pStreamType)>& fnRequestTypeUpdateCallback);
 
    protected:
        class cImplementation;
        std::unique_ptr<cImplementation> m_pImplementation;
 
};
 
template<typename INTERNAL_QUEUE,
         bool STORE_LAST_SAMPLE = true,
         ant::ISampleStreamAccess::tMode ACCESS_MODE = ant::ISampleStreamAccess::tMode::Push>
class sample_reader : public cSampleReader
{
    protected:
        INTERNAL_QUEUE  m_oQueue;
 
    public:
        sample_reader(): cSampleReader(m_oQueue, ACCESS_MODE, STORE_LAST_SAMPLE)
        {}
};
 
 
 
class cExternalQueuesWrapper: public ISampleReaderQueue
{
    public:
        void RegisterExternalQueue(ISampleReaderQueue * pExternalBuffer)
        {
            m_oExternalQueues.insert(pExternalBuffer);
        }
 
        void UnregisterExternalQueue(ISampleReaderQueue * pExternalBuffer)
        {
            m_oExternalQueues.erase(pExternalBuffer);
        }
 
    public:
        tResult Push(const IStreamItem& oStreamItem, tTimeStamp tsTime) override
        {
            for (auto pQueue : m_oExternalQueues)
            {
                RETURN_IF_FAILED(pQueue->Push(oStreamItem, tsTime));
            }
 
            RETURN_NOERROR;
        };
 
        void Clear() override
        {
            for (auto pQueue: m_oExternalQueues)
            {
                pQueue->Clear();
            }
        }
 
        tResult Pop(IStreamItem& /* oStreamItem */) override
        {
            RETURN_ERROR(ERR_EMPTY);
        }
 
    protected:
        std::set<ISampleReaderQueue*>  m_oExternalQueues;
};
 
class IExternalReaderQueues
{
    protected:
        virtual ~IExternalReaderQueues() = default;
 
    public:
        virtual void RegisterExternalQueue(ISampleReaderQueue * pExternalBuffer) = 0;
        virtual void UnregisterExternalQueue(ISampleReaderQueue * pExternalBuffer) = 0;
};
 
class cExternalQueueSampleReader : public sample_reader<cExternalQueuesWrapper, false>,
                                   public IExternalReaderQueues
{
    public:
        void RegisterExternalQueue(ISampleReaderQueue * pExternalBuffer) override
        {
            m_oQueue.RegisterExternalQueue(pExternalBuffer);
        }
 
        void UnregisterExternalQueue(ISampleReaderQueue * pExternalBuffer) override
        {
            m_oQueue.UnregisterExternalQueue(pExternalBuffer);
        }
};
 
using cDynamicSampleReader = sample_reader<ant::cDynamicSampleReaderQueue>;
 
 
template<tTimeStamp TIME_RANGE, bool STORELASTSAMPLE = true>
using time_limited_sample_reader = sample_reader<ant::time_limited_sample_reader_queue<TIME_RANGE>, STORELASTSAMPLE>;
 
template<size_t MAX_SIZE, bool STORELASTSAMPLE=true>
using size_limited_sample_reader = sample_reader<ant::size_limited_sample_reader_queue<MAX_SIZE>, STORELASTSAMPLE>;
 
using cSingleSampleReader = size_limited_sample_reader<1>;
 
inline tResult make_sample_reader(ISampleReader& oReader,
                                  const char* strNameOfReader,
                                  const ucom::iobject_ptr<const IStreamType>& pStreamType)
{
    oReader.SetName(strNameOfReader);
    return oReader.SetType(pStreamType);
}
 
inline ISampleReader& operator>>(ISampleReader& oSampleReader, ucom::ant::iobject_ptr<const ant::ISample>& pSample)
{
    ucom::ant::object_ptr<const ant::ISample> pEmpty;
    pSample.Reset(pEmpty);
    oSampleReader.GetNextSample(pSample);
    return oSampleReader;
}
 
template<typename DATATYPE>
ISampleReader& operator>>(ISampleReader& oSampleReader, sample_data<DATATYPE>& oSampleData)
{
    ucom::object_ptr<const ant::ISample> pSample;
    if (IS_OK(oSampleReader.GetNextSample(pSample)))
    {
        oSampleData.Reset(pSample);
    }
    else
    {
        oSampleData.Reset();
    }
    return oSampleReader;
}
 
inline ISampleReader& operator>>(ISampleReader& oSampleReader, ucom::ant::iobject_ptr<const ant::IStreamType>& pType)
{
    oSampleReader.GetLastType(pType);
    return oSampleReader;
}
 
inline ISampleReader& operator>>(ISampleReader& oSampleReader, ISampleReader& (*pStreamfunc)(ISampleReader&))
{
    return pStreamfunc(oSampleReader);
}
 
}
 
namespace kiwi
{
 
class cNullReader: public flash::ISampleReader
{
public:
    cNullReader();
 
    ~cNullReader() override;
 
    void SetName(const char* strName) override;
    tResult GetName(base::ant::IString&& strName) override;
    tResult SetType(const ucom::ant::iobject_ptr<const ant::IStreamType>& pStreamType) override;
    tResult GetType(ucom::ant::iobject_ptr<const ant::IStreamType>& pStreamType) const override;
    tResult BeginStreaming(ant::ISampleStream& pStream) override;
    tResult EndStreaming() override;
    tResult SetStreamerPin(const ucom::ant::iobject_ptr<flash::IStreamerPin>& pStreamerPin) override;
 
    tResult GetNextSample(ucom::ant::iobject_ptr<const ant::ISample>& pSample) override;
    tResult GetLastSample(ucom::ant::iobject_ptr<const ant::ISample>& pSample) override;
    tResult GetLastType(ucom::ant::iobject_ptr<const ant::IStreamType>& pType) override;
    tResult SetStreamError(tResult oError) override;
    tResult ReadAllAvailableItems() override;
 
    void SetAcceptTypeCallback(const std::function<tResult(const ucom::ant::iobject_ptr<const ant::IStreamType>& pStreamType)> & fnAcceptTypeCallback);
    void SetStreamErrorCallback(const std::function<tResult(tResult oStreamError)> & fnStreamErrorCallback);
 
protected:
    class cImplementation;
    std::unique_ptr<cImplementation> m_pImplementation;
};
 
}
 
using ant::cDynamicSampleReaderQueue;
using ant::size_limited_sample_reader_queue;
using ant::time_limited_sample_reader_queue;
 
using flash::cSampleReader;
using flash::cExternalQueueSampleReader;
using ant::cExternelQueueSampleReader;
using flash::sample_reader;
using flash::time_limited_sample_reader;
using flash::size_limited_sample_reader;
using flash::cSingleSampleReader;
using flash::cDynamicSampleReader;
 
using flash::make_sample_reader;
 
using kiwi::cNullReader;
 
 
} //namespace streaming
} //namespace adtf