requestable_writer< BaseWriter >

Writer that registers a request broker (cRequestBroker) at the connected sample stream. More...

#include <requestable_writer.h>

Inheritance diagram for requestable_writer< BaseWriter >:

Public Member Functions
	requestable_writer ()
	Default constructor.
void	SetCallbacks (std::function< void(uint32_t, const base::IProperties *)> fnEnable, std::function< void(uint32_t)> fnDisable)
	Sets the callbacks for activating and deactivating substreams.
tResult	SetType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType) override
	Sets the initial stream type of a streamer.
tResult	ChangeType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType) override
	Communicates a type change for all following samples.
tResult	ChangeType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType, bool bDisableCurrentRequests, bool bCallDisableCallback)
	Writes a new Stream Type to the Sample Stream.
tResult	ChangeType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType, bool bDisableCurrentRequests, bool bCallDisableCallback, bool bSync)
tResult	EndStreaming () override
	End streaming.
Public Member Functions inherited from writer_with_request_broker< adtf::streaming::cSampleWriter >
void	SetBroker (const ucom::ant::iobject_ptr< IStreamingRequests > &pBroker)
	Use this to set the broker instance after the writer has been created.
tResult	BeginStreaming (streaming::ant::ISampleStream &oSampleStream) override
	Begin streaming on the given sample stream.
tResult	EndStreaming () override
	End streaming.
Public Member Functions inherited from cSampleWriter
void	SetName (const char *strName) override
	Sets the name of the streamer.
tResult	GetName (base::ant::IString &&strName) override
	Retrieves the name of the streamer.
tResult	GetType (ucom::ant::iobject_ptr< const ant::IStreamType > &pStreamType) const override
	Returns the initial stream type of the streamer.
tResult	SetStreamerPin (const ucom::ant::iobject_ptr< IStreamerPin > &pStreamerPin) override
	Sets the pin that the streamer is associated with.
tResult	Write (const ucom::ant::iobject_ptr< const ant::ISample > &pSample) override
	Writes a sample to the connected sample stream.
virtual tResult	Transmit (const ucom::ant::iobject_ptr< const ant::ISample > &pSample)
	For compatibility purposes, call Write() instead.
tResult	Flush () override
	Flushes the writer.
tResult	SetStreamError (tResult oError) override
	Reports and handles an error via the sample stream.
tResult	ManualTrigger (base::flash::tNanoSeconds tmActivationTime=base::flash::tNanoSeconds {-1}) override
	Calls all connected runners synchronously in this call.

Private Attributes
ucom::ant::object_ptr< cRequestBroker >	m_pBroker

Additional Inherited Members
Protected Attributes inherited from cSampleWriter
std::unique_ptr< cImplementation >	m_pImplementation

Detailed Description

template<typename BaseWriter = adtf::streaming::cSampleWriter>
class adtf::streaming::hollow::requestable_writer< BaseWriter >

Writer that registers a request broker (cRequestBroker) at the connected sample stream.

Example Usage

#include <adtffiltersdk/adtf_filtersdk.h>
#include <set>
#include <mutex>
 
class cMyFilter: public adtf::filter::cFilter
{
    public:
        ADTF_CLASS_ID_NAME(cMyFilter, "my_filter.filter.example.cid", "My Filter");
 
        cMyFilter()
        {
            CreateRunner("generate_samples");
 
            adtf::filter::cSubStreamTypes oSubStreams;
            oSubStreams.SetSubStream("stream1", 0, adtf::streaming::stream_type_plain<uint32_t>());
            oSubStreams.SetSubStream("stream2", 1, adtf::streaming::stream_type_plain<uint32_t>());
            oSubStreams.SetSubStream("stream3", 2, adtf::streaming::stream_type_plain<uint32_t>());
 
            m_pWriter = CreateOutputPin<adtf::streaming::requestable_writer<>>("output", oSubStreams);
            m_pWriter->SetCallbacks([this](uint32_t nSubStreamId, const adtf::base::IProperties*)
                                    {
                                        if (nSubStreamId > 2)
                                        {
                                            THROW_ERROR_DESC(ERR_INVALID_ARG, "No substream with id %d available", nSubStreamId);
                                        }
                                        std::lock_guard<std::mutex> oGuard(m_oSubStreamsLock);
                                        m_oEnabledSubStreams.insert(nSubStreamId);
                                    },
                                    [this](uint32_t nSubStreamId)
                                    {
                                        std::lock_guard<std::mutex> oGuard(m_oSubStreamsLock);
                                        m_oEnabledSubStreams.erase(nSubStreamId);
                                    });
        }       
 
        tResult Process(adtf::base::tNanoSeconds tmTrigger, adtf::streaming::IRunner* /*pRunner*/) override
        {
            ++m_nTriggerCounter;
 
            std::lock_guard<std::mutex> oGuard(m_oSubStreamsLock);
            for (const auto& nSubStreamId: m_oEnabledSubStreams)
            {
                adtf::streaming::output_sample_data<uint32_t> oNewData(tmTrigger, m_nTriggerCounter, nSubStreamId);
                m_pWriter->Write(oNewData.Release());
            }
 
            RETURN_NOERROR;
        }
 
    private:
        adtf::streaming::requestable_writer<>* m_pWriter = nullptr;
        std::mutex m_oSubStreamsLock;
        std::set<uint32_t> m_oEnabledSubStreams;
        uint32_t m_nTriggerCounter = 0;
};

Definition at line 125 of file requestable_writer.h.

Constructor & Destructor Documentation

requestable_writer()

template<typename BaseWriter = adtf::streaming::cSampleWriter>

requestable_writer ( )

inline

Default constructor.

Definition at line 131 of file requestable_writer.h.

References writer_with_request_broker< BaseWriter >::SetBroker(), and adtf::ucom::ucom_object_ptr_cast().

Member Function Documentation

ChangeType() [1/3]

template<typename BaseWriter = adtf::streaming::cSampleWriter>

tResult ChangeType ( const ucom::ant::iobject_ptr< const ant::IStreamType > & pType )

inlineoverridevirtual

Communicates a type change for all following samples.

Parameters

[in]

pType

The new stream type

Returns

Standard result. Errors will be reported via the associated sample stream.

Reimplemented from cSampleWriter.

Definition at line 157 of file requestable_writer.h.

References ChangeType().

Referenced by ChangeType(), and ChangeType().

ChangeType() [2/3]

template<typename BaseWriter = adtf::streaming::cSampleWriter>

tResult ChangeType ( const ucom::ant::iobject_ptr< const ant::IStreamType > & pType, bool bDisableCurrentRequests, bool bCallDisableCallback )

inline

Writes a new Stream Type to the Sample Stream.

Parameters

[in]	pType	The new Stream Type.
[in]	bDisableCurrentRequests	If true, current requests will be disabled.
[in]	bCallDisableCallback	If bDisableCurrentRequests was true, whether or not to call the disable callback for all ongoing requests.

Returns

Standard result.

Definition at line 171 of file requestable_writer.h.

References ChangeType(), RETURN_IF_FAILED, and RETURN_NOERROR.

ChangeType() [3/3]

template<typename BaseWriter = adtf::streaming::cSampleWriter>

tResult ChangeType ( const ucom::ant::iobject_ptr< const ant::IStreamType > & pType, bool bDisableCurrentRequests, bool bCallDisableCallback, bool bSync )

inline

Definition at line 180 of file requestable_writer.h.

EndStreaming()

template<typename BaseWriter = adtf::streaming::cSampleWriter>

tResult EndStreaming ( )

inlineoverridevirtual

End streaming.

Returns

Standard result.

Reimplemented from cSampleWriter.

Definition at line 202 of file requestable_writer.h.

References writer_with_request_broker< BaseWriter >::EndStreaming(), and RETURN_NOERROR.

SetCallbacks()

template<typename BaseWriter = adtf::streaming::cSampleWriter>

void SetCallbacks ( std::function< void(uint32_t, const base::IProperties *)> fnEnable, std::function< void(uint32_t)> fnDisable )

inline

Sets the callbacks for activating and deactivating substreams.

Example Usage

#include <adtffiltersdk/adtf_filtersdk.h>
#include <set>
#include <mutex>
 
class cMyFilter: public adtf::filter::cFilter
{
    public:
        ADTF_CLASS_ID_NAME(cMyFilter, "my_filter.filter.example.cid", "My Filter");
 
        cMyFilter()
        {
            CreateRunner("generate_samples");
 
            adtf::filter::cSubStreamTypes oSubStreams;
            oSubStreams.SetSubStream("stream1", 0, adtf::streaming::stream_type_plain<uint32_t>());
            oSubStreams.SetSubStream("stream2", 1, adtf::streaming::stream_type_plain<uint32_t>());
            oSubStreams.SetSubStream("stream3", 2, adtf::streaming::stream_type_plain<uint32_t>());
 
            m_pWriter = CreateOutputPin<adtf::streaming::requestable_writer<>>("output", oSubStreams);
            m_pWriter->SetCallbacks([this](uint32_t nSubStreamId, const adtf::base::IProperties*)
                                    {
                                        if (nSubStreamId > 2)
                                        {
                                            THROW_ERROR_DESC(ERR_INVALID_ARG, "No substream with id %d available", nSubStreamId);
                                        }
                                        std::lock_guard<std::mutex> oGuard(m_oSubStreamsLock);
                                        m_oEnabledSubStreams.insert(nSubStreamId);
                                    },
                                    [this](uint32_t nSubStreamId)
                                    {
                                        std::lock_guard<std::mutex> oGuard(m_oSubStreamsLock);
                                        m_oEnabledSubStreams.erase(nSubStreamId);
                                    });
        }       
 
        tResult Process(adtf::base::tNanoSeconds tmTrigger, adtf::streaming::IRunner* /*pRunner*/) override
        {
            ++m_nTriggerCounter;
 
            std::lock_guard<std::mutex> oGuard(m_oSubStreamsLock);
            for (const auto& nSubStreamId: m_oEnabledSubStreams)
            {
                adtf::streaming::output_sample_data<uint32_t> oNewData(tmTrigger, m_nTriggerCounter, nSubStreamId);
                m_pWriter->Write(oNewData.Release());
            }
 
            RETURN_NOERROR;
        }
 
    private:
        adtf::streaming::requestable_writer<>* m_pWriter = nullptr;
        std::mutex m_oSubStreamsLock;
        std::set<uint32_t> m_oEnabledSubStreams;
        uint32_t m_nTriggerCounter = 0;
};

Parameters

[in,out]	fnEnable	Called when the first consumer request samples of a substream.
[in,out]	fnDisable	Called when the last consumer cancels its request for samples of a substream.

Definition at line 145 of file requestable_writer.h.

SetType()

template<typename BaseWriter = adtf::streaming::cSampleWriter>

tResult SetType ( const ucom::ant::iobject_ptr< const ant::IStreamType > & pStreamType )

inlineoverridevirtual

Sets the initial stream type of a streamer.

Parameters

[in]

pStreamType

The stream type

Returns

Standard result.

Reimplemented from cSampleWriter.

Definition at line 151 of file requestable_writer.h.

Member Data Documentation

m_pBroker

template<typename BaseWriter = adtf::streaming::cSampleWriter>

ucom::ant::object_ptr<cRequestBroker> m_pBroker

private

Definition at line 210 of file requestable_writer.h.