rtp-tools: update pacing based on packet size and time it would take to play

aes67.c

@@ -252,12 +252,10 @@ static void aes67_rx_net(struct work_struct *work)
 		return;
 	}
 
-	msg.msg_flags = MSG_WAITFORONE;
+	msg.msg_flags = MSG_DONTWAIT;
 
 	snd_printk(KERN_INFO "Starting network receive loop\n");
-	for (;;) {
+	while (stream->running) {
-		if (!stream->running)
-			break;
 
 		struct kvec iv;
 		iv.iov_base = recv_buf;
@@ -267,10 +265,7 @@ static void aes67_rx_net(struct work_struct *work)
 					iv.iov_len, msg.msg_flags);
 
 		if (msglen == -EAGAIN) {
-			snd_printk(
+			msleep(10);
-				KERN_WARNING
-				"Failed to receive message. With EAGAIN, Sleeping and trying again\n");
-			msleep(1000);
 			continue;
 		}
 
@@ -506,7 +501,7 @@ static int aes67_stream_create(struct aes67_stream **stream, int direction)
 
 	struct sockaddr_in addr = { .sin_family = AF_INET,
 				    .sin_port = htons(9375),
-				    .sin_addr = { htonl(INADDR_LOOPBACK) } };
+				    .sin_addr = { htonl(INADDR_ANY) } };
 
 	strm->socket->ops->bind(strm->socket, (struct sockaddr *)&addr,
 				sizeof(addr));

tools/rtp-tools/.gitignore

@@ -0,0 +1,2 @@
+target/
+Cargo.lock

tools/rtp-tools/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "rtp-tools"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+anyhow = "1.0.87"
+clap = { version = "4.5.17", features = ["derive"] }

tools/rtp-tools/rtp-tools

@@ -0,0 +1 @@
+target/debug/rtp-tools

tools/rtp-tools/src/main.rs

@@ -0,0 +1,102 @@
+use std::{cmp::min, fs::File, io::Read, vec::Vec, time::Duration};
+
+use anyhow::Result;
+
+use clap::{Parser, Subcommand};
+
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+struct Args {
+    #[arg(short, long, value_name = "x.x.x.x:port")]
+    bind: Option<String>,
+
+    #[command(subcommand)]
+    command: Option<Commands>,
+}
+
+#[derive(Subcommand, Debug)]
+enum Commands {
+    Play {
+        file: std::path::PathBuf,
+        addr: String,
+    },
+    Listen,
+}
+
+const MTU: usize = 1500;
+
+fn pacing(sample_rate: u32, bitdepth: usize, size: usize) -> Duration {
+    let samples = size / bitdepth;
+    let seconds = samples as f32 / sample_rate as f32;
+    let just_seconds = seconds.floor();
+    let nanos = (seconds - just_seconds) * 1e9;
+
+    Duration::new(just_seconds as u64, nanos as u32)
+}
+
+fn main() -> Result<()> {
+    let cli = Args::parse();
+
+    let local_addr = match cli.bind {
+        Some(s) => s,
+        None => "0.0.0.0:6767".to_string(),
+    };
+
+    match &cli.command {
+        Some(Commands::Play { file, addr }) => {
+            println!("Playing {file:?} -> {addr}");
+            //open file
+            let mut fd = File::open(file)?;
+
+            //read file into memory
+            let mut file_buffer = Vec::with_capacity(1500 * 4096); //16 packets
+            let buffer_len = fd.read_to_end(&mut file_buffer)?;
+
+            //Construct the size of data allowed
+            //TODO: Support variable MTU. Also calculate RTP packet size
+            let header_len = MTU - 8 - 176;
+            let mut packet_len = buffer_len / header_len;
+            if buffer_len % header_len > 0 {
+                packet_len += 1
+            }
+
+            let mut packets: Vec<Vec<u8>> = vec![Vec::with_capacity(header_len); packet_len];
+            let loop_len = min(buffer_len, header_len);
+
+            'outer: for i in 0..loop_len {
+                for j in 0..packet_len {
+                    let index = i + (header_len * j);
+                    if index == buffer_len {
+                        break 'outer;
+                    }
+                    packets[j].push(file_buffer[index]);
+                }
+            }
+
+            //open socket
+            let udp_socket = std::net::UdpSocket::bind(local_addr)?;
+            udp_socket.connect(addr)?;
+
+            let pacing = pacing(48000, 16, header_len);
+            let mut i = 1;
+            for packet in packets {
+                std::thread::sleep(pacing);
+                print!("Playing chunk {i} of {packet_len}\r");
+                udp_socket.send(&packet)?;
+                i += 1;
+            }
+
+            println!("\nFinished playing all the things.");
+            //stream file to socket
+            Ok(())
+        }
+        Some(Commands::Listen) => {
+            println!("Listen!");
+            Ok(())
+        }
+        None => {
+            println!("Use --help for commands");
+            Ok(())
+        }
+    }
+}