chanora/crates/chanora_audio/src/audio_event_queue.rs

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use chanora_protocol::InAudioBuf;
use crossbeam::queue::ArrayQueue;

use crate::engine::SessionAudioId;

const PACKET_QUEUE_CAPACITY: usize = 100;
const CONTROL_QUEUE_CAPACITY: usize = 32;

/// A raw inbound voice packet waiting to be inserted into AudioHandler.
#[derive(Debug)]
pub struct AudioPacket {
    /// Client whose TeamSpeak audio packet this belongs to.
    pub client_id: SessionAudioId,
    /// Raw inbound TeamSpeak audio payload accepted by AudioHandler::handle_packet.
    pub data: InAudioBuf,
}

/// Control commands from the main thread to the audio callback.
#[derive(Debug)]
pub enum AudioCommand {
    /// Set a client's output volume.
    SetVolume(SessionAudioId, f32),
    /// Remove a client's decode queue.
    // TODO: Wire to client disconnect path; handled in callback but no
    // producer currently pushes this command.
    #[allow(dead_code)]
    RemoveClient(SessionAudioId),
}

/// Lock-free bridge between the inbound forwarder / main thread and the
/// audio callback. The callback owns the consumer halves.
pub struct AudioEventQueue {
    /// Bounded lossy queue for raw voice packets. On overflow, the push
    /// fails and the packet is dropped (counted via `packets_dropped`).
    /// Capacity: 100 packets (~2 seconds at 50pps, far more than needed).
    pub packet_queue: ArrayQueue<AudioPacket>,
    /// Bounded reliable queue for control commands (volume, client removal).
    /// On overflow, the caller retries. Capacity: 32 commands.
    pub control_queue: ArrayQueue<AudioCommand>,
    /// Atomic counter for dropped packets (for diagnostics).
    pub packets_dropped: AtomicU64,
}

impl AudioEventQueue {
    /// Create the Android audio event bridge with fixed queue capacities.
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            packet_queue: ArrayQueue::new(PACKET_QUEUE_CAPACITY),
            control_queue: ArrayQueue::new(CONTROL_QUEUE_CAPACITY),
            packets_dropped: AtomicU64::new(0),
        })
    }

    /// Create a producer handle sharing this queue.
    pub fn producer(queue: &Arc<Self>) -> AudioEventProducer {
        AudioEventProducer {
            queue: Arc::clone(queue),
        }
    }

    /// Create a consumer handle sharing this queue.
    pub fn consumer(queue: &Arc<Self>) -> AudioEventConsumer {
        AudioEventConsumer {
            queue: Arc::clone(queue),
        }
    }
}

/// Producer side used by the inbound forwarder and engine control methods.
#[derive(Clone)]
pub struct AudioEventProducer {
    queue: Arc<AudioEventQueue>,
}

impl AudioEventProducer {
    /// Push a raw voice packet, incrementing the drop counter if full.
    pub fn push_packet(&self, packet: AudioPacket) -> Result<(), AudioPacket> {
        self.queue.packet_queue.push(packet).map_err(|packet| {
            self.queue.packets_dropped.fetch_add(1, Ordering::Relaxed);
            packet
        })
    }

    /// Push a control command, returning it unchanged if the queue is full.
    pub fn push_control(&self, cmd: AudioCommand) -> Result<(), AudioCommand> {
        self.queue.control_queue.push(cmd)
    }

    /// Shared queue backing this producer.
    pub fn queue(&self) -> Arc<AudioEventQueue> {
        Arc::clone(&self.queue)
    }
}

/// Consumer side used by the Android output callback.
pub struct AudioEventConsumer {
    queue: Arc<AudioEventQueue>,
}

impl AudioEventConsumer {
    /// Pop up to `cap` queued packets.
    pub fn drain_packets(&self, cap: usize) -> impl Iterator<Item = AudioPacket> + '_ {
        let mut drained = 0;
        std::iter::from_fn(move || {
            if drained >= cap {
                return None;
            }
            let packet = self.queue.packet_queue.pop();
            if packet.is_some() {
                drained += 1;
            }
            packet
        })
    }

    /// Pop all currently queued controls.
    pub fn drain_controls(&self) -> impl Iterator<Item = AudioCommand> + '_ {
        std::iter::from_fn(move || self.queue.control_queue.pop())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn empty_packet(id: u64) -> AudioPacket {
        let audio = chanora_protocol::AudioData::S2C {
            codec: chanora_protocol::CodecType::OpusVoice,
            id: 0x1234,
            from: 0x5678,
            data: &[1, 2, 3],
        };
        let out = chanora_protocol::OutAudio::new(&audio);
        AudioPacket {
            client_id: SessionAudioId(id),
            data: InAudioBuf::try_new(chanora_protocol::Direction::S2C, out.data().to_vec())
                .unwrap(),
        }
    }

    #[test]
    fn packet_overflow_increments_drop_counter() {
        let queue = AudioEventQueue::new();
        let producer = AudioEventQueue::producer(&queue);

        for i in 0..PACKET_QUEUE_CAPACITY {
            let packet = empty_packet(i as u64);
            assert!(producer.push_packet(packet).is_ok());
        }

        let overflow = empty_packet(999);
        assert!(producer.push_packet(overflow).is_err());
        assert_eq!(queue.packets_dropped.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn consumer_drains_packets_and_controls() {
        let queue = AudioEventQueue::new();
        let producer = AudioEventQueue::producer(&queue);
        let consumer = AudioEventQueue::consumer(&queue);

        producer
            .push_control(AudioCommand::SetVolume(SessionAudioId(7), 0.5))
            .unwrap();
        producer.push_packet(empty_packet(42)).unwrap();

        let packets: Vec<_> = consumer.drain_packets(8).collect();
        assert_eq!(packets.len(), 1);
        assert_eq!(packets[0].client_id, SessionAudioId(42));

        let controls: Vec<_> = consumer.drain_controls().collect();
        assert_eq!(controls.len(), 1);
        match controls[0] {
            AudioCommand::SetVolume(id, vol) => {
                assert_eq!(id, SessionAudioId(7));
                assert_eq!(vol, 0.5);
            }
            AudioCommand::RemoveClient(_) => panic!("unexpected remove-client command"),
        }
    }
}