samplestream.h

Go to the documentation of this file.

#pragma once
#include "samplestream_intf.h"
#include "sampleinstream_intf.h"
#include "sampleoutstream_intf.h"
#include "streamtype.h"
#include "samplestream_statistics_intf.h"
#include "samplestreamaccess_intf.h"
#include "triggerpipe_intf.h"
#include "triggerpipe.h"
#include "streamingrequests_intf.h"
#include "interface_binding_proxy_intf.h"
#include "reference_clock_intf.h"
 
#include <adtfbase/adtf_base.h>
#include <adtfucom3/adtf_ucom3.h>
 
#include <mutex>
#include <optional>
#include <string>
#include <memory>
#include <unordered_map>
#include <vector>
 
namespace adtf
{
namespace streaming
{
 
namespace private_implementations
{
class cStreamingRequestsProxy;
}
 
namespace detail
{

class cSampleStreamBase :
    public ucom::ant::enable_object_ptr_from_this<ISampleStream>,
    private hollow::IStreamTypeHandler
{
public:
    cSampleStreamBase(cTriggerPipeItemBase& oPipeItem, ISampleStream& oSelf) noexcept;
    ~cSampleStreamBase();
 
    cSampleStreamBase() = delete;
    cSampleStreamBase(const cSampleStreamBase&) = delete;
    cSampleStreamBase& operator=(const cSampleStreamBase&) = delete;
 
public: // implements the sample stream
    tResult GetType(ucom::ant::iobject_ptr<const IStreamType>& pStreamType) const noexcept;
    tResult SetStreamError(const tResult& oError) noexcept;
    tResult AttachRouting(const ucom::ant::iobject_ptr<ISampleStream>& pSampleStreamTo) noexcept;
    tResult DetachRouting(const ucom::ant::iobject_ptr<ISampleStream>& pSampleStreamTo) noexcept;
    tResult Open(ucom::ant::iobject_ptr<ISampleInStream>& pInStream,
                 ISampleStream::IPushReadEventSink*& pPushEventSink,
                 ISampleStreamAccess::tMode ui32Mode,
                 size_t szQueueSize) noexcept;
    tResult Open(ucom::ant::iobject_ptr<ISampleOutStream>& pOutStream,
                 ISampleStreamAccess::tMode ui32Mode,
                 size_t szQueueSize) noexcept;
 
public: // implements the trigger pipe source
    tResult RegisterSubItem(const ucom::ant::iobject_ptr<ITriggerPipeItem>& pSubRun,
                            ITriggerPipeItem::tPriority ui32Prio) noexcept;
    tResult UnregisterSubItem(const ucom::ant::iobject_ptr<ITriggerPipeItem>& pSubRun) noexcept;
 
public: // implements hollow::IInternalBindingProxy and vision::IBlockableSampleStream
    tResult GetInternalBindingProxy(ucom::ant::iobject_ptr<IBindingProxy>& pBindingProxy) noexcept;
    void BlockSampleStreaming(bool bBlock) noexcept;
    bool IsSampleStreamingBlocked() const noexcept;
 
public: // implements wolverine::ISampleStreamStatistics, @see cStatistics
    wolverine::ISampleStreamStatistics::tSampleStreamStatistic GetStatistics() noexcept;

    void EnableSampleSizeStatistics(bool bEnable) noexcept;

    void ResetStatistics() noexcept;
 
public: // policies
    void SetForwardAllTriggers(bool bForwardAll) noexcept;

    void SetSubStreamFilter(const char* strSubStream) noexcept;
 
private:
    class cOutStream;
    class cPushStream;
 
    struct tItemFilter;
 
    struct tSink;
    using tSinkList = std::vector<tSink>;
 
    struct tDirectChild;
    using tDirectChildren = std::vector<tDirectChild>;
 
    class cStatistics;
 
private:
    void UpdateStreamTypeHandler() noexcept;
    void HandleNewStreamType(const ucom::ant::iobject_ptr<const IStreamType>& pStreamType) noexcept override;
 
    void Push(const ucom::ant::iobject_ptr<const ISample>& pSample) noexcept;
    void Push(const ucom::ant::iobject_ptr<const IStreamType>& pStreamType) noexcept;
    bool ResolveSubstreamFilter(const ucom::ant::iobject_ptr<const IStreamType>& pStreamType,
                                ucom::ant::object_ptr<const IStreamType>& pSubstreamType,
                                uint32_t& nSubstreamId) noexcept;
    void ReportUnresolvedSubstreamFilter() noexcept;
 
    tResult AttachBindingProxy(const ucom::ant::iobject_ptr<ISampleStream>& pSampleStreamTo) noexcept;
    void DetachBindingProxy(const ucom::ant::iobject_ptr<ISampleStream>& pSampleStreamTo) noexcept;
 
    void HandleSubItemTriggerError(tResult oError) noexcept;
    bool IsInFoldChain(const cSampleStreamBase* pStream) const noexcept;
    const cSampleStreamBase* FindFilterAtOrAbove() const noexcept;
    const cSampleStreamBase* FindFilterAtOrBelow() const noexcept;
    bool HasWriter() const noexcept;
    bool IsAttached() const noexcept;
    bool IsAttachedToOther(const cSampleStreamBase* pStream) const noexcept;
    void SetFoldedInto(cSampleStreamBase* pStream) noexcept;
 
    tDirectChildren::iterator FindDirectChild(const cSampleStreamBase* pChild) noexcept;
    tDirectChildren::iterator FindTriggerChild(const ITriggerPipeItem* pItem) noexcept;
    tDirectChildren::iterator AcquireDirectChild(cSampleStreamBase* pChild) noexcept;
 
    void ApplyPolicy(tSink& sSink) noexcept;
    static const void* GetSinkKey(const tSink& sSink) noexcept;
    bool HasDirectReader() const noexcept;
    void CollectStatistics(cStatistics& oRoot, const tSink& sFromParent, bool bTriggerReachable) noexcept;
 
    void ApplySubstreamFilter(tItemFilter& sFilter) noexcept;
    const tSinkList& RebuildSinks(bool bRoot) noexcept;
    cSampleStreamBase& GetFoldRoot() noexcept;
    void NotifyTopologyChanged() noexcept;
    void FlushInitialTypes() noexcept;
    void UnregisterPushSink(cPushStream* pSink) noexcept;
    void DropChild(cSampleStreamBase* pChild) noexcept;

    std::string GetFullName() const noexcept;
 
private:
    // This only stores the hierachy of the trigger pipe but is no longer used to run it.
    cTriggerPipeItemBase& m_oPipeItem;
    ISampleStream& m_oSelf;
 
    bool m_bForwardAllTrigger = true;
    bool m_bSamplesBlocked = false;
    std::string m_strSubstreamFilter;
    ucom::ant::object_ptr<const IStreamType> m_pFilteredType;
    ucom::ant::object_ptr<const IStreamType> m_pSubstreamType;
    std::optional<uint32_t> m_nSubstreamIdFilter;
    tResult m_nUnresolvedError = ERR_NOERROR;

    ucom::ant::object_ptr<IBindingProxy> m_pInternalBindingProxy;
    private_implementations::cStreamingRequestsProxy* m_pRequestsProxy = nullptr;
    ucom::ant::object_ptr<IStreamingRequest> m_pRequest;
 
    cSampleStreamBase* m_pFoldedInto = nullptr;
    bool m_bInitialTypesPending = false;
    ucom::ant::object_ptr<const IStreamType> m_pLastType;
    mutable std::mutex m_oTypeMutex;
    ucom::ant::weak_object_ptr<ISampleOutStream> m_pCurrentWriter;

    tSinkList m_oSinks;
    std::mutex m_oSinkMutex;
    tDirectChildren m_oDirectChildren;
 
    std::unique_ptr<cStatistics> m_pStatistics;
};
 
} // namespace detail

template<typename... INTERFACES>
class sample_stream :
    public ucom::object<INamedGraphObject,
                        trigger_pipe_item<hollow::IInternalBindingProxy,
                                          ISampleStream,
                                          vision::IBlockableSampleStream,
                                          wolverine::ISampleStreamStatistics,
                                          INTERFACES...>>,
    public detail::cSampleStreamBase
{
    using base_type = ucom::object<INamedGraphObject,
                                   trigger_pipe_item<hollow::IInternalBindingProxy,
                                                     ISampleStream,
                                                     vision::IBlockableSampleStream,
                                                     wolverine::ISampleStreamStatistics,
                                                     INTERFACES...>>;
 
public:
    sample_stream(): detail::cSampleStreamBase(static_cast<detail::cTriggerPipeItemBase&>(*this), *this)
    {
    }

    sample_stream(const char* strName): sample_stream()
    {
        base_type::SetName(strName);
    }
 
public: // implements ant::ISampleStream
    tResult GetType(ucom::ant::iobject_ptr<const IStreamType>& pStreamType) const noexcept override
    {
        return detail::cSampleStreamBase::GetType(pStreamType);
    }

    tTimeStamp GetTime() const noexcept override
    {
        return 0;
    }
 
    tResult SetStreamError(const tResult& oError) noexcept override
    {
        return detail::cSampleStreamBase::SetStreamError(oError);
    }
 
    tResult AttachRouting(const ucom::ant::iobject_ptr<ISampleStream>& pSampleStreamTo) noexcept override
    {
        return detail::cSampleStreamBase::AttachRouting(pSampleStreamTo);
    }
 
    tResult DetachRouting(const ucom::ant::iobject_ptr<ISampleStream>& pSampleStreamTo) noexcept override
    {
        return detail::cSampleStreamBase::DetachRouting(pSampleStreamTo);
    }

    tResult Open([[maybe_unused]] const char* strName,
                 ucom::ant::iobject_ptr<ISampleInStream>& pInStream,
                 [[maybe_unused]] const ucom::ant::iobject_ptr<const IStreamType>& pInitialAcceptedStreamType,
                 ISampleStream::IPushReadEventSink*& pPushEventSink,
                 ISampleStreamAccess::tMode ui32Mode,
                 size_t szQueueSize) noexcept override
    {
        return detail::cSampleStreamBase::Open(pInStream, pPushEventSink, ui32Mode, szQueueSize);
    }

    tResult Open([[maybe_unused]] const char* strName,
                 ucom::ant::iobject_ptr<ISampleOutStream>& pOutStream,
                 ISampleStreamAccess::tMode ui32Mode,
                 size_t szQueueSize) noexcept override
    {
        return detail::cSampleStreamBase::Open(pOutStream, ui32Mode, szQueueSize);
    }
 
public: // implements ant::ITriggerPipeSource
    tResult RegisterSubItem(const ucom::ant::iobject_ptr<ITriggerPipeItem>& pSubRun,
                            ITriggerPipeItem::tPriority ui32Prio) noexcept override
    {
        return detail::cSampleStreamBase::RegisterSubItem(pSubRun, ui32Prio);
    }
 
    tResult UnregisterSubItem(const ucom::ant::iobject_ptr<ITriggerPipeItem>& pSubRun) noexcept override
    {
        return detail::cSampleStreamBase::UnregisterSubItem(pSubRun);
    }

    tResult ChangePriority(const ucom::ant::iobject_ptr<ITriggerPipeItem>& /* pSubRun */,
                           ITriggerPipeItem::tPriority /* ui32Prio */) noexcept override
    {
        RETURN_ERROR(ERR_NOT_SUPPORTED);
    }
 
public: // implements hollow::IInternalBindingProxy
    tResult GetInternalBindingProxy(ucom::ant::iobject_ptr<IBindingProxy>& pBindingProxy) noexcept override
    {
        return detail::cSampleStreamBase::GetInternalBindingProxy(pBindingProxy);
    }
 
public: // implements vision::IBlockableSampleStream
    void BlockSampleStreaming(bool bBlock) noexcept override
    {
        detail::cSampleStreamBase::BlockSampleStreaming(bBlock);
    }
 
    bool IsSampleStreamingBlocked() const noexcept override
    {
        return detail::cSampleStreamBase::IsSampleStreamingBlocked();
    }
 
public: // implements wolverine::ISampleStreamStatistics
    wolverine::ISampleStreamStatistics::tSampleStreamStatistic GetStatistics() noexcept override
    {
        return detail::cSampleStreamBase::GetStatistics();
    }

    void EnableSampleSizeStatistics(bool bEnable) noexcept override
    {
        detail::cSampleStreamBase::EnableSampleSizeStatistics(bEnable);
    }

    void ResetStatistics() noexcept override
    {
        detail::cSampleStreamBase::ResetStatistics();
    }
};

class cSampleStream : public sample_stream<>
{
public:
    ADTF_CLASS_ID_NAME(cSampleStream, "sample_stream.streaming.adtf.cid", "Sample Stream");
 
public:
    using sample_stream<>::sample_stream;
};
 
} // namespace streaming
} // namespace adtf