serial::vita49::Vita49OutputSink Class Reference

#include "vita49_output_sink.h"

Inheritance diagram for serial::vita49::Vita49OutputSink:

Collaboration diagram for serial::vita49::Vita49OutputSink:

Public Member Functions
	Vita49OutputSink (std::unique_ptr< DatagramSender > sender=nullptr, core::ReceiverOutputTelemetryCallback telemetry_callback=nullptr)
	~Vita49OutputSink () override
	Vita49OutputSink (const Vita49OutputSink &)=delete
Vita49OutputSink &	operator= (const Vita49OutputSink &)=delete
	Vita49OutputSink (Vita49OutputSink &&)=delete
Vita49OutputSink &	operator= (Vita49OutputSink &&)=delete
void	initializeRun (const core::OutputConfig &config, std::string simulation_name) override
std::uint32_t	registerStream (const core::ReceiverStreamDescriptor &stream) override
void	openStream (std::uint32_t stream_id, RealType first_sample_time) override
void	submitBlock (const core::ReceiverSampleBlock &block) override
void	emitContextHeartbeat (RealType simulation_time) override
void	closeStream (std::uint32_t stream_id) override
core::OutputStats	finalize () override
core::OutputStats	snapshotStats () const override

Detailed Description

Definition at line 25 of file vita49_output_sink.h.

Constructor & Destructor Documentation

Vita49OutputSink() [1/3]

serial::vita49::Vita49OutputSink::Vita49OutputSink ( std::unique_ptr< DatagramSender >  sender = nullptr, core::ReceiverOutputTelemetryCallback  telemetry_callback = nullptr  )

explicit

Definition at line 52 of file vita49_output_sink.cpp.

                                                                                               :
        _telemetry_callback(std::move(telemetry_callback)), _provided_sender(std::move(sender))
    {
    }

~Vita49OutputSink()

serial::vita49::Vita49OutputSink::~Vita49OutputSink ( )

override

Definition at line 58 of file vita49_output_sink.cpp.

    {
        if (!_finalized)
        {
            try
            {
                (void)finalize();
            }
            catch (...)
            {
            }
        }
    }

References finalize(), and max.

Here is the call graph for this function:

Vita49OutputSink() [2/3]

serial::vita49::Vita49OutputSink::Vita49OutputSink ( const Vita49OutputSink &  )

delete

Vita49OutputSink() [3/3]

serial::vita49::Vita49OutputSink::Vita49OutputSink ( Vita49OutputSink &&  )

delete

Member Function Documentation

closeStream()

void serial::vita49::Vita49OutputSink::closeStream ( std::uint32_t  stream_id )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 197 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        emitTelemetry(consumeSenderDropsLocked(), false);
        auto& state = stateFor(stream_id);
        if (!state.closed)
        {
            emitContext(stream_id, state.last_context_time, false, true);
            state.closed = true;
            emitTelemetry({}, true);
        }
    }

References max.

emitContextHeartbeat()

void serial::vita49::Vita49OutputSink::emitContextHeartbeat ( RealType  simulation_time )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 184 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        emitTelemetry(consumeSenderDropsLocked(), false);
        for (auto& [stream_id, state] : _streams)
        {
            if (!state.closed && simulation_time - state.last_context_time >= 1.0)
            {
                emitContext(stream_id, simulation_time, false, false);
            }
        }
    }

References max.

finalize()

core::OutputStats serial::vita49::Vita49OutputSink::finalize ( )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 210 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        if (_finalized)
        {
            auto stats = snapshotStatsLocked();
            emitTelemetry({}, true);
            return stats;
        }
 
        if (_sender)
        {
            _sender->flush();
            emitTelemetry(consumeSenderDropsLocked(), false);
        }
 
        for (auto& [stream_id, state] : _streams)
        {
            if (!state.closed)
            {
                emitContext(stream_id, state.last_context_time, false, true);
                state.closed = true;
            }
        }
 
        if (_sender)
        {
            _sender->stop();
            emitTelemetry(consumeSenderDropsLocked(), false);
        }
 
        core::OutputStats stats{.mode = core::OutputMode::Vita49Udp,
                                .epoch_unix_nanoseconds = _packetizer
                                    ? std::optional<std::uint64_t>(_packetizer->epochUnixNanoseconds())
                                    : std::nullopt,
                                .streams = {}};
        for (auto& [stream_id, state] : _streams)
        {
            if (_sender)
            {
                state.stats.late_packet_count = _sender->latePacketCount(stream_id);
            }
            stats.streams.push_back(state.stats);
        }
        _finalized = true;
        emitTelemetry({}, true);
        return stats;
    }

References max, core::OutputStats::mode, and core::Vita49Udp.

Referenced by ~Vita49OutputSink().

Here is the caller graph for this function:

initializeRun()

void serial::vita49::Vita49OutputSink::initializeRun ( const core::OutputConfig &  config, std::string  simulation_name  )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 72 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        if (config.mode != core::OutputMode::Vita49Udp)
        {
            throw std::invalid_argument("Vita49OutputSink requires VITA output mode");
        }
        if (config.vita49.host.empty() || config.vita49.port == 0)
        {
            throw std::invalid_argument("VITA output requires destination host and port");
        }
        if (config.vita49.queue_depth == 0)
        {
            throw std::invalid_argument("VITA output queue depth must be positive");
        }
 
        _config = config;
        _simulation_name = std::move(simulation_name);
        const auto epoch_ns = config.vita49.epoch_unix_nanoseconds.value_or(defaultEpochNanoseconds());
        _packetizer =
            std::make_unique<Vita49Packetizer>(epoch_ns, config.vita49.adc_fullscale, config.vita49.max_udp_payload);
        _sender = std::make_unique<PacedSender>(
            _provided_sender ? std::move(_provided_sender) : std::make_unique<UdpSender>(), config.vita49.queue_depth);
        _sender->open(config.vita49.host, config.vita49.port);
        _sender->start(params::startTime());
        _last_stats_emit = std::chrono::steady_clock::time_point::min();
        _last_packet_trace_emit = std::chrono::steady_clock::now();
        _pending_packet_traces.clear();
        _trace_sequence = 0;
        _initialized = true;
        _finalized = false;
        emitTelemetry({}, true);
    }

References max, params::startTime(), and core::Vita49Udp.

Here is the call graph for this function:

openStream()

void serial::vita49::Vita49OutputSink::openStream ( std::uint32_t  stream_id, RealType  first_sample_time  )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 126 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        emitContext(stream_id, first_sample_time, true, false);
        stateFor(stream_id).opened = true;
        emitTelemetry({}, true);
    }

References max.

Referenced by submitBlock().

Here is the caller graph for this function:

operator=() [1/2]

Vita49OutputSink & serial::vita49::Vita49OutputSink::operator= ( const Vita49OutputSink &  )

delete

operator=() [2/2]

Vita49OutputSink & serial::vita49::Vita49OutputSink::operator= ( Vita49OutputSink &&  )

delete

registerStream()

std::uint32_t serial::vita49::Vita49OutputSink::registerStream ( const core::ReceiverStreamDescriptor &  stream )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 106 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        const auto stream_id = _registry.registerStream(stream);
        if (!_streams.contains(stream_id))
        {
            StreamState state;
            state.descriptor = stream;
            state.stats.receiver_id = stream.receiver_id;
            state.stats.receiver_name = stream.receiver_name;
            state.stats.stream_id = stream_id;
            state.stats.mode = stream.mode.empty() ? "unknown" : stream.mode;
            state.stats.sample_rate = stream.sample_rate;
            state.stats.reference_frequency = stream.reference_frequency;
            _streams.emplace(stream_id, std::move(state));
            emitTelemetry({}, true);
        }
        return stream_id;
    }

References max, core::ReceiverStreamDescriptor::mode, core::ReceiverStreamDescriptor::receiver_id, core::ReceiverStreamDescriptor::receiver_name, core::ReceiverStreamDescriptor::reference_frequency, serial::vita49::StreamRegistry::registerStream(), and core::ReceiverStreamDescriptor::sample_rate.

Referenced by submitBlock().

Here is the call graph for this function:

Here is the caller graph for this function:

snapshotStats()

core::OutputStats serial::vita49::Vita49OutputSink::snapshotStats ( ) const

overridevirtual

Reimplemented from core::ReceiverOutputSink.

Definition at line 259 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        return snapshotStatsLocked();
    }

References max.

submitBlock()

void serial::vita49::Vita49OutputSink::submitBlock ( const core::ReceiverSampleBlock &  block )

overridevirtual

Implements core::ReceiverOutputSink.

Definition at line 134 of file vita49_output_sink.cpp.

    {
        std::scoped_lock const lock(_mutex);
        if (!_initialized || !_packetizer)
        {
            throw std::logic_error("VITA output sink has not been initialized");
        }
        emitTelemetry(consumeSenderDropsLocked(), false);
        const auto stream_id = registerStream(block.stream);
        auto& state = stateFor(stream_id);
        if (!state.opened)
        {
            openStream(stream_id, block.first_sample_time);
        }
 
        const RealType sample_rate = block.sample_rate > 0.0 ? block.sample_rate : state.stats.sample_rate;
        auto result = _packetizer->packetize(block, stream_id, state.packet_counts, state.sample_loss_pending);
        state.sample_loss_pending = false;
        for (auto& packet : result.packets)
        {
            const auto packet_sample_count = packet.sample_count;
            const auto packet_first_sample_time = packet.first_sample_time;
            const auto packet_end_sample_time = sample_rate > 0.0
                ? packet_first_sample_time + static_cast<RealType>(packet_sample_count) / sample_rate
                : packet_first_sample_time;
            const auto packet_over_range = packet.over_range;
            const auto packet_timestamp = toCoreTimestamp(packet.timestamp);
            const auto packet_end_timestamp =
                tryCoreTimestampFromEpoch(_packetizer->epochUnixNanoseconds(), packet_end_sample_time);
            if (!enqueuePacket(std::move(packet)))
            {
                continue;
            }
            state.stats.samples_emitted += packet_sample_count;
            ++state.stats.packets_emitted;
            if (!state.stats.first_sample_time.has_value())
            {
                state.stats.first_sample_time = packet_first_sample_time;
                state.stats.first_timestamp = packet_timestamp;
            }
            state.stats.end_sample_time = packet_end_sample_time;
            state.stats.end_timestamp = packet_end_timestamp;
            if (packet_over_range)
            {
                state.over_range_pending = true;
            }
        }
        state.stats.over_range_count += result.over_range_count;
    }

References max, openStream(), and registerStream().

Here is the call graph for this function:

---

The documentation for this class was generated from the following files:

• packages/libfers/src/serial/vita49/vita49_output_sink.h
• packages/libfers/src/serial/vita49/vita49_output_sink.cpp