graph_object.h

Go to the documentation of this file.

#pragma once
#include "stream_type_helper.h"
#include "configurable_runner.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
{

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,
             typename CALLABLE = std::function<tResult(base::flash::tNanoSeconds)>,
             std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool> =
                 true>
    ReaderType* CreateInputPinWithCallback(const char* strName,
                                           const cStreamTypeHelper& oType,
                                           CALLABLE&& fnTriggerCallback,
                                           bool bForwardTriggerViaOutputPins = true);

    template<typename ReaderType = cPinReader,
             typename CALLABLE = std::function<tResult(const ucom::ant::iobject_ptr<const streaming::ant::ISample>&)>,
             std::enable_if_t<std::is_invocable_r_v<tResult,
                                                    std::decay_t<CALLABLE>,
                                                    const ucom::ant::iobject_ptr<const streaming::ant::ISample>&>,
                              bool> = true>
    ReaderType* CreateInputPinWithCallback(const char* strName,
                                           const cStreamTypeHelper& oType,
                                           CALLABLE&& fnSampleCallback,
                                           bool bForwardTriggerViaOutputPins = true);

    template<typename ReaderType = cPinReader,
             typename CALLABLE = std::function<tResult(base::flash::tNanoSeconds, const void*, size_t)>,
             std::enable_if_t<
                 std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds, const void*, size_t>,
                 bool> = true>
    ReaderType* CreateInputPinWithCallback(const char* strName,
                                           const cStreamTypeHelper& oType,
                                           CALLABLE&& fnSampleDataCallback,
                                           bool bForwardTriggerViaOutputPins = true);

    template<
        typename Type,
        typename ReaderType = cPinReader,
        typename CALLABLE = std::function<tResult(base::flash::tNanoSeconds, const Type&)>,
        std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds, const Type&>,
                         bool> = true>
    ReaderType* CreateInputPinWithCallback(const char* strName,
                                           const cStreamTypeHelper& oType,
                                           CALLABLE&& 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);

    template<typename RequestableWriterType = streaming::requestable_writer<>>
    RequestableWriterType*
        CreateRequestableOutputPin(const char* strName,
                                   const cStreamTypeHelper& oType = streaming::ant::stream_meta_type_anonymous());

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

    template<typename CALLABLE,
             std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool> =
                 true>
    streaming::ant::IRunner* CreateRunner(const char* strName,
                                          CALLABLE&& 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;

    template<typename CALLABLE,
             std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool> =
                 true>
    ucom::ant::object_ptr<streaming::ant::IRunner> RegisterRunner(const char* strName, CALLABLE&& fnRunFunction);

    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);

    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);
 
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 SetDescription(const char* strItem, const char* strDescription);

    void SetEditor(const char* strName, const char* strUrl);

    void SetHelpLink(const char* strUrl);
 
    tResult ActivatePins() override;
 
public:
    enum class tRecoverableAction : uint8_t
    {
        Log = 0, 
        Fatal = 1 
    };

    util::log::ILogChannel* CreateLogChannel(
        std::string_view strName,
        base::elasto::IErrorHandling::tAction eDefaultErrorAction = base::elasto::IErrorHandling::tAction::Stop,
        tRecoverableAction eDefaultRecoverableAction = tRecoverableAction::Log,
        util::log::tLogLevel eDefaultLogLevel = util::log::tLogLevel::Detail);

    util::log::ILogChannel* GetLogChannel(std::string_view strName = {});

    util::log::ILogChannel* a_util_get_current_log_channel() const noexcept;
 
    tResult SetName(const char* strName);
    tResult SetParent(const ucom::ant::IObject* oParentObject);
 
    ucom::object_ptr<services::ant::IMacroResolver> GetParentMacroResolver() const;
 
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);

    template<typename CALLABLE,
             std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool> =
                 true>
    void CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                        CALLABLE&& fnTriggerCallback,
                        bool bForwardTriggerViaOutputPins);

    void CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                        const char* strTriggerFunctionName,
                        bool bForwardTriggerViaOutputPins = true);

    void CreateOutputPin(std::shared_ptr<streaming::flash::ISampleWriter> pWriter);
 
    void ConfigureRunner(streaming::ant::IRunner* pRunner,
                         const cTriggerHint& oTriggerHint,
                         bool bForwardTriggerViaOutputPins);
 
    std::string GetFullName();
 
protected:
    ucom::ant::object_ptr<services::ant::IReferenceClock> _clock;
 
protected:
    class cImplementation;
    std::unique_ptr<cImplementation> m_pImplementation;
 
private:
    void LockedCallEnableSubStream(streaming::flash::ISampleWriter* pWriter,
                                   uint32_t nSubStreamId,
                                   const base::ant::IProperties* pRequestProperties);
    void LockedCallDisableSubStream(streaming::flash::ISampleWriter* pWriter, uint32_t nSubStreamId);
    std::recursive_mutex& GetProcessMutex();
};
 
template<typename CALLABLE,
         std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool>>
void cGraphObject::CreateInputPin(std::shared_ptr<streaming::flash::ISampleReader> pReader,
                                  CALLABLE&& fnTriggerCallback,
                                  bool bForwardTriggerViaOutputPins)
{
    std::string strName;
    pReader->GetName(base::adtf_string<std::string>(&strName));
    const auto strTriggerFunctionName = "process_" + strName;
    const auto pRunner = RegisterRunner(
        strTriggerFunctionName.c_str(),
        [fnTriggerCallback = std::forward<CALLABLE>(fnTriggerCallback), this](base::tNanoSeconds tmTrigger) -> tResult
        {
            std::lock_guard<std::recursive_mutex> oGuard(GetProcessMutex());
            RETURN_IF_FAILED(fnTriggerCallback(tmTrigger));
            RETURN_NOERROR;
        });
    CreateInputPin(pReader, strTriggerFunctionName.c_str(), bForwardTriggerViaOutputPins);
}
 
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,
         typename CALLABLE,
         std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool>>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                                     const cStreamTypeHelper& oType,
                                                     CALLABLE&& fnTriggerCallback,
                                                     bool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(pReader, std::forward<CALLABLE>(fnTriggerCallback), bForwardTriggerViaOutputPins);
    return pReader.get();
}
 
template<typename ReaderType,
         typename CALLABLE,
         std::enable_if_t<std::is_invocable_r_v<tResult,
                                                std::decay_t<CALLABLE>,
                                                const ucom::ant::iobject_ptr<const streaming::ant::ISample>&>,
                          bool>>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                                     const cStreamTypeHelper& oType,
                                                     CALLABLE&& fnSampleCallback,
                                                     bool bForwardTriggerViaOutputPins)
{
    auto pReader = CreateReader<ReaderType>(strName, oType);
    CreateInputPin(
        pReader,
        [fnSampleCallback = std::forward<CALLABLE>(fnSampleCallback), pReader](base::tNanoSeconds) -> tResult
        {
            ucom::object_ptr<const streaming::ISample> pSample;
            while (IS_OK(pReader->GetNextSample(pSample)))
            {
                RETURN_IF_FAILED_DESC(fnSampleCallback(pSample),
                                      "Error while processing sample with timestamp = %" PRIi64
                                      "ns, substream id = %" PRIu32 " and buffer size = %zu",
                                      streaming::get_sample_time(pSample).nCount,
                                      streaming::get_sample_substream_id(pSample),
                                      [&]() -> size_t
                                      {
                                          ucom::object_ptr_shared_locked<const streaming::ISampleBuffer> pBuffer;
                                          if (IS_OK(pSample->Lock(pBuffer)))
                                          {
                                              return pBuffer->GetSize();
                                          }
 
                                          return 0;
                                      }());
            }
 
            RETURN_NOERROR;
        },
        bForwardTriggerViaOutputPins);
    return pReader.get();
}
 
template<typename ReaderType,
         typename CALLABLE,
         std::enable_if_t<
             std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds, const void*, size_t>,
             bool>>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                                     const cStreamTypeHelper& oType,
                                                     CALLABLE&& fnSampleDataCallback,
                                                     bool bForwardTriggerViaOutputPins)
{
    return CreateInputPinWithCallback<ReaderType>(
        strName, oType,
        [fnSampleDataCallback = std::forward<CALLABLE>(fnSampleDataCallback)](
            const ucom::iobject_ptr<const streaming::ISample>& pSample) -> tResult
        {
            ucom::object_ptr_shared_locked<const streaming::ISampleBuffer> pBuffer;
            RETURN_IF_FAILED(pSample->Lock(pBuffer));
            return fnSampleDataCallback(streaming::get_sample_time(pSample), pBuffer->GetPtr(), pBuffer->GetSize());
        },
        bForwardTriggerViaOutputPins);
}
 
template<
    typename Type,
    typename ReaderType,
    typename CALLABLE,
    std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds, const Type&>,
                     bool>>
ReaderType* cGraphObject::CreateInputPinWithCallback(const char* strName,
                                                     const cStreamTypeHelper& oType,
                                                     CALLABLE&& fnDataCallback,
                                                     bool bForwardTriggerViaOutputPins)
{
    return CreateInputPinWithCallback<ReaderType>(
        strName, oType,
        [fnDataCallback = std::forward<CALLABLE>(fnDataCallback)](
            const ucom::ant::iobject_ptr<const streaming::ant::ISample>& pSample) -> tResult
        {
            streaming::sample_data<Type> oSampleData(pSample);
            return fnDataCallback(oSampleData.GetTimeNs(), *oSampleData);
        },
        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
        {
            util::log::cCurrentThreadDefaultLogChannel oLogger(a_util_get_current_log_channel());
            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;
}
 
template<typename CALLABLE,
         std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool>>
streaming::ant::IRunner* cGraphObject::CreateRunner(const char* strName,
                                                    CALLABLE&& fnRunFunction,
                                                    const cTriggerHint& oTriggerHint,
                                                    bool bForwardTriggerViaOutputPins)
{
    auto pRunner = RegisterRunner(strName, std::forward<CALLABLE>(fnRunFunction));
    ConfigureRunner(pRunner.Get(), oTriggerHint, bForwardTriggerViaOutputPins);
    return pRunner.Get();
}
 
template<typename CALLABLE,
         std::enable_if_t<std::is_invocable_r_v<tResult, std::decay_t<CALLABLE>, base::flash::tNanoSeconds>, bool>>
ucom::ant::object_ptr<streaming::ant::IRunner> cGraphObject::RegisterRunner(const char* strName,
                                                                            CALLABLE&& fnRunFunction)
{
    ucom::object_ptr<streaming::ant::IRunner> pRunner =
        ucom::ant::make_object_ptr<private_implementations::cConfigurableRunner>(
            strName, GetLogChannel(),
            [this, fnRunFunction = std::forward<CALLABLE>(fnRunFunction),
             strName = std::string(strName)](base::flash::tNanoSeconds tmTrigger) -> tResult
            {
                RETURN_IF_FAILED_DESC(fnRunFunction(tmTrigger),
                                      "Error while processing trigger with timestamp %" PRIi64 "ns on %s.%s",
                                      tmTrigger.nCount, GetFullName().c_str(), strName.c_str());
 
                RETURN_NOERROR;
            });
    THROW_IF_FAILED(RegisterRunner(pRunner));
    return pRunner;
}
 
template<typename RequestableWriterType>
RequestableWriterType* cGraphObject::CreateRequestableOutputPin(const char* strName, const cStreamTypeHelper& oType)
{
    auto pWriter = CreateWriter<RequestableWriterType>(strName, oType);
    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();
}
 
} // namespace filter
} // namespace adtf