refactored tap core into its own crate to eventually support a gui

tap_core/src/audio_source_thread.rs

@@ -0,0 +1,171 @@
+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(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}")
+    }
+}

tap_core/src/lib.rs

@@ -0,0 +1,3 @@
+pub mod audio_source_thread;
+pub mod tx_thread;
+pub mod util;

tap_core/src/tx_thread.rs

@@ -0,0 +1,102 @@
+use crate::util::STEREO_FRAME_SIZE;
+use crate::util::{CIRC_BUFFER_SIZE, TAP_PACKET_MAX_DATA_LEN, ThreadMessage};
+use chrono::Utc;
+use circular_buffer::FixedCircularBuffer;
+use log::{error, info, warn};
+use opus::Encoder;
+use prost::Message;
+use std::io::{Read, Write};
+use std::sync::mpsc::{Receiver, RecvError};
+use tap_interface::tap::message::TapMessage;
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("Failed to receive message: {0}")]
+    RecvFailed(#[from] RecvError),
+    #[error("IO error on offload: {0}")]
+    IoFailure(#[from] std::io::Error),
+    #[error("Encode error: {0}")]
+    EncodeFailure(#[from] opus::Error),
+}
+
+pub struct TxThread<T: Read + Write> {
+    receiver: Receiver<ThreadMessage>,
+    encoder: Encoder,
+    opus_connection: T,
+    circular_buffer: FixedCircularBuffer<f32, CIRC_BUFFER_SIZE>,
+    seq_count: u32,
+    exit: bool,
+}
+
+impl<T: Read + Write> TxThread<T> {
+    pub fn new(receiver: Receiver<ThreadMessage>, encoder: Encoder, opus_connection: T) -> Self {
+        Self {
+            receiver,
+            encoder,
+            opus_connection,
+            circular_buffer: Default::default(),
+            seq_count: 0,
+            exit: false,
+        }
+    }
+
+    fn encode(&mut self, output_buffer: &mut [u8]) -> Result<usize, Error> {
+        let encode_buffer: Vec<f32> = self.circular_buffer.drain(0..STEREO_FRAME_SIZE).collect();
+        Ok(self.encoder.encode_float(&encode_buffer, output_buffer)?)
+    }
+
+    fn send_opus_frame(&mut self) -> Result<(), Error> {
+        let mut output_buffer = vec![0; TAP_PACKET_MAX_DATA_LEN];
+        let size = self.encode(&mut output_buffer)?;
+
+        output_buffer.truncate(size);
+
+        let tap_message = TapMessage {
+            seq: self.seq_count,
+            timestamp: Utc::now().timestamp() as u64,
+            audio_data: output_buffer,
+        };
+
+        self.opus_connection
+            .write_all(tap_message.encode_to_vec().as_slice())?;
+
+        (self.seq_count, _) = self.seq_count.overflowing_add(1);
+        Ok(())
+    }
+
+    fn handle_data_msg(&mut self, data: Vec<f32>) -> Result<(), Error> {
+        if self.circular_buffer.len() + data.len() > CIRC_BUFFER_SIZE {
+            warn!("Will drop data in circular buffer");
+        }
+
+        self.circular_buffer.extend_from_slice(&data);
+
+        if self.circular_buffer.len() >= STEREO_FRAME_SIZE {
+            self.send_opus_frame()?;
+        }
+
+        Ok(())
+    }
+
+    fn handle_msg(&mut self) -> Result<(), Error> {
+        let msg = self.receiver.recv()?;
+
+        match msg {
+            ThreadMessage::Data(data) => self.handle_data_msg(data),
+            ThreadMessage::Exit => {
+                self.exit = true;
+                Ok(())
+            }
+        }
+    }
+
+    pub fn thread(mut self) {
+        info!("Starting TX Thread...");
+        loop {
+            if let Err(err) = self.handle_msg() {
+                error!("Failed to process message: {err:?}")
+            }
+        }
+    }
+}

tap_core/src/util.rs

@@ -0,0 +1,17 @@
+use cpal::SampleRate;
+
+pub const SAMPLE_RATE_RAW: usize = 48_000;
+
+pub const SAMPLE_RATE: SampleRate = SAMPLE_RATE_RAW as u32;
+pub const AUDIO_FRAME_RATE: usize = 50;
+pub const MONO_FRAME_SIZE: usize = SAMPLE_RATE_RAW / AUDIO_FRAME_RATE;
+pub const STEREO_FRAME_SIZE: usize = 2 * MONO_FRAME_SIZE;
+pub const CIRC_BUFFER_SIZE: usize = STEREO_FRAME_SIZE * 10;
+pub const TAP_PACKET_MAX_DATA_LEN: usize = 32 * 1024;
+
+#[derive(Debug, Clone)]
+#[allow(dead_code)]
+pub enum ThreadMessage {
+    Data(Vec<f32>),
+    Exit,
+}