dvmtrstream.cc

// SPDX-License-Identifier: GPL-2.0-only
/*
 * Digital Voice Modem - Trunk Recorder Stream Plugin
 * GPLv2 Open Source. Use is subject to license terms.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 *  Copyright (C) 2025,2026 Patrick McDonnell, W3AXL
 *  Copyright (C) 2026 Bryan Biedenkapp, N2PLL
 *  Copyright (C) 2026 C. Lovell, K7CBL
 *
 */
#include "../../trunk-recorder/plugin_manager/plugin_api.h"
#include "../../trunk-recorder/recorders/recorder.h"
#include <boost/array.hpp>
#include <boost/asio.hpp>
#include <boost/dll/alias.hpp> // for BOOST_DLL_ALIAS
#include <boost/foreach.hpp>
#include <boost/endian/conversion.hpp>
#include <queue>
#include <mutex>
#include <thread>
#include <chrono>
#include <memory>
#include <cstring>
#include <vector>

using namespace boost::asio;

typedef struct plugin_t plugin_t;
typedef struct stream_t stream_t;

// ---------------------------------------------------------------------------
//  Macros
// ---------------------------------------------------------------------------

/**
 * @brief Sets a uint32_t into 4 bytes of a buffer/array. (32-bit value).
 * @ingroup common
 * @param val uint32_t value to set
 * @param buffer uint8_t buffer to set value on
 * @param offset Offset within uint8_t buffer
 */
#define SET_UINT32(val, buffer, offset)                 \
            buffer[0U + offset] = (val >> 24) & 0xFFU;  \
            buffer[1U + offset] = (val >> 16) & 0xFFU;  \
            buffer[2U + offset] = (val >> 8) & 0xFFU;   \
            buffer[3U + offset] = (val >> 0) & 0xFFU;

#ifdef DVMTRSTREAM_ANALOG_SUPPORT
/**
 * @brief Downsamples Trunk Recorder conventional analog plugin audio.
 *
 * Trunk Recorder sends analog plugin audio at 16 kHz while digital audio is
 * sent at 8 kHz. DVMBridge expects the existing 8 kHz mono signed PCM stream,
 * so the analog build decimates to 8 kHz before using the normal send path.
 */
std::vector<int16_t> downsample16kTo8k(const int16_t* samples, int sampleCount) {
    std::vector<int16_t> downsampled;

    if (!samples || sampleCount <= 0) {
        return downsampled;
    }

    downsampled.reserve((sampleCount + 1) / 2);

    for (int i = 0; i + 1 < sampleCount; i += 2) {
        int32_t mixed = static_cast<int32_t>(samples[i]) + static_cast<int32_t>(samples[i + 1]);
        downsampled.push_back(static_cast<int16_t>(mixed / 2));
    }

    if ((sampleCount % 2) != 0) {
        downsampled.push_back(samples[sampleCount - 1]);
    }

    return downsampled;
}
#endif

// ---------------------------------------------------------------------------
//  Structure Declaration
// ---------------------------------------------------------------------------

struct plugin_t {
    Config *config;
};

/**
 * @brief Represents an individual stream audio chunk.
 */

struct audio_chunk_t {
    std::vector<uint8_t> data;
    int32_t tgid;
    int32_t src;
    
    audio_chunk_t(const uint8_t* chunk_data, size_t size, int32_t tg, int32_t source) 
        : data(chunk_data, chunk_data + size), tgid(tg), src(source) {}
};

/**
 * @brief Represents an individual call audio stream.
 */

struct stream_t {
    long TGID;
    std::string address;
    std::string short_name;
    long port;
    ip::udp::endpoint remote_endpoint;
    std::queue<audio_chunk_t> chunk_queue;
    std::mutex queue_mutex;
    std::chrono::steady_clock::time_point last_enqueue_time{};
    std::chrono::steady_clock::time_point last_send_time{};
    bool call_active = false;
    bool silence_leader_injected = false;  // Track if silence leader has been added
    int32_t last_call_tgid = -1;  // Last talkgroup ID
    int32_t last_call_src = -1;   // Last source ID
#ifdef DVMTRSTREAM_ANALOG_SUPPORT
    std::vector<uint8_t> pending_audio;
    int32_t pending_audio_tgid = -1;
    int32_t pending_audio_src = -1;
    std::chrono::steady_clock::time_point pending_audio_updated;
#endif
};

std::vector<std::shared_ptr<stream_t>> streams;

/**
 * @brief Represents an dvmbridge endpoint group.
 */

 struct endpoint_group_t {
    std::string endpoint_key;  // "address:port"
    std::vector<std::shared_ptr<stream_t>> streams;
    std::shared_ptr<boost::asio::steady_timer> timer;
    size_t current_stream_index = 0;
    bool timer_active = false;
    int inter_stream_delay_ms = 0;  // Delay between switching streams (ms)
    bool switching_stream = false;  // True when in delay period between streams
    int silence_leader_ms = 0;  // Silence to inject before stream audio (ms)
    bool strict_call_serialization = true;
    int call_gap_hold_ms = 250;
    int send_tick_ms = 15;
    int call_active_stale_ms = 1200;
};

std::vector<std::shared_ptr<endpoint_group_t>> endpoint_groups;

// ---------------------------------------------------------------------------
//  Class Declaration
// ---------------------------------------------------------------------------

/**
 * @brief Implements the actual DVM trunk-recorder dvmbridge stream plugin.
 */
class DVMTRStream : public Plugin_Api {
    typedef boost::asio::io_service io_service;
    io_service io;
    ip::udp::endpoint remote_endpoint;
    ip::udp::socket udp_socket{io};
    std::unique_ptr<std::thread> io_thread;
    std::unique_ptr<io_service::work> work_guard;

    int global_silence_leader = 0; // global silence leader in ms
    int global_inter_stream_delay = 0; // global inter-stream delay in ms
    bool global_strict_call_serialization = true;
    int global_call_gap_hold_ms = 250;
    int global_send_tick_ms = 15;
    int global_call_active_stale_ms = 1200;
#ifdef DVMTRSTREAM_ANALOG_SUPPORT
    bool logged_analog_downsample = false;
#endif

public:
    /**
     * @brief Initializes a new instance of the DVMTRStream class
     */
    DVMTRStream() { }

    /**
     * @brief Helper method to parse configuration data from JSON.
     * @param config_data 
     * @returns int 
     */
    int parse_config(json config_data) {
        std::map<std::string, std::shared_ptr<endpoint_group_t>> endpoint_map;

        // read global inter-stream delay if configured
        if (config_data.contains("interCallDelay")) {
            global_inter_stream_delay = config_data["interCallDelay"];
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: inter-stream/inter-call delay set to " << global_inter_stream_delay << "ms";
        }

        // read global silence leader if configured
        if (config_data.contains("silenceLeader")) {
            global_silence_leader = config_data["silenceLeader"];
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: silence leader set to " << global_silence_leader << "ms";
        }

        if (config_data.contains("strictCallSerialization")) {
            global_strict_call_serialization = config_data["strictCallSerialization"];
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: strict call serialization "
                                    << (global_strict_call_serialization ? "enabled" : "disabled");
        }

        if (config_data.contains("callGapHoldMs")) {
            global_call_gap_hold_ms = config_data["callGapHoldMs"];
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: call gap hold set to " << global_call_gap_hold_ms << "ms";
        }

        if (config_data.contains("sendTickMs")) {
            global_send_tick_ms = config_data["sendTickMs"];
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: send tick set to " << global_send_tick_ms << "ms";
        }

        if (config_data.contains("callActiveStaleMs")) {
            global_call_active_stale_ms = config_data["callActiveStaleMs"];
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: active call stale timeout set to " << global_call_active_stale_ms << "ms";
        }

        for (json element : config_data["streams"]) {
            auto stream = std::make_shared<stream_t>();
            stream->TGID = element["TGID"];
            stream->address = element["address"];
            stream->port = element["port"];
            stream->short_name = element["shortName"];   // we require shortName because dvmbridge can only handle one system ID at a time
            stream->remote_endpoint = ip::udp::endpoint(ip::address::from_string(stream->address), stream->port);

            // create endpoint key
            std::string endpoint_key = stream->address + ":" + std::to_string(stream->port);
            
            // find or create endpoint group
            if (endpoint_map.find(endpoint_key) == endpoint_map.end()) {
                auto group = std::make_shared<endpoint_group_t>();
                group->endpoint_key = endpoint_key;
                group->inter_stream_delay_ms = global_inter_stream_delay;
                group->silence_leader_ms = global_silence_leader;
                group->strict_call_serialization = global_strict_call_serialization;
                group->call_gap_hold_ms = global_call_gap_hold_ms;
                group->send_tick_ms = global_send_tick_ms;
                group->call_active_stale_ms = global_call_active_stale_ms;
                endpoint_map[endpoint_key] = group;
                endpoint_groups.push_back(group);
                BOOST_LOG_TRIVIAL(info) << "dvmtrstream: created endpoint group for " << endpoint_key;
            }

            endpoint_map[endpoint_key]->streams.push_back(stream);
            streams.push_back(stream);

            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: will stream audio from TGID " << stream->TGID << " on System " << stream->short_name << " to " << stream->address << " on port " << stream->port;
        }

        BOOST_LOG_TRIVIAL(info) << "dvmtrstream: configured " << endpoint_groups.size() << " endpoint group(s) with " << streams.size() << " total stream(s)";
        return 0;
    }

    /**
     * @brief Audio streaming handler called by trunk-recorder for each audio buffer.
     * @param call 
     * @param recorder 
     * @param samples 
     * @param sampleCount 
     * @returns int 
     */
    int audio_stream(Call* call, Recorder* recorder, int16_t* samples, int sampleCount) {
        System* call_system = call->get_system();
        int32_t call_tgid = call->get_talkgroup();
        int32_t call_src = call->get_current_source_id();
        std::string call_short_name = call->get_short_name();
        std::vector<unsigned long> unsigned_patched_talkgroups = call_system->get_talkgroup_patch(call_tgid);
        std::vector<long> patched_talkgroups;

        // convert unsigned long to signed long, preserving negative values
        for (auto tgid : unsigned_patched_talkgroups) {
            patched_talkgroups.push_back(static_cast<long>(tgid));
        }

        if (call_src == -1) {
            if (call->get_transmissions().size() > 0) {
                // get the source from the most recent transmission
                auto transmissions = call->get_transmissions();
                call_src = transmissions.back().source;
                BOOST_LOG_TRIVIAL(debug) << "using source " << call_src << " from most recent transmission";
            } else {
                BOOST_LOG_TRIVIAL(debug) << "no source found for call - leaving src as -1";
            }
        }

        Recorder& local_recorder = *recorder;
        int recorder_id = local_recorder.get_num();
        long wav_hz = local_recorder.get_wav_hz();
        int16_t* audio_samples = samples;
        int audio_sample_count = sampleCount;
#ifdef DVMTRSTREAM_ANALOG_SUPPORT
        std::vector<int16_t> downsampled_samples;
#endif

        // ignore audio if it's not 8kHz
        if (wav_hz != 8000) {
#ifdef DVMTRSTREAM_ANALOG_SUPPORT
            if (wav_hz == 16000) {
                downsampled_samples = downsample16kTo8k(samples, sampleCount);
                audio_samples = downsampled_samples.data();
                audio_sample_count = static_cast<int>(downsampled_samples.size());

                if (!logged_analog_downsample) {
                    BOOST_LOG_TRIVIAL(info) << "dvmtrstream: downsampling Trunk Recorder analog audio from 16000 Hz to 8000 Hz for DVMBridge";
                    logged_analog_downsample = true;
                }
            } else
#endif
            {
                BOOST_LOG_TRIVIAL(warning) << "ignoring audio at " << wav_hz << " Hz samplerate, not currently supported!";
                return 1;
            }
        }

        BOOST_FOREACH (auto& stream, streams) {
            if (0 == stream->short_name.compare(call_short_name) || (0 == stream->short_name.compare(""))) { // Check if shortName matches or is not specified
                if (patched_talkgroups.size() == 0) {
                    patched_talkgroups.push_back(call_tgid); // call_info.talkgroup may be negative - we cast stream.TGID to signed for comparison
                }
                BOOST_FOREACH (auto TGID, patched_talkgroups) {
                    if ((TGID == static_cast<long>(stream->TGID))) {
                        BOOST_LOG_TRIVIAL(debug) << "got " << audio_sample_count << " samples - " << audio_sample_count * 2 << " bytes from recorder " << recorder_id << " for TGID " << TGID;

                        int32_t bytes = audio_sample_count * 2;
                        const int16_t chunkSize = 320; // 20ms of 8kHz 16-bit audio (160 samples * 2 bytes)
                        int16_t totalChunks = bytes / chunkSize;
                        int32_t remainingBytes = bytes % chunkSize;

                        uint8_t* sampleBytes = (uint8_t*)audio_samples;

                        // queue chunks instead of sending immediately
                        std::lock_guard<std::mutex> lock(stream->queue_mutex);
                        stream->last_enqueue_time = std::chrono::steady_clock::now();
                        stream->call_active = true;

                        int32_t effective_src = call_src;
                        if (effective_src < 0 && stream->last_call_tgid == call_tgid && stream->last_call_src > 0) {
                            // Reuse previous valid source for this stream while call metadata catches up.
                            effective_src = stream->last_call_src;
                        }
                        
                        // are we injecting a silence leader?
                        if (global_silence_leader > 0) {
                            // check if call source or talkgroup changed
                            bool call_changed = (stream->last_call_tgid != call_tgid || stream->last_call_src != effective_src);

                            // inject silence leader if configured and call changed
                            if (call_changed && !stream->silence_leader_injected) {
                                // find the endpoint group for this stream to get silence_leader_ms
                                int silence_leader_ms = 0;
                                for (auto& group : endpoint_groups) {
                                    for (auto& grp_stream : group->streams) {
                                        if (grp_stream.get() == stream.get()) {
                                            silence_leader_ms = group->silence_leader_ms;
                                            break;
                                        }
                                    }
                                    if (silence_leader_ms > 0) break;
                                }

                                if (silence_leader_ms > 0) {
                                    // each chunk is 20ms, calculate how many silence chunks needed
                                    int silence_chunks = (silence_leader_ms + 19) / 20;  // round up

                                    uint8_t silenceChunk[320];
                                    ::memset(silenceChunk, 0, sizeof(silenceChunk));

                                    for (int i = 0; i < silence_chunks; i++) {
                                        audio_chunk_t silent_chunk(silenceChunk, chunkSize, call_tgid, effective_src);
                                        stream->chunk_queue.push(silent_chunk);
                                    }

                                    stream->silence_leader_injected = true;
                                    BOOST_LOG_TRIVIAL(debug) << "injected " << silence_chunks << " silence chunks (" 
                                                            << (silence_chunks * 20) << "ms) before audio for TGID " << TGID 
                                                            << " SRC " << effective_src << " (call changed)";
                                }
                            }
                        }

                        // update last call tracking
                        stream->last_call_tgid = call_tgid;
                        stream->last_call_src = effective_src;
                        
#ifdef DVMTRSTREAM_ANALOG_SUPPORT
                        if (wav_hz == 16000) {
                            if (bytes > 0) {
                                // Analog 16 kHz callbacks may be smaller than one 8 kHz DVMBridge frame.
                                // Accumulate after downsampling so we do not stretch audio with per-callback padding.
                                if (!stream->pending_audio.empty() &&
                                    (stream->pending_audio_tgid != call_tgid || stream->pending_audio_src != call_src)) {
                                    uint8_t paddedChunk[320];
                                    ::memset(paddedChunk, 0, sizeof(paddedChunk));
                                    ::memcpy(paddedChunk, stream->pending_audio.data(), stream->pending_audio.size());

                                    audio_chunk_t audio_chunk(paddedChunk, chunkSize, stream->pending_audio_tgid, stream->pending_audio_src);
                                    stream->chunk_queue.push(audio_chunk);
                                    stream->pending_audio.clear();
                                }

                                stream->pending_audio_tgid = call_tgid;
                                stream->pending_audio_src = effective_src;
                                stream->pending_audio_updated = std::chrono::steady_clock::now();
                                stream->pending_audio.insert(stream->pending_audio.end(), sampleBytes, sampleBytes + bytes);

                                size_t consumedBytes = 0;
                                while (stream->pending_audio.size() - consumedBytes >= static_cast<size_t>(chunkSize)) {
                                    audio_chunk_t audio_chunk(stream->pending_audio.data() + consumedBytes, chunkSize, call_tgid, effective_src);
                                    stream->chunk_queue.push(audio_chunk);
                                    consumedBytes += chunkSize;
                                }

                                if (consumedBytes > 0) {
                                    stream->pending_audio.erase(stream->pending_audio.begin(), stream->pending_audio.begin() + consumedBytes);
                                }
                            }
                        } else
#endif
                        {
                            // queue complete chunks
                            for (int i = 0; i < totalChunks; i++) {
                                // calculate the starting position of the current chunk
                                uint8_t* curChunkStart = sampleBytes + (i * chunkSize);

                                // queue the chunk
                                audio_chunk_t audio_chunk(curChunkStart, chunkSize, call_tgid, effective_src);
                                stream->chunk_queue.push(audio_chunk);
                            }

                            // handle partial chunk - pad with silence to full 320 bytes
                            if (remainingBytes > 0) {
                                uint8_t paddedChunk[320];
                                ::memset(paddedChunk, 0, sizeof(paddedChunk)); // Zero = silence for 16-bit audio

                                // copy partial audio data
                                uint8_t* partialStart = sampleBytes + (totalChunks * chunkSize);
                                ::memcpy(paddedChunk, partialStart, remainingBytes);

                                // queue padded chunk
                                audio_chunk_t audio_chunk(paddedChunk, chunkSize, call_tgid, effective_src);
                                stream->chunk_queue.push(audio_chunk);

                                BOOST_LOG_TRIVIAL(debug) << "padded partial chunk: " << remainingBytes << " bytes + "
                                                         << (chunkSize - remainingBytes) << " silence bytes for TGID " << TGID;
                            }
                        }
                        
                        size_t queue_size = stream->chunk_queue.size();
                        if (queue_size > 50) {
                            BOOST_LOG_TRIVIAL(warning) << "stream queue for TGID " << TGID << " has " << queue_size << " chunks (" << queue_size * 20 << "ms backlog)";
                        }
                    }
                }
            }
        }
        return 0;
    }

    /**
     * @brief Marks matching stream calls as inactive when trunk-recorder closes a call.
     * @param call_info
     * @returns int 
     */
    int call_end(Call_Data_t call_info) {
        std::vector<long> ended_talkgroups;
        ended_talkgroups.push_back(call_info.talkgroup);
        for (auto patched_tgid : call_info.patched_talkgroups) {
            ended_talkgroups.push_back(static_cast<long>(patched_tgid));
        }

        for (auto& stream : streams) {
            if (!stream) {
                continue;
            }

            if (!(stream->short_name.empty() || stream->short_name == call_info.short_name)) {
                continue;
            }

            bool matches = false;
            for (auto ended_tgid : ended_talkgroups) {
                if (ended_tgid == stream->TGID) {
                    matches = true;
                    break;
                }
            }

            if (!matches) {
                continue;
            }

            std::lock_guard<std::mutex> lock(stream->queue_mutex);
            stream->call_active = false;
            stream->last_call_tgid = -1;
            stream->last_call_src = -1;
        }

        return 0;
    }

    /**
     * @brief Starts the plugin.
     * @returns int 
     */
    int start() {
        udp_socket.open(ip::udp::v4());

        // create work guard to keep io_service running
        work_guard = std::make_unique<io_service::work>(io);

        // initialize timer for each endpoint group
        for (auto& group : endpoint_groups) {
            group->timer = std::make_shared<boost::asio::steady_timer>(io);
            group->timer_active = true;
            schedule_next_send(group.get());
        }

        // start io_service in a separate thread
        io_thread = std::make_unique<std::thread>([this]() {
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: starting io_service thread";
            io.run();
            BOOST_LOG_TRIVIAL(info) << "dvmtrstream: io_service thread stopped";
        });
        
        return 0;
    }

    /**
     * @brief Stops the plugin.
     * @returns int 
     */
    int stop() {
        // stop all endpoint group timers
        for (auto& group : endpoint_groups) {
            group->timer_active = false;
            if (group->timer) {
                group->timer->cancel();
            }
        }

        // stop io_service
        work_guard.reset();
        io.stop();

        // wait for io_thread to finish
        if (io_thread && io_thread->joinable()) {
            io_thread->join();
        }

        udp_socket.close();
        return 0;
    }

    static boost::shared_ptr<DVMTRStream> create() {
        return boost::shared_ptr<DVMTRStream>(
            new DVMTRStream());
    }

private:
    /**
     * @brief Helper method to schedule the next send for an endpoint group.
     * @param group 
     */
    void schedule_next_send(endpoint_group_t* group) {
        if (!group || !group->timer || !group->timer_active) 
            return;

        group->timer->expires_from_now(std::chrono::milliseconds(group->send_tick_ms));
        group->timer->async_wait([this, group](const boost::system::error_code& error) {
            if (!error && group->timer_active) {
                send_next_chunk(group);
            }
        });
    }

    /**
     * @brief Helper method to send the next audio chunk for an endpoint group.
     * @param group 
     */
    void send_next_chunk(endpoint_group_t* group) {
        if (!group || group->streams.empty()) {
            schedule_next_send(group);
            return;
        }

        auto has_recent_activity = [group](const std::shared_ptr<stream_t>& stream) {
            if (!stream) {
                return false;
            }

            std::lock_guard<std::mutex> lock(stream->queue_mutex);
            if (stream->last_enqueue_time == std::chrono::steady_clock::time_point{} &&
                stream->last_send_time == std::chrono::steady_clock::time_point{}) {
                return false;
            }

            auto now = std::chrono::steady_clock::now();
            auto last_activity = stream->last_enqueue_time;
            if (stream->last_send_time > last_activity) {
                last_activity = stream->last_send_time;
            }

            return now - last_activity < std::chrono::milliseconds(group->call_gap_hold_ms);
        };

        // find next stream with data in this endpoint group, starting from current_stream_index
        stream_t* active_stream = nullptr;
        size_t start_index = group->current_stream_index;

        if (group->strict_call_serialization && group->current_stream_index < group->streams.size()) {
            auto& pinned_stream = group->streams[group->current_stream_index];
            bool pinned_call_active = false;

            // scope is intentional
            {
                std::lock_guard<std::mutex> lock(pinned_stream->queue_mutex);
#ifdef DVMTRSTREAM_ANALOG_SUPPORT
                if (pinned_stream->chunk_queue.empty() && !pinned_stream->pending_audio.empty()) {
                    auto now = std::chrono::steady_clock::now();
                    if (now - pinned_stream->pending_audio_updated >= std::chrono::milliseconds(50)) {
                        uint8_t paddedChunk[320];
                        ::memset(paddedChunk, 0, sizeof(paddedChunk));
                        ::memcpy(paddedChunk, pinned_stream->pending_audio.data(), pinned_stream->pending_audio.size());

                        audio_chunk_t audio_chunk(paddedChunk, sizeof(paddedChunk), pinned_stream->pending_audio_tgid, pinned_stream->pending_audio_src);
                        pinned_stream->chunk_queue.push(audio_chunk);
                        pinned_stream->pending_audio.clear();
                    }
                }
#endif
                if (!pinned_stream->chunk_queue.empty()) {
                    active_stream = pinned_stream.get();
                }
                pinned_call_active = pinned_stream->call_active;
            }

            if (!active_stream && pinned_call_active) {
                if (has_recent_activity(pinned_stream)) {
                    schedule_next_send(group);
                    return;
                }

                // scope is intentional
                {
                    std::lock_guard<std::mutex> lock(pinned_stream->queue_mutex);
                    auto now = std::chrono::steady_clock::now();
                    auto last_activity = pinned_stream->last_enqueue_time;
                    if (pinned_stream->last_send_time > last_activity) {
                        last_activity = pinned_stream->last_send_time;
                    }

                    if (last_activity != std::chrono::steady_clock::time_point{}) {
                        auto idle_for = now - last_activity;
                        if (idle_for < std::chrono::milliseconds(group->call_active_stale_ms)) {
                            schedule_next_send(group);
                            return;
                        }
                    } else {
                        schedule_next_send(group);
                        return;
                    }

                    BOOST_LOG_TRIVIAL(info) << "dvmtrstream: releasing stale active call latch for endpoint "
                                            << group->endpoint_key << " TGID " << pinned_stream->TGID
                                            << " after " << group->call_active_stale_ms << "ms idle";
                    pinned_stream->call_active = false;
                }
            }

            if (!active_stream && has_recent_activity(pinned_stream)) {
                schedule_next_send(group);
                return;
            }

            // In strict mode, if the pinned stream has data, never scan/switch
            // to another stream in this cycle. This enforces hard serialization.
            if (active_stream != nullptr) {
                goto SEND_ACTIVE_STREAM;
            }

            start_index = (group->current_stream_index + 1) % group->streams.size();
        }

        for (size_t i = 0; i < group->streams.size(); i++) {
            size_t check_index = (start_index + i) % group->streams.size();
            auto& stream = group->streams[check_index];

            if (group->strict_call_serialization && check_index == group->current_stream_index && active_stream != nullptr) {
                continue;
            }

            std::lock_guard<std::mutex> lock(stream->queue_mutex);
#ifdef DVMTRSTREAM_ANALOG_SUPPORT
            if (stream->chunk_queue.empty() && !stream->pending_audio.empty()) {
                auto now = std::chrono::steady_clock::now();
                if (now - stream->pending_audio_updated >= std::chrono::milliseconds(50)) {
                    uint8_t paddedChunk[320];
                    ::memset(paddedChunk, 0, sizeof(paddedChunk));
                    ::memcpy(paddedChunk, stream->pending_audio.data(), stream->pending_audio.size());

                    audio_chunk_t audio_chunk(paddedChunk, sizeof(paddedChunk), stream->pending_audio_tgid, stream->pending_audio_src);
                    stream->chunk_queue.push(audio_chunk);

                    BOOST_LOG_TRIVIAL(debug) << "flushed analog partial chunk: " << stream->pending_audio.size() << " bytes + "
                                             << (sizeof(paddedChunk) - stream->pending_audio.size()) << " silence bytes for TGID "
                                             << stream->pending_audio_tgid;

                    stream->pending_audio.clear();
                }
            }
#endif
            if (!stream->chunk_queue.empty()) {
                // found stream with data - check if it's the current one or need to switch
                if (check_index != group->current_stream_index) {
                    group->current_stream_index = check_index;
                    BOOST_LOG_TRIVIAL(debug) << "endpoint " << group->endpoint_key << ": Switching to stream TGID " << stream->TGID;
                }
                active_stream = stream.get();
                break;
            }
        }

        if (!active_stream) {
            // no streams have data in this group, check again later
            schedule_next_send(group);
            return;
        }

SEND_ACTIVE_STREAM:

        // send chunk from active stream
        std::unique_lock<std::mutex> lock(active_stream->queue_mutex);
        audio_chunk_t chunk = active_stream->chunk_queue.front();
        active_stream->chunk_queue.pop();
        bool has_more = !active_stream->chunk_queue.empty();
        active_stream->last_send_time = std::chrono::steady_clock::now();
        lock.unlock();

        // send the chunk
        boost::system::error_code error;

        uint8_t chunkBuffer[332];
        ::memset(chunkBuffer, 0, sizeof(chunkBuffer));
        SET_UINT32(chunk.data.size(), chunkBuffer, 0);
        std::memcpy(chunkBuffer + 4, chunk.data.data(), chunk.data.size());

        uint32_t dstId = static_cast<uint32_t>(chunk.tgid);

        // if the actual talkgroup is negative, set to 0
        if (chunk.tgid < 0) {
            dstId = 0;
        }

        // DVMBridge payload stores 24-bit IDs; preserve lower bits for larger values.
        dstId &= 0xFFFFFF;

        SET_UINT32(dstId, chunkBuffer, 4 + chunk.data.size());

        uint32_t srcId = static_cast<uint32_t>(chunk.src);

        // if the actual source is negative, set to 0
        if (chunk.src < 0) {
            srcId = 0;
        }

        // DVMBridge payload stores 24-bit IDs; preserve lower bits for larger values.
        srcId &= 0xFFFFFF;

        SET_UINT32(srcId, chunkBuffer, 8 + chunk.data.size());

        std::vector<boost::asio::const_buffer> txBuffer;
        txBuffer.push_back(buffer(chunkBuffer, sizeof(chunkBuffer)));

        udp_socket.send_to(txBuffer, active_stream->remote_endpoint, 0, error);

        if (error) {
            BOOST_LOG_TRIVIAL(error) << "error sending UDP packet: " << error.message();
        }

        // if current stream is empty, move to next stream on next cycle
        if (!has_more) {
            // scope is intentional
            {
                std::lock_guard<std::mutex> stream_lock(active_stream->queue_mutex);
                // reset silence leader flag for next transmission
                active_stream->silence_leader_injected = false;
                if (active_stream->chunk_queue.empty() && !active_stream->call_active) {
                    active_stream->last_call_tgid = -1;
                    active_stream->last_call_src = -1;
                }
            }

            if (group->strict_call_serialization) {
                schedule_next_send(group);
                return;
            }

            // check if inter-stream delay is configured
            if (group->inter_stream_delay_ms > 0) {
                // enter delay period before switching to next stream
                group->switching_stream = true;
                BOOST_LOG_TRIVIAL(debug) << "endpoint " << group->endpoint_key 
                                         << ": Stream TGID " << active_stream->TGID 
                                         << " complete, delaying " << group->inter_stream_delay_ms << "ms before next stream";

                // schedule with delay
                if (group->timer && group->timer_active) {
                    group->timer->expires_from_now(std::chrono::milliseconds(group->inter_stream_delay_ms));
                    group->timer->async_wait([this, group](const boost::system::error_code& error) {
                        if (!error && group->timer_active) {
                            group->switching_stream = false;
                            group->current_stream_index = (group->current_stream_index + 1) % group->streams.size();
                            send_next_chunk(group);
                        }
                    });
                }
                return;  // don't schedule normal send, we scheduled the delayed one
            } else {
                // no delay, switch immediately
                group->current_stream_index = (group->current_stream_index + 1) % group->streams.size();
            }
        }

        // schedule next send for this endpoint group
        schedule_next_send(group);
    }
};

BOOST_DLL_ALIAS(
    DVMTRStream::create, // <-- this function is exported with...
    create_plugin        // <-- ...this alias name
)