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
{
public: // implements ISampleReaderQueue
    tResult Push(const IStreamItem& oStreamItem, tTimeStamp /*  tmTime */) override
    {
        std::scoped_lock oGuard(m_oFrontQueue.oMutex);
        RETURN_IF_THROWS(m_oFrontQueue.oQueue.emplace(oStreamItem));
        RETURN_NOERROR;
    }
 
    void Clear() override
    {
        std::scoped_lock oGuard(m_oFrontQueue.oMutex);
        m_oFrontQueue.oQueue = {};
        m_oConsumerQueue = {};
    }
 
    tResult Pop(IStreamItem& oStreamItem) override
    {
        if (m_oConsumerQueue.empty())
        {
            std::scoped_lock oGuard(m_oFrontQueue.oMutex);
            std::swap(m_oConsumerQueue, m_oFrontQueue.oQueue);
        }
        if (m_oConsumerQueue.empty())
        {
            RETURN_ERROR(ERR_EMPTY);
        }
        const auto oResult = m_oConsumerQueue.front().CopyTo(oStreamItem);
        m_oConsumerQueue.pop();
        return oResult;
    }
 
private:
    struct alignas(64)
    {
        std::queue<cStreamItem> oQueue;
        std::mutex oMutex;
    } m_oFrontQueue;
    alignas(64) std::queue<cStreamItem> m_oConsumerQueue;
};
 
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("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 flash
 
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;
};
 
} // namespace kiwi
 
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