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>

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, tBool STORELASTSAMPLE = true>
using time_limited_pin_reader = adtf::streaming::flash::time_limited_sample_reader<TIME_RANGE, STORELASTSAMPLE>;

template<tSize MAX_SIZE, tBool 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;

        tBool 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(tBool bCyclic = true):
                    m_bCyclic(bCyclic)
                {
                }

                tBool GetCyclic() const
                {
                    return m_bCyclic;
                }

            private:
                tBool 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 tChar* strInputName):
                    m_strInputName(strInputName)
                {
                }

                const tChar* 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 tVoid* pSampleData,
                                     tSize nSampleDataSize);

        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPin(const tChar* strName,
                                   tBool bDataInTrigger = true,
                                   tBool bForwardTriggerViaOutputPins = true);

        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPin(const tChar* strName,
                                   const cStreamTypeHelper& oType,
                                   tBool bDataInTrigger = true,
                                   tBool bForwardTriggerViaOutputPins = true);

        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const tChar* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(base::flash::tNanoSeconds)> fnTriggerCallback,
                                               tBool bForwardTriggerViaOutputPins = true);

        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const tChar* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(const ucom::ant::iobject_ptr<const streaming::ant::ISample>&)> fnSampleCallback,
                                               tBool bForwardTriggerViaOutputPins = true);

        template <typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const tChar* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(base::flash::tNanoSeconds, const tVoid*, tSize)> fnSampleDataCallback,
                                               tBool bForwardTriggerViaOutputPins = true);

        template <typename Type, typename ReaderType = cPinReader>
        ReaderType* CreateInputPinWithCallback(const tChar* strName,
                                               const cStreamTypeHelper& oType,
                                               std::function<tResult(base::flash::tNanoSeconds, const Type&)> fnDataCallback,
                                               tBool 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 tChar* strName);

        template <typename WriterType = cPinWriter>
        WriterType* CreateOutputPin(const tChar* strName, const cStreamTypeHelper& oType);

        streaming::ant::IRunner* CreateRunner(const tChar* strName,
                                              const cTriggerHint& oTriggerHint = cNoTriggerHint(),
                                              tBool bForwardTriggerViaOutputPins = true);

        streaming::ant::IRunner* CreateRunner(const tChar* strName,
                                              std::function<tResult(base::flash::tNanoSeconds)> fnRunFunction,
                                              const cTriggerHint& oTriggerHint = cNoTriggerHint(),
                                              tBool bForwardTriggerViaOutputPins = true);

        template <typename Interface, typename Instance>
        tVoid CreateInterfaceServer(const tChar* strName, ucom::ant::enable_object_ptr_from_this<Instance>* pInstance);

        template <typename Interface>
        tVoid CreateInterfaceServer(const tChar* strName, const ucom::ant::iobject_ptr<Interface>& pInstance);

        template <typename Interface>
        interface_client<Interface> CreateInterfaceClient(const tChar* strName);

        template <typename ReaderType = cPinReader>
        std::shared_ptr<ReaderType> CreateReader(const tChar* strName,
                                                 const cStreamTypeHelper& oType);

        template <typename WriterType = cPinWriter>
        std::shared_ptr<WriterType> CreateWriter(const tChar* strName,
                                                 const cStreamTypeHelper& oType);

        using cRuntimeBehaviour::RegisterRunner;

        ucom::ant::object_ptr<streaming::ant::IRunner> RegisterRunner(const tChar* strName, std::function<tResult(base::flash::tNanoSeconds)> fnRunMethod);

        tResult RegisterRunner(const ucom::ant::iobject_ptr<streaming::ant::IRunner>& pRunner) override;

        tResult RegisterRunner(const tChar* strName, base::flash::IRunnable& oRunnable);

        tResult RegisterRunner(const tChar* strName, adtf::base::ant::IRunnable& oRunnable) override;

        tResult ConfigureDataInTrigger(const tChar* strRunnerName, const tChar* strPinName);

        tResult ConfigureDataOutTrigger(const tChar* strRunnerName, const tChar* strPinName);

        tResult ConfigureTimerTrigger(const tChar* strRunnerName, tTimeStamp tmPeriod);

        tResult ConfigureThreadTrigger(const tChar* strRunnerName, tBool bCallCyclic);

    public:
        virtual tResult RequestDynamicInputPin(const tChar* strName,
                                               const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType);

        virtual tResult RequestDynamicOutputPin(const tChar* strName,
                                                const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType);

        virtual tResult RequestDynamicInterfaceClient(const tChar* strName,
                                                      const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType);

        virtual tResult RequestDynamicInterfaceServer(const tChar* strName,
                                                      const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType);

    public:
        tResult RequestPin(const tChar* strName,
                           const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType,
                           ucom::ant::iobject_ptr<streaming::ant::IInPin>& pInPin);

        tResult RequestPin(const tChar* strName,
                           const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType,
                           ucom::ant::iobject_ptr<streaming::ant::IOutPin>& pOutPin);

        tResult RequestBindingObject(const tChar* strName,
                                     const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType,
                                     ucom::ant::iobject_ptr<streaming::ant::IBindingClient>& pClient);

        tResult RequestBindingObject(const tChar* strName,
                                     const ucom::ant::iobject_ptr<const streaming::ant::IBindingType>& pType,
                                     ucom::ant::iobject_ptr<streaming::ant::IBindingServer>& pServer);

    public:
        tVoid RegisterPropertyVariable(const tChar* strName,
                                       base::ant::cPropertyVariable& oPropertyVariable);

        tVoid SetDescription(const tChar* strDescription);

        tVoid SetEditor(const tChar* strName, const tChar* strUrl);    

    protected:
        tVoid SetupStreamer(std::shared_ptr<streaming::flash::ISampleStreamer> pStreamer,
                            const tChar* strName,
                            const ucom::ant::iobject_ptr<const streaming::ant::IStreamType>& pType);

        tVoid CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             tBool bDataInTrigger = true,
                             tBool bForwardTriggerViaOutputPins = true);

        tVoid CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             std::function<tResult(base::flash::tNanoSeconds)> fnTriggerCallback,
                             tBool bForwardTriggerViaOutputPins = true);

        tVoid CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             std::function<tResult(const ucom::ant::iobject_ptr<const streaming::ant::ISample>&)> fnSampleCallback,
                             tBool bForwardTriggerViaOutputPins = true);

        tVoid CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                             std::function<tResult(base::flash::tNanoSeconds, const tVoid*, tSize)> fnSampleDataCallback,
                             tBool bForwardTriggerViaOutputPins = true);

        tVoid 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 tChar* strName,
                                    tBool bDataInTrigger,
                                    tBool bForwardTriggerViaOutputPins)
{
    return CreateInputPin<ReaderType>(strName,
                                      streaming::stream_meta_type_anonymous(),
                                      bDataInTrigger, bForwardTriggerViaOutputPins);
}

template <typename ReaderType>
ReaderType* cGraphObject::CreateInputPin(const tChar* strName,
                                    const cStreamTypeHelper& oType,
                                    tBool bDataInTrigger,
                                    tBool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, bDataInTrigger, bForwardTriggerViaOutputPins);
    return pReader.get();
}

template <typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const tChar* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult(base::flash::tNanoSeconds)> fnTriggerCallback,
                                    tBool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, fnTriggerCallback, bForwardTriggerViaOutputPins);
    return pReader.get();
}

template <typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const tChar* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult(const ucom::ant::iobject_ptr<const streaming::ant::ISample>&)> fnSampleCallback,
                                    tBool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, fnSampleCallback, bForwardTriggerViaOutputPins);
    return pReader.get();
}

template<typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const tChar* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult (base::flash::tNanoSeconds, const tVoid*, tSize)> fnSampleDataCallback,
                                    tBool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, fnSampleDataCallback, bForwardTriggerViaOutputPins);
    return pReader.get();
}


template <typename Type, typename ReaderType>
ReaderType* cGraphObject::CreateInputPinWithCallback(const tChar* strName,
                                    const cStreamTypeHelper& oType,
                                    std::function<tResult(base::flash::tNanoSeconds, const Type&)> fnDataCallback,
                                    tBool 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 tChar* strName)
{
    return CreateOutputPin<WriterType>(strName, cStreamTypeHelper(nullptr));
}

template <typename WriterType>
WriterType* cGraphObject::CreateOutputPin(const tChar* strName, const cStreamTypeHelper& oType)
{
    auto pWriter = CreateWriter<WriterType>(strName, oType);
    CreateOutputPin(pWriter);
    return pWriter.get();
}

template <typename Interface, typename Instance>
tVoid cGraphObject::CreateInterfaceServer(const tChar* 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>
tVoid cGraphObject::CreateInterfaceServer(const tChar* 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 tChar* 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 tChar* 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 tChar* 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 tChar* strName,
                                                          const cStreamTypeHelper& oType = streaming::ant::stream_meta_type_anonymous());

        virtual tResult EnableSubStream(streaming::flash::ISampleWriter* pWriter,
                                        tUInt32 nSubStreamId,
                                        const base::ant::IProperties* pRequestProperties = nullptr);

        virtual tVoid DisableSubStream(streaming::flash::ISampleWriter* pWriter, tUInt32 nSubStreamId);

        tVoid SetHelpLink(const tChar* strUrl);

        tVoid SetDescription(const tChar* strItem, const tChar* strDescription);
        using flash::cGraphObject::SetDescription;

    private:
        tVoid LockedCallEnableSubStream(streaming::flash::ISampleWriter* pWriter,
                                        tUInt32 nSubStreamId,
                                        const base::ant::IProperties* pRequestProperties);
        tVoid LockedCallDisableSubStream(streaming::flash::ISampleWriter* pWriter, tUInt32 nSubStreamId);
};

template <typename RequestableWriterType>
RequestableWriterType* cGraphObject::CreateRequestableOutputPin(const tChar* 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