joeyahines/tap
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Switch to using protobuf and remove audio sink

Browse Source
This commit is contained in:
joeyahines committed 2025-08-31 15:01:18 -06:00
1 parent 1f250f448d
commit 3e6552d012
13 files changed
+224 -505
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/audio_record.rs
+11 -12
View File
@@ -1,17 +1,16 @@
use crate::tap_packet::TapPacketHeader;
use crate::udp_connection::OpusUdpConnection;
use crate::util::{
SAMPLE_RATE, SAMPLE_RATE_RAW, STEREO_FRAME_SIZE, TAP_PACKET_MAX_DATA_LEN,
};
use crate::util::{SAMPLE_RATE, SAMPLE_RATE_RAW, STEREO_FRAME_SIZE, TAP_PACKET_MAX_DATA_LEN};
use anyhow::Error;
use audiopus::coder::Decoder;
use audiopus::packet::Packet;
use audiopus::{Channels, MutSignals};
use chrono::Utc;
use log::{error, warn};
use prost::Message;
use std::io::{ErrorKind, Read};
use std::net::SocketAddr;
use std::path::Path;
use tap_interface::tap::message::TapMessage;
pub fn record(addr: SocketAddr, output_path: &Path, duration_ms: u64) -> Result<(), Error> {
let mut decoder = Decoder::new(SAMPLE_RATE, Channels::Stereo)?;
@@ -32,7 +31,7 @@ pub fn record(addr: SocketAddr, output_path: &Path, duration_ms: u64) -> Result<
let deadline = std::time::Instant::now() + std::time::Duration::from_millis(duration_ms);
while std::time::Instant::now() < deadline {
let mut packet = vec![0; TAP_PACKET_MAX_DATA_LEN + TapPacketHeader::serialized_size()];
let mut packet = vec![0; TAP_PACKET_MAX_DATA_LEN];
let mut output_buffer = vec![0.0; STEREO_FRAME_SIZE];
let signals = MutSignals::try_from(&mut output_buffer)?;
@@ -49,26 +48,26 @@ pub fn record(addr: SocketAddr, output_path: &Path, duration_ms: u64) -> Result<
};
let data_size = if let Some(new_packet) = new_packet {
let hdr = TapPacketHeader::from_buffer(new_packet)?;
let tap_message = TapMessage::decode(new_packet)?;
if !hdr.check_acceptable_latency_tolerance() {
if tap_message.timestamp - Utc::now().timestamp() as u64 > 1 {
warn!(
"Packet outside of tolerance: packet.timestamp={} time={}",
hdr.timestamp,
tap_message.timestamp,
Utc::now().timestamp()
);
}
if hdr.seq != seq_count {
if tap_message.seq != seq_count {
warn!(
"Found sequence mismatch, expected seq={}, {hdr:?}",
"Found sequence mismatch, expected seq={}, {tap_message:?}",
seq_count
);
}
(seq_count, _) = hdr.seq.overflowing_add(1);
(seq_count, _) = tap_message.seq.overflowing_add(1);
let packet = Packet::try_from(&new_packet[TapPacketHeader::serialized_size()..])?;
let packet = Packet::try_from(&tap_message.audio_data)?;
decoder.decode_float(Some(packet), signals, false)?
} else {
src/audio_sink.rs
-169
View File
@@ -1,169 +0,0 @@
use crate::input_buffer::InputBuffer;
use crate::udp_connection::OpusUdpConnection;
use crate::util::{CIRC_BUFFER_SIZE, SAMPLE_RATE, SAMPLE_RATE_RAW, STEREO_FRAME_SIZE};
use anyhow::Error;
use audiopus::Channels;
use audiopus::coder::Decoder;
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{
BufferSize, Device, FromSample, Host, I24, Sample, SampleRate, SizedSample, StreamConfig,
SupportedStreamConfig,
};
use log::{error, info};
use std::io::{ErrorKind, Read};
use std::net::SocketAddr;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;
const INPUT_BUFFER_SIZE: usize = CIRC_BUFFER_SIZE;
#[derive(Debug, Clone)]
struct Context {
pub addr: SocketAddr,
pub input_buffer: Arc<Mutex<InputBuffer<INPUT_BUFFER_SIZE>>>,
pub data_rady: Arc<Condvar>,
}
pub fn listen(device: &str, addr: SocketAddr, host: Host) -> Result<(), Error> {
// Set up the input device and stream with the default input config.
let device = if device == "default" {
host.default_output_device()
} else {
host.output_devices()?
.find(|x| x.name().map(|y| y == device).unwrap_or(false))
}
.expect("failed to find output device");
info!("Output device: {}", device.name()?);
let config = device.default_output_config()?;
println!("Default output config : {config:?}");
for cfg in device.supported_output_configs()? {
info!("{:?}", cfg);
}
let decoder = Decoder::new(SAMPLE_RATE, Channels::Stereo)?;
let input_buffer = InputBuffer::new(STEREO_FRAME_SIZE, decoder);
let ctx = Context {
addr,
input_buffer: Arc::new(Mutex::new(input_buffer)),
data_rady: Arc::new(Condvar::new()),
};
let stream = stream_setup_for(device, config, ctx)?;
stream.play()?;
thread::park();
Ok(())
}
fn stream_setup_for(
device: Device,
default_config: SupportedStreamConfig,
ctx: Context,
) -> Result<cpal::Stream, anyhow::Error>
where
{
let mut config: StreamConfig = default_config.clone().into();
config.sample_rate = SampleRate(SAMPLE_RATE_RAW as u32);
config.buffer_size = BufferSize::Fixed(STEREO_FRAME_SIZE as u32);
match default_config.sample_format() {
cpal::SampleFormat::I8 => make_stream::<i8>(&device, &config, ctx),
cpal::SampleFormat::I16 => make_stream::<i16>(&device, &config, ctx),
cpal::SampleFormat::I24 => make_stream::<I24>(&device, &config, ctx),
cpal::SampleFormat::I32 => make_stream::<i32>(&device, &config, ctx),
cpal::SampleFormat::I64 => make_stream::<i64>(&device, &config, ctx),
cpal::SampleFormat::U8 => make_stream::<u8>(&device, &config, ctx),
cpal::SampleFormat::U16 => make_stream::<u16>(&device, &config, ctx),
cpal::SampleFormat::U32 => make_stream::<u32>(&device, &config, ctx),
cpal::SampleFormat::U64 => make_stream::<u64>(&device, &config, ctx),
cpal::SampleFormat::F32 => make_stream::<f32>(&device, &config, ctx),
cpal::SampleFormat::F64 => make_stream::<f64>(&device, &config, ctx),
sample_format => Err(anyhow::Error::msg(format!(
"Unsupported sample format '{sample_format}'"
))),
}
}
fn make_stream<T>(
device: &Device,
config: &cpal::StreamConfig,
ctx: Context,
) -> Result<cpal::Stream, anyhow::Error>
where
T: SizedSample + FromSample<f32>,
{
let err_fn = |err| error!("Error building output sound stream: {err}");
let time_at_start = std::time::Instant::now();
info!("Time at start: {time_at_start:?}");
let ctx_recv_thread = ctx.clone();
thread::spawn(move || rx_thread(ctx_recv_thread));
let stream = device.build_output_stream(
config,
move |output: &mut [T], _: &cpal::OutputCallbackInfo| process_frame(output, &ctx),
err_fn,
None,
)?;
Ok(stream)
}
fn rx_thread(ctx: Context) {
info!("Starting RX thread...");
let mut udp_connection = OpusUdpConnection::new(ctx.addr, ctx.addr).unwrap();
udp_connection
.set_read_timeout(Some(Duration::from_millis(20)))
.unwrap();
let mut packet = vec![0; 32 * 1024];
loop {
let new_packet = match udp_connection.read(&mut packet) {
Ok(data_len) => Some(&packet[0..data_len]),
Err(err) => {
if err.kind() == ErrorKind::WouldBlock {
None
} else {
error!("Got error trying to recv from UDP: {err}");
continue;
}
}
};
let mut input_buffer = ctx.input_buffer.lock().expect("Mutex poisoned");
if let Err(err) = input_buffer.recv_packet(new_packet) {
error!("Adding packet to buffer: {err}")
} else {
ctx.data_rady.notify_all();
}
}
}
fn process_frame<SampleType>(output: &mut [SampleType], ctx: &Context)
where
SampleType: Sample + FromSample<f32>,
{
let mut input_buffer = ctx.input_buffer.lock().expect("Mutex poisoned");
let mut input_buffer = if input_buffer.samples_in_buffer() < output.len() {
while input_buffer.samples_in_buffer() < output.len() {
input_buffer = ctx.data_rady.wait(input_buffer).expect("Mutex poisoned");
}
input_buffer
} else {
input_buffer
};
input_buffer.read_frame_from_buffer(output)
}
src/audio_source.rs
+1 -3
View File
@@ -1,8 +1,6 @@
use crate::tx_thread::TxThread;
use crate::udp_connection::OpusUdpConnection;
use crate::util::{
SAMPLE_RATE, SAMPLE_RATE_RAW, STEREO_FRAME_SIZE, ThreadMessage,
};
use crate::util::{SAMPLE_RATE, SAMPLE_RATE_RAW, STEREO_FRAME_SIZE, ThreadMessage};
use anyhow::Error;
use audiopus::coder::Encoder;
use audiopus::{Application, Bitrate, Channels, Signal};
src/input_buffer.rs
-110
View File
@@ -1,110 +0,0 @@
use crate::tap_packet::{TapPacketError, TapPacketHeader};
use audiopus::MutSignals;
use audiopus::coder::Decoder;
use audiopus::packet::Packet;
use chrono::Utc;
use circular_buffer::CircularBuffer;
use cpal::{FromSample, Sample};
use log::warn;
use thiserror::Error;
#[derive(Error, Debug)]
#[allow(clippy::enum_variant_names)]
pub enum Error {
#[error("IO error on Output: {0}")]
IoError(#[from] std::io::Error),
#[error("Decode error: {0}")]
DecodeError(#[from] audiopus::Error),
#[error("Tap packet error: {0}")]
TapPacketError(#[from] TapPacketError),
}
#[derive(Debug)]
pub struct InputBuffer<const N: usize> {
circular_buffer: CircularBuffer<N, f32>,
frame_size: usize,
decoder: Decoder,
expected_seq: u32,
}
impl<const N: usize> InputBuffer<N> {
pub fn new(frame_size: usize, decoder: Decoder) -> Self {
Self {
circular_buffer: CircularBuffer::new(),
frame_size,
decoder,
expected_seq: 0,
}
}
fn decode(&mut self, input_buffer: &[u8], output_buffer: &mut [f32]) -> Result<usize, Error> {
let hdr = TapPacketHeader::from_buffer(input_buffer)?;
if !hdr.check_acceptable_latency_tolerance() {
warn!(
"Packet outside of tolerance: packet.timestamp={} time={}",
hdr.timestamp,
Utc::now().timestamp()
);
}
if hdr.seq != self.expected_seq {
warn!(
"Found sequence mismatch, expected seq={}, {hdr:?}",
self.expected_seq
);
}
(self.expected_seq, _) = hdr.seq.overflowing_add(1);
let packet = Packet::try_from(&input_buffer[TapPacketHeader::serialized_size()..])?;
let signals = MutSignals::try_from(output_buffer)?;
Ok(self.decoder.decode_float(Some(packet), signals, false)?)
}
fn missing_packet(&mut self, output_buffer: &mut [f32]) -> Result<usize, Error> {
let signals = MutSignals::try_from(output_buffer)?;
Ok(self.decoder.decode_float(None, signals, false)?)
}
fn write_frame_to_buffer(&mut self, frame: &[f32]) {
self.circular_buffer.extend_from_slice(frame)
}
pub fn recv_packet(&mut self, input_buffer: Option<&[u8]>) -> Result<(), Error> {
let mut output_buffer = vec![0.0; self.frame_size];
let size = match input_buffer {
Some(buf) => self.decode(buf, &mut output_buffer)?,
None => self.missing_packet(&mut output_buffer)?,
} * 2;
self.write_frame_to_buffer(&output_buffer[..size]);
Ok(())
}
pub fn samples_in_buffer(&self) -> usize {
self.circular_buffer.len()
}
pub fn read_frame_from_buffer<SampleType: Sample + FromSample<f32>>(
&mut self,
output_buffer: &mut [SampleType],
) {
let drain_count = output_buffer.len().min(self.circular_buffer.len());
let frame: Vec<f32> = self.circular_buffer.drain(..drain_count).collect();
for (ndx, output) in output_buffer.iter_mut().enumerate() {
let value: SampleType = if ndx < frame.len() {
SampleType::from_sample(frame[ndx])
} else {
SampleType::from_sample(0.0f32)
};
*output = value;
}
}
}
src/main.rs
-15
View File
@@ -1,15 +1,10 @@
mod audio_record;
mod audio_sink;
mod audio_source;
mod input_buffer;
mod rx_thread;
mod tap_packet;
mod tx_thread;
mod udp_connection;
mod util;
use crate::audio_record::record;
use crate::audio_sink::listen;
use clap::{Parser, Subcommand};
use log::LevelFilter;
use std::net::SocketAddr;
@@ -35,15 +30,6 @@ enum Tasks {
dest: SocketAddr,
},
/// Listen to data from a tap
Listen {
/// The audio device to use
#[arg(short, long, default_value_t = String::from("default"))]
device: String,
/// UDP addr to receive data on
addr: SocketAddr,
},
Record {
/// UDP addr to receive data on
addr: SocketAddr,
@@ -65,7 +51,6 @@ fn main() -> Result<(), anyhow::Error> {
match opt.command {
Tasks::Tap { device, src, dest } => audio_source::tap(&device, src, dest, host),
Tasks::Listen { device, addr } => listen(&device, addr, host),
Tasks::Record {
addr,
duration_ms,
src/rx_thread.rs
-127
View File
@@ -1,127 +0,0 @@
use crate::tap_packet::TapPacketError;
use crate::udp_connection::OpusUdpConnection;
use crate::util::{CIRC_BUFFER_SIZE, ThreadMessage};
use audiopus::coder::Decoder;
use circular_buffer::CircularBuffer;
use std::sync::mpsc::Sender;
use thiserror::Error;
#[derive(Error, Debug)]
#[allow(clippy::enum_variant_names)]
pub enum Error {
#[error("IO error on Output: {0}")]
IoError(#[from] std::io::Error),
#[error("Decode error: {0}")]
DecodeError(#[from] audiopus::Error),
#[error("Tap packet error: {0}")]
TapPacketError(#[from] TapPacketError),
}
pub struct RxThread {
sender: Sender<ThreadMessage>,
decoder: Decoder,
opus_udp_connection: OpusUdpConnection,
circular_buffer: CircularBuffer<CIRC_BUFFER_SIZE, f32>,
expected_seq: u32,
exit: bool,
}
//impl RxThread {
// pub fn new(sender: Sender<ThreadMessage>, decoder: Decoder, opus_udp_connection: OpusUdpConnection) -> Self {
// Self {
// sender,
// decoder,
// opus_udp_connection,
// circular_buffer: Default::default(),
// expected_seq: 0,
// exit: false,
// }
// }
//
// fn decode(&mut self, input_buffer: &[u8], output_buffer: &mut [f32]) -> Result<usize, Error> {
// let hdr = TapPacketHeader::from_buffer(input_buffer)?;
//
// if !hdr.check_acceptable_latency_tolerance() {
// warn!(
// "Packet outside of tolerance: packet.timestamp={} time={}",
// hdr.timestamp,
// Utc::now().timestamp()
// );
// }
//
// if hdr.seq != self.expected_seq {
// warn!(
// "Found sequence mismatch, expected seq={}, {hdr:?}",
// self.expected_seq
// );
// }
//
// (self.expected_seq, _) = hdr.seq.overflowing_add(1);
//
// let packet = Packet::try_from(&input_buffer[TapPacketHeader::serialized_size()..])?;
//
// let signals = MutSignals::try_from(output_buffer)?;
// Ok(self.decoder.decode_float(Some(packet), signals, false)?)
// }
//
// fn missing_packet(&mut self, output_buffer: &mut [f32]) -> Result<usize, Error> {
// let signals = MutSignals::try_from(output_buffer)?;
// Ok(self.decoder.decode_float(None, signals, false)?)
// }
//
// fn write_frame_to_buffer(&mut self, frame: &[f32]) {
// self.circular_buffer.extend_from_slice(frame)
// }
//
//
// pub fn recv_packet(&mut self, input_buffer: Option<&[u8]>) -> Result<(), Error> {
// let mut output_buffer = vec![0.0; STEREO_FRAME_SIZE];
//
// let size = match input_buffer {
// Some(buf) => self.decode(buf, &mut output_buffer)?,
// None => self.missing_packet(&mut output_buffer)?,
// };
//
// self.write_frame_to_buffer(&output_buffer[..size]);
//
// Ok(())
// }
//
// pub fn send_next_frame(
// &mut self,
// ) {
// let drain_count = output_buffer.len().min(self.circular_buffer.len());
//
// let frame: Vec<f32> = self.circular_buffer.drain(..drain_count).collect();
//
// for (ndx, output) in output_buffer.iter_mut().enumerate() {
// let value: SampleType = if ndx < frame.len() {
// SampleType::from_sample(frame[ndx])
// } else {
// SampleType::from_sample(0.0f32)
// };
//
// *output = value;
// }
// }
//
// fn handle_udp_msg(&mut self) -> Result<(), Error> {
// let mut udp_packet = vec![0; TAP_PACKET_MAX_DATA_LEN + TapPacketHeader::serialized_size()];
//
// let size = self.opus_udp_connection.read(&mut udp_packet)?;
//
// let udp_packet = if size == 0 {
// None
// }
// else {
// Some(udp_packet.as_slice())
// };
//
// self.recv_packet(udp_packet)?;
//
//
// Ok(())
// }
//
// pub fn thread(self)
//}
src/tap_packet.rs
-52
View File
@@ -1,52 +0,0 @@
use chrono::Utc;
use thiserror::Error;
#[derive(Error, Debug)]
pub enum TapPacketError {
#[error("Buffer is too small to serialize data into")]
BufTooSmall,
}
#[derive(Debug, Clone)]
pub struct TapPacketHeader {
pub seq: u32,
pub timestamp: i64,
}
impl TapPacketHeader {
pub fn write_to_buffer(&self, buf: &mut [u8]) -> Result<(), TapPacketError> {
if buf.len() < Self::serialized_size() {
return Err(TapPacketError::BufTooSmall);
}
buf[0..size_of::<u32>()].copy_from_slice(self.seq.to_be_bytes().as_slice());
buf[size_of::<u32>()..Self::serialized_size()]
.copy_from_slice(self.timestamp.to_be_bytes().as_slice());
Ok(())
}
pub fn check_acceptable_latency_tolerance(&self) -> bool {
Utc::now().timestamp().abs_diff(self.timestamp) < 1
}
pub fn from_buffer(buf: &[u8]) -> Result<Self, TapPacketError> {
if buf.len() < Self::serialized_size() {
return Err(TapPacketError::BufTooSmall);
}
let mut seq_bytes = [0u8; size_of::<u32>()];
seq_bytes.copy_from_slice(&buf[0..size_of::<u32>()]);
let seq = u32::from_be_bytes(seq_bytes);
let mut timestamp_bytes = [0u8; size_of::<i64>()];
timestamp_bytes.copy_from_slice(&buf[size_of::<u32>()..Self::serialized_size()]);
let timestamp = i64::from_be_bytes(timestamp_bytes);
Ok(Self { seq, timestamp })
}
pub fn serialized_size() -> usize {
size_of::<u32>() + size_of::<i64>()
}
}
src/tx_thread.rs
+13 -17
View File
@@ -1,26 +1,24 @@
use crate::tap_packet::{TapPacketError, TapPacketHeader};
use crate::udp_connection::OpusUdpConnection;
use crate::util::STEREO_FRAME_SIZE;
use crate::util::{CIRC_BUFFER_SIZE, TAP_PACKET_MAX_DATA_LEN, ThreadMessage};
use audiopus::coder::Encoder;
use chrono::Utc;
use circular_buffer::CircularBuffer;
use log::{error, info, warn};
use prost::Message;
use std::io::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("Buffer too small to handle error")]
BufferTooSmall,
#[error("IO error on offload: {0}")]
IoError(#[from] std::io::Error),
IoFailure(#[from] std::io::Error),
#[error("Encode error: {0}")]
EncodeError(#[from] audiopus::Error),
#[error("Tap packet error: {0}")]
TapPacketError(#[from] TapPacketError),
EncodeFailure(#[from] audiopus::Error),
}
pub struct TxThread {
@@ -54,21 +52,19 @@ impl TxThread {
}
fn send_opus_frame(&mut self) -> Result<(), Error> {
let mut output_buffer =
vec![0; TAP_PACKET_MAX_DATA_LEN + TapPacketHeader::serialized_size()];
let size = self.encode(&mut output_buffer[TapPacketHeader::serialized_size()..])?;
let mut output_buffer = vec![0; TAP_PACKET_MAX_DATA_LEN];
let size = self.encode(&mut output_buffer)?;
let hdr = TapPacketHeader {
output_buffer.truncate(size);
let tap_message = TapMessage {
seq: self.seq_count,
timestamp: chrono::Utc::now().timestamp(),
timestamp: Utc::now().timestamp() as u64,
audio_data: output_buffer,
};
hdr.write_to_buffer(&mut output_buffer)?;
let total_size = TapPacketHeader::serialized_size() + size;
self.opus_udp_connection
.write_all(&output_buffer[0..total_size])?;
.write_all(tap_message.encode_to_vec().as_slice())?;
(self.seq_count, _) = self.seq_count.overflowing_add(1);
Ok(())
src/udp_connection.rs
+1
View File
@@ -16,6 +16,7 @@ impl OpusUdpConnection {
})
}
#[allow(dead_code)]
pub fn set_read_timeout(&mut self, timeout: Option<Duration>) -> std::io::Result<()> {
self.udp_socket.set_read_timeout(timeout)
}
src/util.rs
+1
View File
@@ -9,6 +9,7 @@ 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,