tap/tap_core/src/audio_source_thread.rs

use crate::tx_thread::TxThread;
use crate::util::{SAMPLE_RATE, SAMPLE_RATE_RAW, STEREO_FRAME_SIZE, ThreadMessage};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{BufferSize, FromSample, Host, Sample, SampleRate, Stream, StreamConfig};
use log::{debug, error, info};
use opus::{Application, Bitrate, Channels, Encoder, Signal};
use std::io::{Read, Write};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::thread;
use thiserror::Error;

#[derive(Debug, Error)]
pub enum Error {
    #[error("CPAL error: {0}")]
    CpalError(#[from] cpal::Error),
    #[error("Opus error: {0}")]
    OpusError(#[from] opus::Error),
    #[error("Unknown sample error format")]
    UnknownSampleErrorFormat,
    #[error("Unknown device")]
    UnknownDevice,
}

#[derive(Debug)]
pub enum Commands {
    SetDevice(String),
    Pause,
    Resume,
}

pub struct AudioSourceThread {
    device: String,
    host: Host,
    cmd_channel: Receiver<Commands>,
}

impl AudioSourceThread {
    pub fn new(device: String, host: Host, cmd_channel: Receiver<Commands>) -> Self {
        Self {
            device,
            host,
            cmd_channel,
        }
    }

    fn setup_stream(&mut self, audio_sender: &Sender<ThreadMessage>) -> Result<Stream, Error> {
        let device = if self.device == "default" {
            self.host.default_input_device()
        } else {
            let id = &self.device.parse()?;
            self.host.device_by_id(id)
        };

        let device = device.ok_or(Error::UnknownDevice)?;

        info!("Input device: {}", device.id()?);

        let config = device
            .default_input_config()
            .expect("Failed to get default input config");

        let mut stream_config = StreamConfig::from(config);
        let count = stream_config.channels;
        stream_config.sample_rate = SAMPLE_RATE_RAW as SampleRate;
        stream_config.buffer_size = BufferSize::Fixed(STEREO_FRAME_SIZE as u32 / count as u32);

        info!("Stream config: {stream_config:?}");

        for cfg in device.supported_input_configs()? {
            debug!("{:?}", cfg)
        }

        let callback_sender = audio_sender.clone();
        let stream = match config.sample_format() {
            cpal::SampleFormat::I8 => device.build_input_stream(
                stream_config,
                move |data, _: &_| write_input_data::<i8>(data, &callback_sender),
                err_fn,
                None,
            )?,
            cpal::SampleFormat::I16 => device.build_input_stream(
                stream_config,
                move |data, _: &_| write_input_data::<i16>(data, &callback_sender),
                err_fn,
                None,
            )?,
            cpal::SampleFormat::I32 => device.build_input_stream(
                stream_config,
                move |data, _: &_| write_input_data::<i32>(data, &callback_sender),
                err_fn,
                None,
            )?,
            cpal::SampleFormat::F32 => device.build_input_stream(
                stream_config,
                move |data, _: &_| write_input_data::<f32>(data, &callback_sender),
                err_fn,
                None,
            )?,
            sample_format => {
                error!("Unknown sample format: '{sample_format}'");
                return Err(Error::UnknownSampleErrorFormat);
            }
        };

        Ok(stream)
    }

    pub fn worker<T: Read + Write + Send + 'static>(mut self, connection: T) -> Result<(), Error> {
        let encoder = Self::setup_encoder()?;
        let (sender, receiver) = channel();
        let tx_thread = TxThread::new(receiver, encoder, connection);
        let thread = thread::spawn(move || tx_thread.thread());

        let mut stream = self.setup_stream(&sender)?;
        info!("Running...");
        stream.play()?;

        loop {
            let msg = self.cmd_channel.recv();

            if let Ok(msg) = msg {
                match msg {
                    Commands::SetDevice(dev) => {
                        info!("Setting device to {dev}...");
                        self.device = dev;
                        drop(stream);
                        stream = self.setup_stream(&sender)?;
                        stream.play()?;
                    }
                    Commands::Pause => {
                        info!("Pausing Stream...");
                        stream.pause()?;
                    }
                    Commands::Resume => {
                        info!("Playing stream...");
                        stream.play()?;
                    }
                }
            } else {
                error!("Channel closing, stopping threads");
                drop(stream);
                drop(sender);
                thread.join().unwrap();
                return Ok(());
            }
        }
    }

    fn setup_encoder() -> Result<Encoder, Error> {
        let mut encoder = Encoder::new(SAMPLE_RATE, Channels::Stereo, Application::Audio)?;
        encoder.set_bitrate(Bitrate::Max)?;
        encoder.set_complexity(10)?;
        encoder.set_signal(Signal::Music)?;
        Ok(encoder)
    }
}

fn err_fn(error: cpal::Error) {
    error!("Got an error processing audio data: {error}")
}

fn write_input_data<T>(input: &[T], ctx: &Sender<ThreadMessage>)
where
    T: Sample,
    f32: FromSample<T>,
{
    let sample: Vec<f32> = input.iter().map(|s| f32::from_sample(*s)).collect();

    if let Err(err) = ctx.send(ThreadMessage::Data(sample)) {
        error!("Got error when sending data: {err}")
    }
}