rtp_audio_receiver.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: AGPL-3.0-or-later
/*
 * Project: RAVENNAKIT (RAVENNA / AES67 / ST2110-30 SDK)
 * Copyright (c) 2024-2025 Sound on Digital
 *
 * This file is part of RAVENNAKIT.
 *
 * RAVENNAKIT is dual-licensed:
 *   1) Under the terms of the GNU Affero General Public License as published by
 *      the Free Software Foundation, either version 3 of the License, or
 *      (at your option) any later version (the "AGPL License"); and
 *   2) Under a commercial license from Sound on Digital, for customers who
 *      cannot (or do not wish to) comply with the AGPL License terms.
 *
 * If you obtained this file under the AGPL License, you may redistribute it
 * and/or modify it under the terms of the AGPL License. See the LICENSE
 * file in the project root for details.
 *
 * For commercial licensing, support, and other inquiries, please visit:
 *
 *     https://ravennakit.com
 *
 */
 
#pragma once
 
#include "rtp_filter.hpp"
#include "rtp_packet_stats.hpp"
#include "rtp_ringbuffer.hpp"
#include "rtp_session.hpp"
#include "ravennakit/aes67/aes67_constants.hpp"
#include "ravennakit/core/audio/audio_buffer_view.hpp"
#include "ravennakit/core/math/interval_stats.hpp"
#include "ravennakit/core/math/sliding_stats.hpp"
#include "ravennakit/core/net/asio/asio_helpers.hpp"
#include "ravennakit/core/sync/atomic_rw_lock.hpp"
#include "ravennakit/core/util/id.hpp"
#include "ravennakit/core/util/safe_function.hpp"
#include "ravennakit/ptp/ptp_instance.hpp"
 
#include <boost/asio.hpp>
#include <boost/container/static_vector.hpp>
#include <boost/lockfree/spsc_value.hpp>
 
#ifndef RAV_MAX_NUM_READERS
    #define RAV_MAX_NUM_READERS 16
#endif
 
namespace rav::rtp {
 
struct AudioReceiver {
    static constexpr auto k_max_num_readers = RAV_MAX_NUM_READERS;

    static constexpr auto k_max_num_redundant_sessions = 2;  // How many redundant paths

    static constexpr uint64_t k_receive_timeout_ms = 1000;

    static constexpr uint32_t k_buffer_size_ms = 200;
 
    using ArrayOfAddresses = std::array<ip_address_v4, k_max_num_redundant_sessions>;
 
    struct StreamInfo {
        Session session;
        Filter filter;
        uint16_t packet_time_frames {};
 
        [[nodiscard]] auto tie() const {
            return std::tie(session, filter, packet_time_frames);
        }
 
        friend bool operator==(const StreamInfo& lhs, const StreamInfo& rhs) {
            return lhs.tie() == rhs.tie();
        }
 
        friend bool operator!=(const StreamInfo& lhs, const StreamInfo& rhs) {
            return lhs.tie() != rhs.tie();
        }
 
        [[nodiscard]] bool is_valid() const {
            return session.valid() && packet_time_frames > 0;
        }
    };
 
    struct ReaderParameters {
        AudioFormat audio_format;
        std::array<StreamInfo, k_max_num_redundant_sessions> streams;
 
        [[nodiscard]] auto tie() const {
            return std::tie(audio_format, streams);
        }
 
        friend bool operator==(const ReaderParameters& lhs, const ReaderParameters& rhs) {
            return lhs.tie() == rhs.tie();
        }
 
        friend bool operator!=(const ReaderParameters& lhs, const ReaderParameters& rhs) {
            return lhs.tie() != rhs.tie();
        }
 
        [[nodiscard]] bool is_valid() const {
            if (!audio_format.is_valid()) {
                return false;
            }
            for (auto& stream : streams) {
                if (stream.is_valid()) {
                    return true;  // At least one stream needs to be valid.
                }
            }
            return false;
        }
    };

    enum class StreamState {
        inactive,
        receiving,
        no_consumer,
    };
 
    explicit AudioReceiver(boost::asio::io_context& io_context);
    ~AudioReceiver();

    [[nodiscard]] bool add_reader(Id id, const ReaderParameters& parameters, const ArrayOfAddresses& interfaces);

    [[nodiscard]] bool remove_reader(Id id);

    [[nodiscard]] bool set_interfaces(const ArrayOfAddresses& interfaces);

    void read_incoming_packets();

    [[nodiscard]] std::optional<uint32_t> read_data_realtime(
        Id id, uint8_t* buffer, size_t buffer_size, std::optional<uint32_t> at_timestamp, std::optional<uint32_t> require_delay
    );

    [[nodiscard]] std::optional<uint32_t> read_audio_data_realtime(
        Id id, AudioBufferView<float>& output_buffer, std::optional<uint32_t> at_timestamp, std::optional<uint32_t> require_delay
    );

    std::optional<PacketStats::Counters> get_packet_stats(Id reader_id, size_t stream_index);

    [[nodiscard]] std::optional<StreamState> get_stream_state(Id reader_id, size_t stream_index) const;
 
    struct SocketWithContext {
        explicit SocketWithContext(boost::asio::io_context& io_context) : socket(io_context) {}
 
        AtomicRwLock rw_lock;
        udp_socket socket;
        uint16_t port {};
    };
 
    struct PacketBuffer {
        uint32_t timestamp;
        uint16_t seq;
        uint16_t data_len;
        uint64_t recv_time;
        std::array<uint8_t, aes67::constants::k_max_payload> payload;
    };
 
    struct StreamContext {
        Session session;
        Filter filter;
        uint16_t packet_time_frames {};
        std::optional<WrappingUint32> rtp_ts;
        ip_address_v4 interface;
        FifoBuffer<PacketBuffer, Fifo::Spsc> packets;
        FifoBuffer<uint16_t, Fifo::Spsc> packets_too_old;
        PacketStats packet_stats;
        boost::lockfree::spsc_value<PacketStats::Counters, boost::lockfree::allow_multiple_reads<true>> packet_stats_counters;
        std::atomic<bool> reset_max_values {false};
        IntervalStats packet_interval_stats;
        WrappingUint64 prev_packet_time_ns;
        std::atomic<StreamState> state {StreamState::inactive};
    };

    struct Reader {
        AtomicRwLock rw_lock;
        Id id;
        AudioFormat audio_format;
        std::array<StreamContext, k_max_num_redundant_sessions> streams;
 
        // Audio thread
        Ringbuffer receive_buffer;
        std::vector<uint8_t> read_audio_data_buffer;
        std::optional<WrappingUint32> most_recent_ts;  // ts of the latest received data
        WrappingUint32 next_ts_to_read;
    };

    SafeFunction<bool(udp_socket&, ip_address_v4, ip_address_v4)> join_multicast_group;

    SafeFunction<bool(udp_socket&, ip_address_v4, ip_address_v4)> leave_multicast_group;
 
    ptp::Instance::Subscriber ptp_instance_subscriber;
 
    static constexpr auto k_max_num_sessions = k_max_num_readers * k_max_num_redundant_sessions;
 
    boost::container::static_vector<SocketWithContext, k_max_num_sessions> sockets;
    boost::container::static_vector<Reader, k_max_num_readers> readers;
 
    uint64_t last_time_maintenance {};
};

[[nodiscard]] const char* to_string(AudioReceiver::StreamState state);
 
}  // namespace rav::rtp