requestable_writer< BaseWriter >

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

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. More...
tResult	SetType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType) override
	Sets the initial stream type of a streamer. More...
tResult	ChangeType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType) override
	Communicates a type change for all following samples. More...
tResult	ChangeType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType, bool bDisableCurrentRequests, bool bCallDisableCallback)
	Writes a new Stream Type to the Sample Stream. More...
tResult	ChangeType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType, bool bDisableCurrentRequests, bool bCallDisableCallback, bool bSync)
tResult	EndStreaming () override
	End streaming. More...
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. More...
tResult	BeginStreaming (streaming::ant::ISampleStream &oSampleStream) override
	Begin streaming on the given sample stream. More...
tResult	EndStreaming () override
	End streaming. More...
Public Member Functions inherited from cSampleWriter
void	SetName (const char *strName) override
	Sets the name of the streamer. More...
tResult	GetName (base::ant::IString &&strName) override
	Retrieves the name of the streamer. More...
tResult	SetType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pStreamType) override
	Sets the initial stream type of a streamer. More...
tResult	GetType (ucom::ant::iobject_ptr< const ant::IStreamType > &pStreamType) const override
	Returns the initial stream type of the streamer. More...
tResult	SetStreamerPin (const ucom::ant::iobject_ptr< IStreamerPin > &pStreamerPin) override
	Sets the pin that the streamer is associated with. More...
tResult	Write (const ucom::ant::iobject_ptr< const ant::ISample > &pSample) override
	Writes a sample to the connected sample stream. More...
virtual tResult	Transmit (const ucom::ant::iobject_ptr< const ant::ISample > &pSample)
	For compatibility purposes, call Write() instead. More...
tResult	ChangeType (const ucom::ant::iobject_ptr< const ant::IStreamType > &pType) override
	Communicates a type change for all following samples. More...
tResult	Flush () override
	Flushes the writer. More...
tResult	SetStreamError (tResult oError) override
	Reports and handles an error via the sample stream. More...
tResult	ManualTrigger (base::flash::tNanoSeconds tmActivationTime=base::flash::tNanoSeconds {-1}) override
	Calls all connected runners synchronously in this call. More...

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.

Member Function Documentation

ChangeType() [1/2]

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.

Implements ISampleWriter.

Definition at line 157 of file requestable_writer.h.

Referenced by requestable_writer< BaseWriter >::ChangeType().

ChangeType() [2/2]

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 requestable_writer< BaseWriter >::ChangeType(), RETURN_IF_FAILED, and RETURN_NOERROR.

EndStreaming()

tResult EndStreaming ( )

inlineoverridevirtual

End streaming.

Returns

Standard result.

Implements ISampleStreamer.

Definition at line 202 of file requestable_writer.h.

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

SetCallbacks()

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

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.

Implements ISampleStreamer.

Definition at line 151 of file requestable_writer.h.