#![cfg_attr(not(target_os = "macos"), allow(dead_code))]
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream, UdpSocket};
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
pub const LOOPBACK_NAME: &str = "blackhole";
pub const INSTALL_HINT: &str =
"install the loopback driver first: brew install --cask blackhole-2ch";
const BITRATE: &str = "256k";
pub struct Session {
port: u16,
host_ip: String,
shutdown: Arc<AtomicBool>,
children: Arc<Mutex<Vec<Child>>>,
previous_output: Option<u32>,
}
impl Session {
pub fn sonos_uri(&self) -> String {
format!(
"x-rincon-mp3radio://{}:{}/mnml.mp3",
self.host_ip, self.port
)
}
pub fn is_alive(&self) -> bool {
!self.shutdown.load(Ordering::Relaxed)
}
pub fn stop(self) {
self.shutdown.store(true, Ordering::Relaxed);
let _ = TcpStream::connect((self.host_ip.as_str(), self.port));
if let Ok(mut children) = self.children.lock() {
for mut child in children.drain(..) {
let _ = child.kill();
let _ = child.wait();
}
}
#[cfg(target_os = "macos")]
if let Some(id) = self.previous_output {
let _ = super::coreaudio::set_default_output(id);
}
}
#[cfg(target_os = "macos")]
pub fn start(_room: &str, player_host: &str) -> Result<Self, String> {
let device = super::coreaudio::find_output(LOOPBACK_NAME)
.ok_or_else(|| format!("no loopback audio device found — {INSTALL_HINT}"))?;
let ffmpeg = ffmpeg_bin().ok_or("ffmpeg not found on PATH")?;
let index = avfoundation_index(&ffmpeg, &device.name)?;
let host_ip = local_ip_towards(player_host)
.ok_or("could not work out this Mac's address on the network")?;
let listener = TcpListener::bind((host_ip.as_str(), 0))
.map_err(|e| format!("could not open a local stream port: {e}"))?;
let port = listener
.local_addr()
.map_err(|e| format!("could not read the stream port: {e}"))?
.port();
let previous_output = super::coreaudio::default_output().map(|d| d.id);
super::coreaudio::set_default_output(device.id)?;
let shutdown = Arc::new(AtomicBool::new(false));
let children: Arc<Mutex<Vec<Child>>> = Arc::new(Mutex::new(Vec::new()));
let allowed_peer = player_host.parse::<std::net::IpAddr>().ok();
serve(
listener,
ffmpeg,
index,
shutdown.clone(),
children.clone(),
allowed_peer,
);
Ok(Session {
port,
host_ip,
shutdown,
children,
previous_output,
})
}
#[cfg(not(target_os = "macos"))]
pub fn start(_room: &str, _player_host: &str) -> Result<Self, String> {
Err("streaming this Mac's audio to a Sonos is macOS-only today".to_string())
}
}
fn serve(
listener: TcpListener,
ffmpeg: PathBuf,
index: String,
shutdown: Arc<AtomicBool>,
children: Arc<Mutex<Vec<Child>>>,
allowed_peer: Option<std::net::IpAddr>,
) {
std::thread::spawn(move || {
for conn in listener.incoming() {
if shutdown.load(Ordering::Relaxed) {
break;
}
let Ok(conn) = conn else { continue };
if !peer_allowed(&conn, allowed_peer) {
continue;
}
let (ffmpeg, index) = (ffmpeg.clone(), index.clone());
let (shutdown, children) = (shutdown.clone(), children.clone());
std::thread::spawn(move || {
pump(conn, &ffmpeg, &index, &shutdown, &children);
});
}
});
}
fn peer_allowed(conn: &TcpStream, allowed_peer: Option<std::net::IpAddr>) -> bool {
let (Some(allowed), Ok(peer)) = (allowed_peer, conn.peer_addr()) else {
return false;
};
peer.ip() == allowed
}
fn pump(
mut conn: TcpStream,
ffmpeg: &PathBuf,
index: &str,
shutdown: &Arc<AtomicBool>,
children: &Arc<Mutex<Vec<Child>>>,
) {
let mut scratch = [0u8; 1024];
let _ = conn.read(&mut scratch);
let headers = "HTTP/1.0 200 OK\r\n\
Content-Type: audio/mpeg\r\n\
Cache-Control: no-cache, no-store\r\n\
icy-name: mnml — this Mac\r\n\
Connection: close\r\n\r\n";
if conn.write_all(headers.as_bytes()).is_err() {
return;
}
let Ok(mut child) = Command::new(ffmpeg)
.args(encoder_args(index))
.stdout(Stdio::piped())
.stderr(Stdio::null())
.stdin(Stdio::null())
.spawn()
else {
return;
};
let Some(mut out) = child.stdout.take() else {
let _ = child.kill();
return;
};
let child_id = child.id();
if let Ok(mut guard) = children.lock() {
guard.push(child);
}
let mut buf = [0u8; 8192];
while !shutdown.load(Ordering::Relaxed) {
match out.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
if conn.write_all(&buf[..n]).is_err() {
break; }
}
Err(_) => break,
}
}
if let Ok(mut guard) = children.lock()
&& let Some(pos) = guard.iter().position(|c| c.id() == child_id)
{
let mut child = guard.remove(pos);
let _ = child.kill();
let _ = child.wait();
}
}
fn encoder_args(index: &str) -> Vec<String> {
vec![
"-hide_banner".into(),
"-loglevel".into(),
"error".into(),
"-f".into(),
"avfoundation".into(),
"-i".into(),
format!(":{index}"),
"-ac".into(),
"2".into(),
"-ar".into(),
"44100".into(),
"-c:a".into(),
"libmp3lame".into(),
"-b:a".into(),
BITRATE.into(),
"-flush_packets".into(),
"1".into(),
"-f".into(),
"mp3".into(),
"pipe:1".into(),
]
}
fn ffmpeg_bin() -> Option<PathBuf> {
if let Some(paths) = std::env::var_os("PATH") {
for dir in std::env::split_paths(&paths) {
let candidate = dir.join("ffmpeg");
if candidate.is_file() {
return Some(candidate);
}
}
}
["/opt/homebrew/bin/ffmpeg", "/usr/local/bin/ffmpeg"]
.iter()
.map(PathBuf::from)
.find(|p| p.is_file())
}
fn avfoundation_index(ffmpeg: &PathBuf, device_name: &str) -> Result<String, String> {
let out = Command::new(ffmpeg)
.args([
"-hide_banner",
"-f",
"avfoundation",
"-list_devices",
"true",
"-i",
"",
])
.output()
.map_err(|e| format!("could not run ffmpeg: {e}"))?;
let listing = String::from_utf8_lossy(&out.stderr);
parse_avfoundation_index(&listing, device_name)
.ok_or_else(|| format!("ffmpeg cannot see the '{device_name}' input device"))
}
fn parse_avfoundation_index(listing: &str, device_name: &str) -> Option<String> {
let needle = device_name.to_ascii_lowercase();
let mut in_audio = false;
for line in listing.lines() {
let lower = line.to_ascii_lowercase();
if lower.contains("audio devices:") {
in_audio = true;
continue;
}
if lower.contains("video devices:") {
in_audio = false;
continue;
}
if !in_audio {
continue;
}
let Some(open) = line.rfind('[') else {
continue;
};
let Some(close) = line[open..].find(']').map(|i| i + open) else {
continue;
};
let index = line[open + 1..close].trim();
if !index.chars().all(|c| c.is_ascii_digit()) || index.is_empty() {
continue;
}
if line[close + 1..]
.trim()
.to_ascii_lowercase()
.contains(&needle)
{
return Some(index.to_string());
}
}
None
}
fn local_ip_towards(host: &str) -> Option<String> {
let socket = UdpSocket::bind("0.0.0.0:0").ok()?;
socket.connect((host, super::soap::PORT)).ok()?;
Some(socket.local_addr().ok()?.ip().to_string())
}
pub fn didl(room: &str) -> String {
let title = super::soap::escape(&format!("mnml — this Mac → {room}"));
format!(
"<DIDL-Lite xmlns:dc=\"http://purl.org/dc/elements/1.1/\" \
xmlns:upnp=\"urn:schemas-upnp-org:metadata-1-0/upnp/\" \
xmlns=\"urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/\">\
<item id=\"-1\" parentID=\"-1\" restricted=\"true\">\
<dc:title>{title}</dc:title>\
<upnp:class>object.item.audioItem.audioBroadcast</upnp:class>\
</item></DIDL-Lite>"
)
}
#[cfg(test)]
mod tests {
use super::*;
const LISTING: &str = "\
[AVFoundation indev @ 0x14f704080] AVFoundation video devices:
[AVFoundation indev @ 0x14f704080] [0] FaceTime HD Camera
[AVFoundation indev @ 0x14f704080] [1] Capture screen 0
[AVFoundation indev @ 0x14f704080] AVFoundation audio devices:
[AVFoundation indev @ 0x14f704080] [0] MacBook Pro Microphone
[AVFoundation indev @ 0x14f704080] [1] BlackHole 2ch
";
#[test]
fn finds_the_loopback_input_index() {
assert_eq!(
parse_avfoundation_index(LISTING, "BlackHole 2ch").as_deref(),
Some("1")
);
}
#[test]
fn ignores_the_video_section_when_indexes_collide() {
assert_eq!(
parse_avfoundation_index(LISTING, "MacBook Pro Microphone").as_deref(),
Some("0")
);
assert!(parse_avfoundation_index(LISTING, "FaceTime HD Camera").is_none());
}
#[test]
fn missing_device_is_none_not_a_guess() {
assert!(parse_avfoundation_index(LISTING, "Loopback Audio").is_none());
assert!(parse_avfoundation_index("", "BlackHole 2ch").is_none());
}
#[test]
fn encoder_args_capture_audio_only_and_stream_to_stdout() {
let args = encoder_args("1");
assert!(args.contains(&":1".to_string()), "audio-only input spec");
assert!(args.contains(&"pipe:1".to_string()));
assert_eq!(args.last().unwrap(), "pipe:1");
assert!(args.windows(2).any(|w| w[0] == "-f" && w[1] == "mp3"));
}
#[test]
fn didl_escapes_the_room_name() {
let d = didl("Kids' <Room>");
assert!(d.contains("Kids' <Room>"));
assert!(d.contains("audioBroadcast"));
}
#[test]
fn install_hint_names_the_actual_formula() {
assert!(INSTALL_HINT.contains("blackhole-2ch"));
}
fn check_with_real_socket(allowed: Option<std::net::IpAddr>) -> bool {
let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
let addr = listener.local_addr().unwrap();
let client = std::thread::spawn(move || TcpStream::connect(addr).unwrap());
let (server_side, _) = listener.accept().unwrap();
let _c = client.join().unwrap();
peer_allowed(&server_side, allowed)
}
#[test]
fn the_expected_player_is_allowed() {
assert!(check_with_real_socket(Some("127.0.0.1".parse().unwrap())));
}
#[test]
fn any_other_lan_host_is_refused() {
assert!(!check_with_real_socket(Some(
"192.168.1.131".parse().unwrap()
)));
}
#[test]
fn an_unparseable_player_host_fails_closed() {
assert!(!check_with_real_socket(None));
}
}