graph_object.h

Go to the documentation of this file.

 
#pragma once
#include "stream_type_helper.h"
 
#include <adtfbase/configuration.h>
#include <adtfstreaming3/samplereader.h>
#include <adtfstreaming3/samplewriter.h>
#include <adtfstreaming3/inpin_intf.h>
#include <adtfstreaming3/outpin_intf.h>
#include <adtfstreaming3/pin_creation.h>
#include <adtfstreaming3/samplestreamer_intf.h>
#include <adtfstreaming3/interface_binding_client_intf.h>
#include <adtfstreaming3/interface_binding_server_intf.h>
#include <adtfstreaming3/interface_binding_object.h>
#include <adtfstreaming3/data_binding.h>
#include <adtfstreaming3/interface_binding.h>
#include <adtfstreaming3/runtime_behaviour.h>
#include <adtfstreaming3/requestable_writer.h>
#include <adtfmediadescription/sample_codec.h>
#include <adtfmediadescription/media_description_type.h>
#include <adtfmediadescription/type_reflection.h>
#include <adtfmediadescription/md_sample_data.h>
#include <adtfstreaming3/streammetatypeanonymous.h>
 
#include <memory>
#include <functional>
#include <numeric>
 
namespace adtf
{
namespace filter
{
namespace flash
{
 
using cPinWriter = adtf::streaming::flash::cSampleWriter;
 
template <typename DATA_TYPE>
using pin_writer = adtf::streaming::flash::sample_writer<DATA_TYPE>;
 
using cPinReader = adtf::streaming::flash::cDynamicSampleReader;
 
template<tTimeStamp TIME_RANGE, bool STORELASTSAMPLE = true>
using time_limited_pin_reader = adtf::streaming::flash::time_limited_sample_reader<TIME_RANGE, STORELASTSAMPLE>;
 
template<size_t MAX_SIZE, bool STORELASTSAMPLE = true>
using size_limited_pin_reader = adtf::streaming::flash::size_limited_sample_reader<MAX_SIZE, STORELASTSAMPLE>;
 
template <typename Interface>
class interface_client
{
    public:
        interface_client() = default;
 
        interface_client(const ucom::ant::iobject_ptr<streaming::ant::IBindingClient>& pClient):
            m_pClient(pClient)
        {}
 
        interface_client(const interface_client&) = default;
        interface_client(interface_client&&) = default;
        interface_client& operator=(const interface_client&) = default;
        interface_client& operator=(interface_client&&) = default;
 
        bool IsValid()
        {
            if (m_pInterface)
            {
                return true;
            }
 
            if (!m_pClient ||
                IS_FAILED(m_pClient->GetServerObject(m_pInterface)))
            {
                return false;
            }
            m_pClient.Reset();
            return true;
        }
 
        Interface& Get() const
        {
            if (m_pInterface)
            {
                return *m_pInterface;
            }
 
            if (!m_pClient)
            {
                THROW_ERROR_DESC(ERR_NOT_INITIALIZED, "interface client has not been created");
            }
 
            auto nResult = m_pClient->GetServerObject(m_pInterface);
            if (IS_FAILED(nResult))
            {
                std::string strName;
                m_pClient->GetName(adtf_string_intf(strName));
                THROW_IF_FAILED_DESC(nResult,
                                     "Unable to aquire server object for client '%s' with interface '%s'.",
                                     strName.c_str(),
                                     adtf::ucom::get_iid<Interface>());
            }
 
            m_pClient.Reset();
            return *m_pInterface;
        }
 
        Interface* operator->() const
        {
            return &Get();
        }
 
    private:
        mutable ucom::ant::object_ptr<streaming::ant::IBindingClient> m_pClient;
        mutable ucom::ant::object_ptr<Interface> m_pInterface;
 
};
 
class cGraphObject:
        public streaming::flash::cNamedGraphObject,
        public streaming::ant::cDataBinding,
        public streaming::ant::cInterfaceBinding,
        public streaming::ant::cRuntimeBehaviour,
        public base::flash::cConfiguration
{
    public:
 
        class cTriggerHint
        {
            public:
                virtual ~cTriggerHint() = default;
        };
 
        class cNoTriggerHint: public cTriggerHint
        {
        };
 
        class cThreadTriggerHint: public cTriggerHint
        {
            public:
                cThreadTriggerHint(bool bCyclic = true):
                    m_bCyclic(bCyclic)
                {
                }
 
                bool GetCyclic() const
                {
                    return m_bCyclic;
                }
 
            private:
                bool m_bCyclic;
        };
 
        class cTimerTriggerHint: public cTriggerHint
        {
            public:
                cTimerTriggerHint(base::flash::tNanoSeconds tmInterval):
                    m_tmInterval(tmInterval)
                {
                }
 
                cTimerTriggerHint(tTimeStamp tmInterval):
                    m_tmInterval(base::flash::duration_cast<base::flash::tNanoSeconds>(tmInterval))
                {
                }
 
                base::flash::tNanoSeconds GetInterval() const
                {
                    return m_tmInterval;
                }
 
            private:
                base::flash::tNanoSeconds m_tmInterval;
        };
 
        class cDataTriggerHint: public cTriggerHint
        {
            public:
                cDataTriggerHint(const char* strInputName):
                    m_strInputName(strInputName)
                {
                }
 
                cDataTriggerHint(const std::vector<std::string>& lstInputNames):
                    m_strInputName(std::accumulate(lstInputNames.begin(),
                                                   lstInputNames.end(),
                                                   std::string{},
                                                   [](const auto& strPrevious, const auto& strNext) { return strPrevious.empty() ? strNext : strPrevious + ";" + strNext; }))
                {
                }
 
                const char* GetInputName() const
                {
                    return m_strInputName.c_str();
                }
 
            private:
                std::string m_strInputName;
        };
 
    public:
        cGraphObject();
 
        ~cGraphObject() override;
 
    public:
        virtual tResult Process(base::flash::tNanoSeconds tmTrigger,
                                streaming::ant::IRunner* pRunner);
 
        virtual tResult ProcessInput(base::flash::tNanoSeconds tmTrigger,
                                     streaming::flash::ISampleReader* pReader);
 
        virtual tResult ProcessInput(streaming::flash::ISampleReader* pReader,
                                     const ucom::ant::iobject_ptr<const streaming::ant::ISample>& pSample);
 
        virtual tResult ProcessInput(streaming::flash::ISampleReader* pReader,
                                     base::flash::tNanoSeconds tmSample,
                                     const void* pSampleData,
                                     size_t nSampleDataSize);
 
        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPin(const char* strName,
                                   bool bDataInTrigger = true,
                                   bool bForwardTriggerViaOutputPins = true);
 
        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPin(const char* strName,
                                   const cStreamTypeHelper& oType,
                                   bool bDataInTrigger = true,
                                   bool bForwardTriggerViaOutputPins = true);
 
        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const char* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(base::flash::tNanoSeconds)> fnTriggerCallback,
                                               bool bForwardTriggerViaOutputPins = true);
 
        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const char* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(const ucom::ant::iobject_ptr<const streaming::ant::ISample>&)> fnSampleCallback,
                                               bool bForwardTriggerViaOutputPins = true);
 
        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const char* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(base::flash::tNanoSeconds, const void*, size_t)> fnSampleDataCallback,
                                               bool bForwardTriggerViaOutputPins = true);
 
        template <typename Type, typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const char* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(base::flash::tNanoSeconds, const Type&)> fnDataCallback,
                                               bool bForwardTriggerViaOutputPins = true);
 
        virtual tResult AcceptType(streaming::flash::ISampleReader* pReader,
                                   const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType);
        template <typename WriterType = cPinWriter>
        WriterType* CreateOutputPin(const char* strName);
 
        template <typename WriterType = cPinWriter>
        WriterType* CreateOutputPin(const char* strName, const cStreamTypeHelper& oType);
 
        streaming::ant::IRunner* CreateRunner(const char* strName,
                                              const cTriggerHint& oTriggerHint = cNoTriggerHint(),
                                              bool bForwardTriggerViaOutputPins = true);
 
        streaming::ant::IRunner* CreateRunner(const char* strName,
                                              std::function<tResult(base::flash::tNanoSeconds)> fnRunFunction,
                                              const cTriggerHint& oTriggerHint = cNoTriggerHint(),
                                              bool bForwardTriggerViaOutputPins = true);
 
        template <typename Interface, typename Instance>
        void CreateInterfaceServer(const char* strName, ucom::ant::enable_object_ptr_from_this<Instance>* pInstance);
 
        template <typename Interface>
        void CreateInterfaceServer(const char* strName, const ucom::ant::iobject_ptr<Interface>& pInstance);
 
        template <typename Interface>
        interface_client<Interface> CreateInterfaceClient(const char* strName);
 
        template <typename ReaderType = cPinReader>
        std::shared_ptr<ReaderType> CreateReader(const char* strName,
                                                 const cStreamTypeHelper& oType);
 
        template <typename WriterType = cPinWriter>
        std::shared_ptr<WriterType> CreateWriter(const char* strName,
                                                 const cStreamTypeHelper& oType);
 
        using cRuntimeBehaviour::RegisterRunner;
 
        ucom::ant::object_ptr<streaming::ant::IRunner> RegisterRunner(const char* strName, std::function<tResult(base::flash::tNanoSeconds)> fnRunMethod);
 
        tResult RegisterRunner(const ucom::ant::iobject_ptr<streaming::ant::IRunner>& pRunner) override;
 
        tResult RegisterRunner(const char* strName, base::flash::IRunnable& oRunnable);
 
        tResult RegisterRunner(const char* strName, adtf::base::ant::IRunnable& oRunnable) override;
 
        tResult ConfigureDataInTrigger(const char* strRunnerName, const char* strPinName);
 
        tResult ConfigureDataOutTrigger(const char* strRunnerName, const char* strPinName);
 
        tResult ConfigureTimerTrigger(const char* strRunnerName, tTimeStamp tmPeriod);
 
        tResult ConfigureThreadTrigger(const char* strRunnerName, bool bCallCyclic);
 
    public:
        virtual tResult RequestDynamicInputPin(const char* strName,
                                               const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType);
 
        virtual tResult RequestDynamicOutputPin(const char* strName,
                                                const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType);
 
        virtual tResult RequestDynamicInterfaceClient(const char* strName,
                                                      const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType);
 
        virtual tResult RequestDynamicInterfaceServer(const char* strName,
                                                      const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType);
 
    public:
        tResult RequestPin(const char* strName,
                           const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType,
                           ucom::ant::iobject_ptr<streaming::ant::IInPin>& pInPin);
 
        tResult RequestPin(const char* strName,
                           const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType,
                           ucom::ant::iobject_ptr<streaming::ant::IOutPin>& pOutPin);
 
        tResult RequestBindingObject(const char* strName,
                                     const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType,
                                     ucom::ant::iobject_ptr<streaming::ant::IBindingClient>& pClient);
 
        tResult RequestBindingObject(const char* strName,
                                     const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType,
                                     ucom::ant::iobject_ptr<streaming::ant::IBindingServer>& pServer);
 
    public:
        void RegisterPropertyVariable(const char* strName,
                                       base::ant::cPropertyVariable& oPropertyVariable);
 
        void SetDescription(const char* strDescription);
 
        void SetEditor(const char* strName, const char* strUrl);    
 
    protected:
        void SetupStreamer(std::shared_ptr<streaming::flash::ISampleStreamer> pStreamer,
                            const char* strName,
                            const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType);
 
        void CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             bool bDataInTrigger = true,
                             bool bForwardTriggerViaOutputPins = true);
 
        void CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             std::function<tResult(base::flash::tNanoSeconds)> fnTriggerCallback,
                             bool bForwardTriggerViaOutputPins = true);
 
        void CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             std::function<tResult(const ucom::ant::iobject_ptr<const streaming::ant::ISample>&)> fnSampleCallback,
                             bool bForwardTriggerViaOutputPins = true);
 
        void CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             std::function<tResult(base::flash::tNanoSeconds, const void*, size_t)> fnSampleDataCallback,
                             bool bForwardTriggerViaOutputPins = true);
 
        void CreateOutputPin(std::shared_ptr<streaming::flash::ISampleWriter> pWriter);
 
    protected:
 
        ucom::ant::object_ptr<services::ant::IReferenceClock> _clock;
 
    protected:
        class cImplementation;
        std::unique_ptr<cImplementation> m_pImplementation;
};
 
template <typename ReaderType>
ReaderType* cGraphObject::CreateInputPin(const char* strName,
                                    bool bDataInTrigger,
                                    bool bForwardTriggerViaOutputPins)
{
    return CreateInputPin<ReaderType>(strName,
                                      streaming::stream_meta_type_anonymous(),
                                      bDataInTrigger, bForwardTriggerViaOutputPins);
}
 
template <typename ReaderType>
ReaderType* cGraphObject::CreateInputPin(const char* strName,
                                    const cStreamTypeHelper& oType,
                                    bool bDataInTrigger,
                                    bool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, bDataInTrigger, bForwardTriggerViaOutputPins);
    return pReader.get();
}
 
template <typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult(base::flash::tNanoSeconds)> fnTriggerCallback,
                                    bool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, fnTriggerCallback, bForwardTriggerViaOutputPins);
    return pReader.get();
}
 
template <typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult(const ucom::ant::iobject_ptr<const streaming::ant::ISample>&)> fnSampleCallback,
                                    bool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, fnSampleCallback, bForwardTriggerViaOutputPins);
    return pReader.get();
}
 
template<typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult (base::flash::tNanoSeconds, const void*, size_t)> fnSampleDataCallback,
                                    bool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, fnSampleDataCallback, bForwardTriggerViaOutputPins);
    return pReader.get();
}
 
 
template <typename Type, typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult(base::flash::tNanoSeconds, const Type&)> fnDataCallback,
                                    bool bForwardTriggerViaOutputPins)
{
    return CreateInputPinWithCallback<ReaderType>(strName, oType, [fnDataCallback](const ucom::ant::iobject_ptr<const streaming::ant::ISample>& pSample) -> tResult
    {
        ucom::object_ptr_shared_locked<const streaming::ant::ISampleBuffer> pBuffer;
        RETURN_IF_FAILED(pSample->Lock(pBuffer));
        if (pBuffer->GetSize() < sizeof(Type))
        {
            RETURN_ERROR_DESC(ERR_INVALID_ARG, "Invalid sample received, its too small for the requested type");
        }
        return fnDataCallback(streaming::get_sample_time(pSample), *static_cast<const Type*>(pBuffer->GetPtr()));
    },
    bForwardTriggerViaOutputPins);
}
 
template <typename WriterType>
WriterType* cGraphObject::CreateOutputPin(const char* strName)
{
    return CreateOutputPin<WriterType>(strName, cStreamTypeHelper(nullptr));
}
 
template <typename WriterType>
WriterType* cGraphObject::CreateOutputPin(const char* strName, const cStreamTypeHelper& oType)
{
    auto pWriter = CreateWriter<WriterType>(strName, oType);
    CreateOutputPin(pWriter);
    return pWriter.get();
}
 
template <typename Interface, typename Instance>
void cGraphObject::CreateInterfaceServer(const char* strName, ucom::ant::enable_object_ptr_from_this<Instance>* pInstance)
{
    auto pInstancePointer = pInstance->object_ptr_from_this();
    if (!pInstancePointer)
    {
        THROW_ERROR_DESC(ERR_NOT_INITIALIZED, "Unable to add interface server '%s', object_ptr_from_this has not been initialized. "
                                              "This AddInterfaceServer overload cannot be called from within the constructor.",
                         strName);
    }
    using namespace adtf::ucom::ant;
    using namespace adtf::streaming::ant;
    object_ptr<IBindingServer> pBindingServer = make_object_ptr<streaming::elasto::cWeakBindingServer>(strName,
                                                                                                       object_ptr<const IBindingType>(make_object_ptr<binding_type<Interface>>()),
                                                                                                       ucom_object_ptr_cast<IObject>(pInstancePointer));
    THROW_IF_FAILED(RegisterBindingObject(pBindingServer));
}
 
template <typename Interface>
void cGraphObject::CreateInterfaceServer(const char* strName, const ucom::ant::iobject_ptr<Interface>& pInstance)
{
    using namespace adtf::ucom::ant;
    using namespace adtf::streaming::ant;
    object_ptr<IBindingServer> pBindingServer = make_object_ptr<cBindingServer>(strName,
                                                                                object_ptr<const IBindingType>(make_object_ptr<binding_type<Interface>>()),
                                                                                ucom_object_ptr_cast<IObject>(pInstance));
    THROW_IF_FAILED(RegisterBindingObject(pBindingServer));
}
 
template <typename Interface>
interface_client<Interface> cGraphObject::CreateInterfaceClient(const char* strName)
{
    ucom::ant::object_ptr<streaming::ant::IBindingClient> pBindingClient =
            ucom::ant::make_object_ptr<streaming::ant::cBindingClient>(strName,
                                                                       ucom::ant::object_ptr<const streaming::ant::IBindingType>(ucom::ant::make_object_ptr<streaming::ant::binding_type<Interface>>()));
    THROW_IF_FAILED(RegisterBindingObject(pBindingClient));
    return pBindingClient;
}
 
template <typename ReaderType>
std::shared_ptr<ReaderType> cGraphObject::CreateReader(const char* strName,
                                                       const cStreamTypeHelper& oType)
{
    auto pReader = std::make_shared<ReaderType>();
    SetupStreamer(pReader, strName, oType.GetStreamType());
    pReader->SetAcceptTypeCallback([pReader = pReader.get(), this](const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType) -> tResult
    {
        return AcceptType(pReader, pType);
    });
    return pReader;
}
 
template <typename WriterType>
std::shared_ptr<WriterType> cGraphObject::CreateWriter(const char* strName,
                                                       const cStreamTypeHelper& oType)
{
    auto pWriter = std::make_shared<WriterType>();
    SetupStreamer(pWriter, strName, oType.GetStreamType());
    return pWriter;
}
 
}
 
namespace hollow
{
 
class cGraphObject: public flash::cGraphObject
{
    public:
        template <typename RequestableWriterType = streaming::requestable_writer<>>
        RequestableWriterType* CreateRequestableOutputPin(const char* strName,
                                                          const cStreamTypeHelper& oType = streaming::ant::stream_meta_type_anonymous());
 
        virtual tResult EnableSubStream(streaming::flash::ISampleWriter* pWriter,
                                        uint32_t nSubStreamId,
                                        const base::ant::IProperties* pRequestProperties = nullptr);
 
        virtual void DisableSubStream(streaming::flash::ISampleWriter* pWriter, uint32_t nSubStreamId);
 
        void SetHelpLink(const char* strUrl);
 
        void SetDescription(const char* strItem, const char* strDescription);
        using flash::cGraphObject::SetDescription;
 
    private:
        void LockedCallEnableSubStream(streaming::flash::ISampleWriter* pWriter,
                                        uint32_t nSubStreamId,
                                        const base::ant::IProperties* pRequestProperties);
        void LockedCallDisableSubStream(streaming::flash::ISampleWriter* pWriter, uint32_t nSubStreamId);
};
 
template <typename RequestableWriterType>
RequestableWriterType* cGraphObject::CreateRequestableOutputPin(const char* strName, const cStreamTypeHelper& oType)
{
    auto pWriter = CreateWriter<RequestableWriterType>(strName, oType);
    flash::cGraphObject::CreateOutputPin(pWriter);
    pWriter->SetCallbacks(std::bind(&cGraphObject::LockedCallEnableSubStream, this, pWriter.get(), std::placeholders::_1, std::placeholders::_2),
                          std::bind(&cGraphObject::LockedCallDisableSubStream, this, pWriter.get(), std::placeholders::_1));
    return pWriter.get();
}
 
}
 
using flash::cPinReader;
using flash::cPinWriter;
using flash::pin_writer;
using flash::size_limited_pin_reader;
using flash::time_limited_pin_reader;
using flash::interface_client;
 
} //namespace filter
} //namespace adtf