Source Code for Demo Data Load Generator Plugin

Location

./src/examples/src/adtf/streaming_services/sources/data_load_generator/

Build Environment

To see how to set up the build environment have a look at ADTF CMake Environment

this implementation shows:

• how to implement a Streaming Source
• how to work with Stream Meta Type adtf::streaming::ant::stream_meta_type_anonymous
• how to use a adtf::system::kernel_timer
• how to work with a adtf::streaming::ant::cSampleWriter

Header

 
#pragma once
 
#include <adtffiltersdk/adtf_filtersdk.h>
#include <adtfsystemsdk/adtf_systemsdk.h>
#include <adtffiltersdk/runner_fallback.h>
#include <optional>
 
class cDataLoadGenerator: public adtf::filter::cSampleStreamingSource
{
    public:
        ADTF_CLASS_ID_NAME(cDataLoadGenerator,
                           "demo_data_load_generator.streaming_source.adtf.cid",
                           "Demo Data Load Generator");
 
        ADTF_CLASS_DEPENDENCIES(REQUIRE_INTERFACE(adtf::services::IReferenceClock),
                                REQUIRE_INTERFACE(adtf::services::IKernel));
 
    public:
        cDataLoadGenerator();
 
        tResult StartStreaming() override;
        tResult StopStreaming() override;
 
    private:
        tResult GenerateData();
 
    private:
        adtf::base::property_variable<double> m_fDataRate;
        adtf::base::property_variable<uint64_t> m_nDataInterval;
        adtf::ucom::object_ptr<adtf::services::IReferenceClock> m_pClock;
        adtf::filter::cRunnerFallback m_oTimer;
        adtf::streaming::ISampleWriter* m_pWriter = nullptr;
        std::optional<adtf::base::tNanoSeconds> m_tmLastTrigger;
};

Implementation

 
#include "demo_data_load_generator.h"
 
using namespace adtf::ucom;
using namespace adtf::streaming;
using namespace adtf::system;
 
ADTF_PLUGIN("Demo Data Load Generator Source Plugin",
            cDataLoadGenerator);
 
cDataLoadGenerator::cDataLoadGenerator()
{
    m_fDataRate.SetDescription("Count of bytes which should be generated per second.");
    RegisterPropertyVariable("data_rate_bytes_per_second", m_fDataRate);
 
    m_nDataInterval.SetDescription("Update interval in microseconds to generate random data. This is only used when no Timer Runner is connected to the 'generate_data' runner.");
    RegisterPropertyVariable("data_interval_us", m_nDataInterval);
 
    m_pWriter = CreateOutputPin("data");
    SetDescription("data", "Provides the randomly generated data");
 
    // For session compatibility reasons we use this fallback helper for the case where no active runner is connected to the runner.
    // In your own implementations use a simple CreateRunner(...) instead!
    m_oTimer = adtf::filter::cRunnerFallback(this, "generate_data", cTimerTriggerHint(*m_nDataInterval), [this]
    {
        auto oResult = GenerateData();
        if (IS_FAILED(oResult))
        {
            m_pWriter->SetStreamError(oResult);
        }
    });
    SetDescription("generate_data",
                   "Connect a Timer Runner that will trigger the data generation. In this case the property `data_interval_us` has no effect."
                   "If this is not connected, the source will create a timer on its own.");
 
    // sets a short description for the component
    SetDescription("Use this Streaming Source to generate random data load configured by 'data_rate_bytes_per_second' and 'data_interval_us' properties.");
 
    // set help link to jump to documentation from ADTF Configuration Editor
    SetHelpLink("$(ADTF_DIR)/doc/adtf_html/page_demo_data_load_generator.html");
}
 
tResult cDataLoadGenerator::StartStreaming()
{
    RETURN_IF_FAILED(cSampleStreamingSource::StartStreaming());
 
    RETURN_IF_FAILED(_runtime->GetObject(m_pClock));
 
    m_tmLastTrigger = std::nullopt;
    RETURN_IF_FAILED(m_oTimer.Activate(m_nDataInterval));
 
    RETURN_NOERROR;
}
 
tResult cDataLoadGenerator::GenerateData()
{
    auto tmNow = m_pClock->GetStreamTimeNs();
 
    if (m_tmLastTrigger)
    {
        const auto tmInterval = tmNow - *m_tmLastTrigger;
        auto nSampleSize = static_cast<size_t>(m_fDataRate * tmInterval.nCount / 1000000000);
        object_ptr<ISample> pSample;
        RETURN_IF_FAILED(alloc_sample(pSample, tmNow));
        {
            object_ptr_locked<ISampleBuffer> pBuffer;
            RETURN_IF_FAILED(pSample->WriteLock(pBuffer, nSampleSize));
        }
 
        m_pWriter->Write(pSample);
        m_pWriter->ManualTrigger(tmNow);
    }
 
    m_tmLastTrigger = tmNow;
    RETURN_NOERROR;
}
 
tResult cDataLoadGenerator::StopStreaming()
{
    m_oTimer.Deactivate();
    return cSampleStreamingSource::StopStreaming();
}