Add metadata streaming using MessagePack and lofty

Signed-off-by: Ivan Bushchik <ivabus@ivabus.dev>

Cargo.toml

@@ -3,8 +3,8 @@ members = ["monoclient", "platform/swiftui/monolib"]
 
 [package]
 name = "lonelyradio"
-description = "TCP radio for singles"
+description = "TCP radio for lonely ones"
-version = "0.1.7"
+version = "0.2.0"
 edition = "2021"
 license = "MIT"
 authors = ["Ivan Bushchik <ivabus@ivabus.dev>"]
@@ -29,7 +29,10 @@ symphonia = { version = "0.5.4", features = [
 ] }
 samplerate = "0.2.4"
 chrono = "0.4"
+rmp-serde = "1.1.2"
+serde = { version = "1.0.197", features = ["derive"] }
+lofty = "0.18.2"
 
 [profile.release]
 opt-level = 3
-lto = true
+#lto = true

README.md

@@ -25,24 +25,12 @@ All files (recursively) will be shuffled and played back. Public log will be dis
 [monoclient](./monoclient) with optional channel separation, hardcoded input (16/44.1/LE).
 
 ```shell
-monoclient <SERVER>:<PORT> s
+monoclient <SERVER>:<PORT>
-```
-
-FFplay (from FFmpeg)
-
-```shell
-nc <SERVER> <PORT> | ffplay -f s16le -vn -ac 2 -ar 44100 -nodisp -autoexit -
-```
-
-MPV
-
-```shell
-nc <SERVER> <PORT> | mpv --audio-channels=stereo --audio-samplerate=44100 --demuxer-rawaudio-format=s16le --demuxer=rawaudio -
 ```
 
 ### Other clients
 
-SwiftUI client is availible in [platform](./platform) directory.
+SwiftUI client is availible in [platform](./platform) directory (not yet adapted for lonelyradio 0.2).
 
 ## License
 

monoclient/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "monoclient"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2021"
 license = "MIT"
 
@@ -8,3 +8,5 @@ license = "MIT"
 clap = { version = "4.4.18", features = ["derive"] }
 rodio = { version = "0.17.3", default-features = false }
 byteorder = "1.5.0"
+rmp-serde = "1.1.2"
+serde = { version = "1.0.197", features = ["derive"] }

monoclient/src/main.rs

@@ -2,9 +2,9 @@ use byteorder::ByteOrder;
 use clap::Parser;
 use rodio::buffer::SamplesBuffer;
 use rodio::{OutputStream, Sink};
-use std::io::Read;
+use serde::Deserialize;
+use std::io::{Read, Write};
 use std::net::TcpStream;
-use std::time::Instant;
 
 // How many samples to cache before playing in samples (both channels) SHOULD BE EVEN
 const BUFFER_SIZE: usize = 2400;
@@ -17,6 +17,15 @@ enum Channel {
 	Stereo,
 }
 
+#[derive(Deserialize, Debug)]
+struct SentMetadata {
+	// In bytes, we need to read next track metadata
+	lenght: u64,
+	title: String,
+	album: String,
+	artist: String,
+}
+
 #[derive(Parser)]
 struct Args {
 	/// Remote address
@@ -28,22 +37,15 @@ struct Args {
 	/// Play only on specified channel, with it if channel = Right => L=0 and R=R, without L=R and R=R. No effect on Stereo
 	single: bool,
 
-	/// More verbose
+	/// Do not erase previously played track from stdout
 	#[arg(short)]
-	verbose: bool,
+	no_backspace: bool,
 }
 
 fn main() {
-	let start = Instant::now();
 	let args = Args::parse();
 	let mut stream = TcpStream::connect(args.address).unwrap();
-	if args.verbose {
+	println!("Connected to {} from {}", stream.peer_addr().unwrap(), stream.local_addr().unwrap());
-		eprintln!(
-			"Connected to {} from {}",
-			stream.peer_addr().unwrap(),
-			stream.local_addr().unwrap()
-		)
-	}
 
 	let channel = match args.channel.to_ascii_lowercase().as_str() {
 		"l" => Channel::Left,
@@ -55,62 +57,86 @@ fn main() {
 	let sink = Sink::try_new(&stream_handle).unwrap();
 	let mut buffer = [0u8; 4];
 	let mut samples = [0f32; BUFFER_SIZE];
-	let mut index = 0usize;
+	let mut latest_msg_len = 0;
-	let mut first = true;
+	print!("Playing: ");
-	while stream.read_exact(&mut buffer).is_ok() {
+	loop {
-		let sample_l = byteorder::LittleEndian::read_i16(&buffer[..2]) as f32 / 32768.0;
+		let mut index = 0usize;
-		let sample_r = byteorder::LittleEndian::read_i16(&buffer[2..]) as f32 / 32768.0;
+
-		// Left channel
+		let md: SentMetadata = rmp_serde::from_read(&stream).unwrap();
-		samples[index] = match channel {
+		let seconds = md.lenght / (2 * 44100);
-			Channel::Left | Channel::Stereo => sample_l,
+		let message = format!(
-			Channel::Right => {
+			"{} - {} - {} ({}:{:02})",
-				if args.single {
+			md.artist,
-					0f32
+			md.album,
-				} else {
+			md.title,
-					sample_r
+			seconds / 60,
-				}
+			seconds % 60
+		);
+		if latest_msg_len != 0 {
+			if args.no_backspace {
+				print!("\nPlaying: ");
+			} else {
+				print!("{}", "\u{8}".repeat(latest_msg_len));
+				print!("{}", " ".repeat(latest_msg_len));
+				print!("{}", "\u{8}".repeat(latest_msg_len));
 			}
-		};
+		}
-		index += 1;
+		print!("{}", message);
-		// Right channel
+		std::io::stdout().flush().unwrap();
-		samples[index] = match channel {
+		latest_msg_len = message.chars().count();
-			Channel::Right | Channel::Stereo => sample_r,
+
-			Channel::Left => {
+		for _ in 0..md.lenght / 2 {
-				if args.single {
+			if stream.read_exact(&mut buffer).is_err() {
-					0f32
+				return;
-				} else {
+			};
-					sample_l
+			let sample_l = byteorder::LittleEndian::read_i16(&buffer[..2]) as f32 / 32768.0;
-				}
+			let sample_r = byteorder::LittleEndian::read_i16(&buffer[2..]) as f32 / 32768.0;
-			}
+			// Left channel
-		};
+			samples[index] = match channel {
-		index += 1;
+				Channel::Left | Channel::Stereo => sample_l,
-		if index == BUFFER_SIZE {
+				Channel::Right => {
-			let mut first_wait_iteration = true;
+					if args.single {
-			// Sink's thread is detached from main thread, so we need to synchronize with it
+						0f32
-			// Why we should synchronize with it?
-			// Let's say, that if we don't synchronize with it, we would have
-			// a lot (no upper limit, actualy) of buffered sound, waiting for playing in sink
-			while sink.len() >= CACHE_SIZE {
-				if args.verbose && first_wait_iteration {
-					eprint!(".");
-					first_wait_iteration = false;
-				}
-				// Sleeping exactly one buffer
-				std::thread::sleep(std::time::Duration::from_secs_f32(
-					(if sink.len() >= 2 {
-						sink.len() - 2
 					} else {
-						1
+						sample_r
-					} as f32) * BUFFER_SIZE as f32
+					}
-						/ 44100.0 / 2.0,
+				}
-				))
+			};
+			index += 1;
+			// Right channel
+			samples[index] = match channel {
+				Channel::Right | Channel::Stereo => sample_r,
+				Channel::Left => {
+					if args.single {
+						0f32
+					} else {
+						sample_l
+					}
+				}
+			};
+			index += 1;
+			if index == BUFFER_SIZE {
+				// Sink's thread is detached from main thread, so we need to synchronize with it
+				// Why we should synchronize with it?
+				// Let's say, that if we don't synchronize with it, we would have
+				// a lot (no upper limit, actualy) of buffered sound, waiting for playing in sink
+				while sink.len() >= CACHE_SIZE {
+					// Sleeping exactly one buffer
+					std::thread::sleep(std::time::Duration::from_secs_f32(
+						(if sink.len() >= 2 {
+							sink.len() - 2
+						} else {
+							1
+						} as f32) * BUFFER_SIZE as f32
+							/ 44100.0 / 2.0,
+					))
+				}
+				sink.append(SamplesBuffer::new(2, 44100, samples.as_slice()));
+				index = 0;
 			}
-			if first && args.verbose {
+		}
-				eprintln!("Started playing in {} ms", (Instant::now() - start).as_millis());
+		while sink.len() != 0 {
-				first = false;
+			std::thread::sleep(std::time::Duration::from_secs_f32(0.01))
-			}
-			sink.append(SamplesBuffer::new(2, 44100, samples.as_slice()));
-			index = 0;
 		}
 	}
 }

src/main.rs

@@ -1,13 +1,18 @@
+use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
 
 use chrono::Local;
 use clap::Parser;
+use lofty::Accessor;
+use lofty::TaggedFileExt;
 use rand::prelude::*;
 use samplerate::ConverterType;
+use serde::Serialize;
 use symphonia::core::audio::SampleBuffer;
 use symphonia::core::codecs::CODEC_TYPE_NULL;
 use symphonia::core::io::MediaSourceStream;
+use symphonia::core::meta::StandardTagKey;
 use symphonia::core::probe::Hint;
 use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpListener, TcpStream};
@@ -26,6 +31,51 @@ struct Args {
 	war: bool,
 }
 
+#[derive(Serialize)]
+struct SentMetadata {
+	// In bytes, we need to read next track metadata
+	lenght: u64,
+	title: String,
+	album: String,
+	artist: String,
+}
+
+async fn stream_samples(
+	track_samples: Vec<f32>,
+	rate: u32,
+	war: bool,
+	md: SentMetadata,
+	s: &mut TcpStream,
+) -> bool {
+	let resampled = if rate != 44100 {
+		samplerate::convert(rate, 44100, 2, ConverterType::Linear, track_samples.as_slice())
+			.unwrap()
+	} else {
+		track_samples
+	};
+	let mut md = md;
+	md.lenght = resampled.len() as u64;
+	if s.write_all(rmp_serde::to_vec(&md).unwrap().as_slice()).await.is_err() {
+		return true;
+	}
+	for sample in resampled {
+		if s.write_all(
+			&(if war {
+				sample.signum() as i16 * 32767
+			} else {
+				(sample * 32768_f32) as i16
+			}
+			.to_le_bytes()),
+		)
+		.await
+		.is_err()
+		{
+			return true;
+		};
+	}
+	false
+}
+
 #[tokio::main]
 async fn main() {
 	let listener = TcpListener::bind(Args::parse().address).await.unwrap();
@@ -35,8 +85,7 @@ async fn main() {
 			.filter_entry(is_not_hidden)
 			.filter_map(|v| v.ok())
 			.map(|x| x.into_path())
-			.filter(track_valid)
+			.filter(|x| track_valid(x))
-			.into_iter()
 			.collect::<Vec<PathBuf>>(),
 	);
 	loop {
@@ -48,7 +97,7 @@ fn is_not_hidden(entry: &DirEntry) -> bool {
 	entry.file_name().to_str().map(|s| entry.depth() == 0 || !s.starts_with('.')).unwrap_or(false)
 }
 
-fn track_valid(track: &PathBuf) -> bool {
+fn track_valid(track: &Path) -> bool {
 	if !track.metadata().unwrap().is_file() {
 		return false;
 	}
@@ -63,12 +112,24 @@ fn track_valid(track: &PathBuf) -> bool {
 async fn stream(mut s: TcpStream, tracklist: Arc<Vec<PathBuf>>) {
 	let args = Args::parse();
 
-	'track: loop {
+	loop {
 		let track = tracklist.choose(&mut thread_rng()).unwrap();
+
+		let mut title = String::new();
+		let mut artist = String::new();
+		let mut album = String::new();
+		let mut file = std::fs::File::open(track).unwrap();
+		let tagged = lofty::read_from(&mut file).unwrap();
+		if let Some(id3v2) = tagged.primary_tag() {
+			title = id3v2.title().unwrap_or("[No tag]".into()).to_string();
+			album = id3v2.album().unwrap_or("[No tag]".into()).to_string();
+			artist = id3v2.artist().unwrap_or("[No tag]".into()).to_string()
+		};
+		let track_message = format!("{} - {} - {}", &artist, &album, &title);
 		eprintln!(
 			"[{}] {} to {}:{}{}",
 			Local::now().to_rfc3339(),
-			track.to_str().unwrap(),
+			track_message,
 			s.peer_addr().unwrap().ip(),
 			s.peer_addr().unwrap().port(),
 			if args.war {
@@ -118,15 +179,27 @@ async fn stream(mut s: TcpStream, tracklist: Arc<Vec<PathBuf>>) {
 			.expect("unsupported codec");
 
 		let track_id = track.id;
-
+		let mut track_rate = 0;
+		let mut samples = vec![];
 		loop {
 			let packet = match format.next_packet() {
 				Ok(packet) => packet,
-				_ => continue 'track,
+				_ => break,
 			};
-
 			while !format.metadata().is_latest() {
 				format.metadata().pop();
+				if let Some(rev) = format.metadata().current() {
+					for tag in rev.tags() {
+						println!("Looped");
+						match tag.std_key {
+							Some(StandardTagKey::Album) => album = tag.value.to_string(),
+							Some(StandardTagKey::Artist) => artist = tag.value.to_string(),
+							Some(StandardTagKey::TrackTitle) => title = tag.value.to_string(),
+							_ => {}
+						}
+						eprintln!("DBG: {} {} {}", &album, &artist, &title)
+					}
+				}
 			}
 
 			if packet.track_id() != track_id {
@@ -135,47 +208,36 @@ async fn stream(mut s: TcpStream, tracklist: Arc<Vec<PathBuf>>) {
 
 			match decoder.decode(&packet) {
 				Ok(decoded) => {
-					let rate = decoded.spec().rate;
+					track_rate = decoded.spec().rate;
 					let mut byte_buf =
 						SampleBuffer::<f32>::new(decoded.capacity() as u64, *decoded.spec());
 					byte_buf.copy_interleaved_ref(decoded);
-					let samples = if rate != 44100 {
+					samples.append(&mut byte_buf.samples_mut().to_vec());
-						samplerate::convert(
-							rate,
-							44100,
-							2,
-							ConverterType::Linear,
-							byte_buf.samples(),
-						)
-						.unwrap()
-					} else {
-						byte_buf.samples().to_vec()
-					};
-					for sample in samples {
-						let result = s
-							.write(
-								&(if args.war {
-									sample.signum() as i16 * 32767
-								} else {
-									(sample * 32768_f32) as i16
-								})
-								.to_le_bytes(),
-							)
-							.await;
-						match result {
-							Err(_) | Ok(0) => {
-								return;
-							}
-							_ => (),
-						};
-					}
 					continue;
 				}
 				_ => {
 					// Handling any error as track skip
-					continue 'track;
+					continue;
 				}
 			}
 		}
+		if !samples.is_empty() {
+			if stream_samples(
+				samples,
+				track_rate,
+				args.war,
+				SentMetadata {
+					lenght: 0,
+					title,
+					album,
+					artist,
+				},
+				&mut s,
+			)
+			.await
+			{
+				break;
+			}
+		}
 	}
 }