tsiry-sandratraina.com / zerod
Headless Rust gRPC daemon to drive Bluetooth, HLS/DASH playback, and snapcast/shairport-sync/squeezelite on Raspberry Pi audio rigs.
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

Add Spotify Connect via librespot subprocess source

`crates/stream/src/sources/librespot.rs` (Linux) spawns `librespot
--backend pipe --format S16 --device -` so the child writes raw S16LE
PCM at 44.1 kHz / 2ch to stdout. A tokio reader converts byte pairs to
i16 via `from_le_bytes` (endian-safe across cross-compile targets),
applies the existing per-stream gain, and pushes into the same
AudioSink as HLS/DASH. `kill_on_drop` on the Child reaps librespot
when `Player::cancel()` aborts the driver task. Stderr is mirrored to
tracing at debug.

Player gets a new `source: AtomicU8` (PlaybackSource: Hls/Dash/Spotify
/Unspecified) and shared `install(player, task)` helper so HLS/DASH
and Spotify use the same singleton-swap path. The HLS/DASH branch now
sets source from `ManifestKind` instead of inferring at status time.

Settings: new `[librespot]` block (enabled / binary / name / bitrate /
cache_path); when `enabled = false`, `StreamService.SpotifyStart`
returns FAILED_PRECONDITION. Non-Linux builds get a stub that returns
"librespot: linux only" so cross-platform dev compiles.

CLI: `zerod stream spotify start [--output ...]` / `stream spotify
stop`. `stream status` now prints `source=Spotify` etc.

author
Tsiry Sandratraina date (Jun 11, 2026, 10:35 PM +0300) commit 434b5c82 parent 757141a0
+692 -46
+25
crates/proto/proto/zerod/v1alpha1/stream.proto
··· 17 17 PLAYER_STATE_ERRORED = 5; 18 18 } 19 19 20 + enum PlaybackSource { 21 + PLAYBACK_SOURCE_UNSPECIFIED = 0; 22 + PLAYBACK_SOURCE_HLS = 1; 23 + PLAYBACK_SOURCE_DASH = 2; 24 + PLAYBACK_SOURCE_SPOTIFY = 3; 25 + } 26 + 20 27 message PlayRequest { 21 28 // HLS (.m3u8) or DASH (.mpd) URL. 22 29 string url = 1; ··· 47 54 optional string error = 6; 48 55 AudioOutput output = 7; 49 56 uint32 volume_percent = 8; // per-stream gain, 0..=100 57 + PlaybackSource source = 9; 50 58 } 51 59 60 + message SpotifyStartRequest { 61 + AudioOutput output = 1; 62 + // Required when output == AUDIO_OUTPUT_PIPE. 63 + optional string pipe_path = 2; 64 + optional string cpal_device = 3; 65 + } 66 + message SpotifyStartResponse {} 67 + 68 + message SpotifyStopRequest {} 69 + message SpotifyStopResponse {} 70 + 52 71 message SetStreamVolumeRequest { 53 72 uint32 volume_percent = 1; // 0..=100 54 73 } ··· 69 88 // Independent of the system ALSA mixer (see VolumeService). 70 89 rpc SetStreamVolume(SetStreamVolumeRequest) returns (SetStreamVolumeResponse); 71 90 rpc GetStreamVolume(GetStreamVolumeRequest) returns (GetStreamVolumeResponse); 91 + // Start a Spotify Connect session via librespot. The daemon advertises 92 + // itself on the LAN; the phone picks it from the Devices list. Returns 93 + // FAILED_PRECONDITION when `[librespot].enabled = false` or the binary 94 + // is missing. 95 + rpc SpotifyStart(SpotifyStartRequest) returns (SpotifyStartResponse); 96 + rpc SpotifyStop(SpotifyStopRequest) returns (SpotifyStopResponse); 72 97 }
+210
crates/proto/src/generated/zerod.v1alpha1.rs
··· 774 774 /// per-stream gain, 0..=100 775 775 #[prost(uint32, tag = "8")] 776 776 pub volume_percent: u32, 777 + #[prost(enumeration = "PlaybackSource", tag = "9")] 778 + pub source: i32, 777 779 } 780 + #[derive(Clone, PartialEq, ::prost::Message)] 781 + pub struct SpotifyStartRequest { 782 + #[prost(enumeration = "AudioOutput", tag = "1")] 783 + pub output: i32, 784 + /// Required when output == AUDIO_OUTPUT_PIPE. 785 + #[prost(string, optional, tag = "2")] 786 + pub pipe_path: ::core::option::Option<::prost::alloc::string::String>, 787 + #[prost(string, optional, tag = "3")] 788 + pub cpal_device: ::core::option::Option<::prost::alloc::string::String>, 789 + } 790 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 791 + pub struct SpotifyStartResponse {} 792 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 793 + pub struct SpotifyStopRequest {} 794 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 795 + pub struct SpotifyStopResponse {} 778 796 #[derive(Clone, Copy, PartialEq, ::prost::Message)] 779 797 pub struct SetStreamVolumeRequest { 780 798 /// 0..=100 ··· 859 877 "PLAYER_STATE_PLAYING" => Some(Self::Playing), 860 878 "PLAYER_STATE_PAUSED" => Some(Self::Paused), 861 879 "PLAYER_STATE_ERRORED" => Some(Self::Errored), 880 + _ => None, 881 + } 882 + } 883 + } 884 + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] 885 + #[repr(i32)] 886 + pub enum PlaybackSource { 887 + Unspecified = 0, 888 + Hls = 1, 889 + Dash = 2, 890 + Spotify = 3, 891 + } 892 + impl PlaybackSource { 893 + /// String value of the enum field names used in the ProtoBuf definition. 894 + /// 895 + /// The values are not transformed in any way and thus are considered stable 896 + /// (if the ProtoBuf definition does not change) and safe for programmatic use. 897 + pub fn as_str_name(&self) -> &'static str { 898 + match self { 899 + Self::Unspecified => "PLAYBACK_SOURCE_UNSPECIFIED", 900 + Self::Hls => "PLAYBACK_SOURCE_HLS", 901 + Self::Dash => "PLAYBACK_SOURCE_DASH", 902 + Self::Spotify => "PLAYBACK_SOURCE_SPOTIFY", 903 + } 904 + } 905 + /// Creates an enum from field names used in the ProtoBuf definition. 906 + pub fn from_str_name(value: &str) -> ::core::option::Option<Self> { 907 + match value { 908 + "PLAYBACK_SOURCE_UNSPECIFIED" => Some(Self::Unspecified), 909 + "PLAYBACK_SOURCE_HLS" => Some(Self::Hls), 910 + "PLAYBACK_SOURCE_DASH" => Some(Self::Dash), 911 + "PLAYBACK_SOURCE_SPOTIFY" => Some(Self::Spotify), 862 912 _ => None, 863 913 } 864 914 } ··· 1113 1163 ); 1114 1164 self.inner.unary(req, path, codec).await 1115 1165 } 1166 + /// Start a Spotify Connect session via librespot. The daemon advertises 1167 + /// itself on the LAN; the phone picks it from the Devices list. Returns 1168 + /// FAILED_PRECONDITION when `[librespot].enabled = false` or the binary 1169 + /// is missing. 1170 + pub async fn spotify_start( 1171 + &mut self, 1172 + request: impl tonic::IntoRequest<super::SpotifyStartRequest>, 1173 + ) -> std::result::Result< 1174 + tonic::Response<super::SpotifyStartResponse>, 1175 + tonic::Status, 1176 + > { 1177 + self.inner 1178 + .ready() 1179 + .await 1180 + .map_err(|e| { 1181 + tonic::Status::unknown( 1182 + format!("Service was not ready: {}", e.into()), 1183 + ) 1184 + })?; 1185 + let codec = tonic::codec::ProstCodec::default(); 1186 + let path = http::uri::PathAndQuery::from_static( 1187 + "/zerod.v1alpha1.StreamService/SpotifyStart", 1188 + ); 1189 + let mut req = request.into_request(); 1190 + req.extensions_mut() 1191 + .insert(GrpcMethod::new("zerod.v1alpha1.StreamService", "SpotifyStart")); 1192 + self.inner.unary(req, path, codec).await 1193 + } 1194 + pub async fn spotify_stop( 1195 + &mut self, 1196 + request: impl tonic::IntoRequest<super::SpotifyStopRequest>, 1197 + ) -> std::result::Result< 1198 + tonic::Response<super::SpotifyStopResponse>, 1199 + tonic::Status, 1200 + > { 1201 + self.inner 1202 + .ready() 1203 + .await 1204 + .map_err(|e| { 1205 + tonic::Status::unknown( 1206 + format!("Service was not ready: {}", e.into()), 1207 + ) 1208 + })?; 1209 + let codec = tonic::codec::ProstCodec::default(); 1210 + let path = http::uri::PathAndQuery::from_static( 1211 + "/zerod.v1alpha1.StreamService/SpotifyStop", 1212 + ); 1213 + let mut req = request.into_request(); 1214 + req.extensions_mut() 1215 + .insert(GrpcMethod::new("zerod.v1alpha1.StreamService", "SpotifyStop")); 1216 + self.inner.unary(req, path, codec).await 1217 + } 1116 1218 } 1117 1219 } 1118 1220 /// Generated server implementations. ··· 1164 1266 tonic::Response<super::GetStreamVolumeResponse>, 1165 1267 tonic::Status, 1166 1268 >; 1269 + /// Start a Spotify Connect session via librespot. The daemon advertises 1270 + /// itself on the LAN; the phone picks it from the Devices list. Returns 1271 + /// FAILED_PRECONDITION when `[librespot].enabled = false` or the binary 1272 + /// is missing. 1273 + async fn spotify_start( 1274 + &self, 1275 + request: tonic::Request<super::SpotifyStartRequest>, 1276 + ) -> std::result::Result< 1277 + tonic::Response<super::SpotifyStartResponse>, 1278 + tonic::Status, 1279 + >; 1280 + async fn spotify_stop( 1281 + &self, 1282 + request: tonic::Request<super::SpotifyStopRequest>, 1283 + ) -> std::result::Result< 1284 + tonic::Response<super::SpotifyStopResponse>, 1285 + tonic::Status, 1286 + >; 1167 1287 } 1168 1288 #[derive(Debug)] 1169 1289 pub struct StreamServiceServer<T> { ··· 1540 1660 let inner = self.inner.clone(); 1541 1661 let fut = async move { 1542 1662 let method = GetStreamVolumeSvc(inner); 1663 + let codec = tonic::codec::ProstCodec::default(); 1664 + let mut grpc = tonic::server::Grpc::new(codec) 1665 + .apply_compression_config( 1666 + accept_compression_encodings, 1667 + send_compression_encodings, 1668 + ) 1669 + .apply_max_message_size_config( 1670 + max_decoding_message_size, 1671 + max_encoding_message_size, 1672 + ); 1673 + let res = grpc.unary(method, req).await; 1674 + Ok(res) 1675 + }; 1676 + Box::pin(fut) 1677 + } 1678 + "/zerod.v1alpha1.StreamService/SpotifyStart" => { 1679 + #[allow(non_camel_case_types)] 1680 + struct SpotifyStartSvc<T: StreamService>(pub Arc<T>); 1681 + impl< 1682 + T: StreamService, 1683 + > tonic::server::UnaryService<super::SpotifyStartRequest> 1684 + for SpotifyStartSvc<T> { 1685 + type Response = super::SpotifyStartResponse; 1686 + type Future = BoxFuture< 1687 + tonic::Response<Self::Response>, 1688 + tonic::Status, 1689 + >; 1690 + fn call( 1691 + &mut self, 1692 + request: tonic::Request<super::SpotifyStartRequest>, 1693 + ) -> Self::Future { 1694 + let inner = Arc::clone(&self.0); 1695 + let fut = async move { 1696 + <T as StreamService>::spotify_start(&inner, request).await 1697 + }; 1698 + Box::pin(fut) 1699 + } 1700 + } 1701 + let accept_compression_encodings = self.accept_compression_encodings; 1702 + let send_compression_encodings = self.send_compression_encodings; 1703 + let max_decoding_message_size = self.max_decoding_message_size; 1704 + let max_encoding_message_size = self.max_encoding_message_size; 1705 + let inner = self.inner.clone(); 1706 + let fut = async move { 1707 + let method = SpotifyStartSvc(inner); 1708 + let codec = tonic::codec::ProstCodec::default(); 1709 + let mut grpc = tonic::server::Grpc::new(codec) 1710 + .apply_compression_config( 1711 + accept_compression_encodings, 1712 + send_compression_encodings, 1713 + ) 1714 + .apply_max_message_size_config( 1715 + max_decoding_message_size, 1716 + max_encoding_message_size, 1717 + ); 1718 + let res = grpc.unary(method, req).await; 1719 + Ok(res) 1720 + }; 1721 + Box::pin(fut) 1722 + } 1723 + "/zerod.v1alpha1.StreamService/SpotifyStop" => { 1724 + #[allow(non_camel_case_types)] 1725 + struct SpotifyStopSvc<T: StreamService>(pub Arc<T>); 1726 + impl< 1727 + T: StreamService, 1728 + > tonic::server::UnaryService<super::SpotifyStopRequest> 1729 + for SpotifyStopSvc<T> { 1730 + type Response = super::SpotifyStopResponse; 1731 + type Future = BoxFuture< 1732 + tonic::Response<Self::Response>, 1733 + tonic::Status, 1734 + >; 1735 + fn call( 1736 + &mut self, 1737 + request: tonic::Request<super::SpotifyStopRequest>, 1738 + ) -> Self::Future { 1739 + let inner = Arc::clone(&self.0); 1740 + let fut = async move { 1741 + <T as StreamService>::spotify_stop(&inner, request).await 1742 + }; 1743 + Box::pin(fut) 1744 + } 1745 + } 1746 + let accept_compression_encodings = self.accept_compression_encodings; 1747 + let send_compression_encodings = self.send_compression_encodings; 1748 + let max_decoding_message_size = self.max_decoding_message_size; 1749 + let max_encoding_message_size = self.max_encoding_message_size; 1750 + let inner = self.inner.clone(); 1751 + let fut = async move { 1752 + let method = SpotifyStopSvc(inner); 1543 1753 let codec = tonic::codec::ProstCodec::default(); 1544 1754 let mut grpc = tonic::server::Grpc::new(codec) 1545 1755 .apply_compression_config(
crates/proto/src/generated/zerod_descriptor.bin

This is a binary file and will not be displayed.

+1 -1
crates/server/src/lib.rs
··· 68 68 interceptor.clone(), 69 69 )) 70 70 .add_service(StreamServiceServer::with_interceptor( 71 - stream::StreamSvc::default(), 71 + stream::StreamSvc::new(settings.librespot.clone()), 72 72 interceptor.clone(), 73 73 )) 74 74 .add_service(SystemdServiceServer::with_interceptor(
+51
crates/server/src/settings.rs
··· 31 31 pub configs: Vec<ManagedConfig>, 32 32 #[serde(default)] 33 33 pub snapcast: SnapcastSettings, 34 + #[serde(default)] 35 + pub librespot: LibrespotSettings, 36 + } 37 + 38 + #[derive(Debug, Clone, Deserialize, Serialize)] 39 + pub struct LibrespotSettings { 40 + /// Allow `StreamService.SpotifyStart` to spawn librespot. When false, 41 + /// the RPC returns `FAILED_PRECONDITION`. 42 + #[serde(default)] 43 + pub enabled: bool, 44 + /// `librespot` binary path or name. Resolved against $PATH when not 45 + /// absolute. 46 + #[serde(default = "default_librespot_binary")] 47 + pub binary: String, 48 + /// Spotify Connect device name as it appears in the phone's Devices 49 + /// list. Empty → "zerod". 50 + #[serde(default = "default_librespot_name")] 51 + pub name: String, 52 + /// 96 / 160 / 320 kbps. Defaults to 320. 53 + #[serde(default = "default_librespot_bitrate")] 54 + pub bitrate: u32, 55 + /// Directory where librespot stores credentials. Empty → librespot's 56 + /// own default. `--disable-audio-cache` is always set so this is 57 + /// credentials only. 58 + #[serde(default)] 59 + pub cache_path: String, 60 + } 61 + 62 + impl Default for LibrespotSettings { 63 + fn default() -> Self { 64 + Self { 65 + enabled: false, 66 + binary: default_librespot_binary(), 67 + name: default_librespot_name(), 68 + bitrate: default_librespot_bitrate(), 69 + cache_path: String::new(), 70 + } 71 + } 72 + } 73 + 74 + fn default_librespot_binary() -> String { 75 + "librespot".to_string() 76 + } 77 + 78 + fn default_librespot_name() -> String { 79 + "zerod".to_string() 80 + } 81 + 82 + fn default_librespot_bitrate() -> u32 { 83 + 320 34 84 } 35 85 36 86 #[derive(Debug, Clone, Deserialize, Serialize)] ··· 127 177 mdns: MdnsSettings::default(), 128 178 configs: Vec::new(), 129 179 snapcast: SnapcastSettings::default(), 180 + librespot: LibrespotSettings::default(), 130 181 } 131 182 } 132 183 }
+67 -12
crates/server/src/stream.rs
··· 2 2 use zerod_proto::v1alpha1::{ 3 3 stream_service_server::StreamService, AudioOutput as ProtoOutput, GetStreamVolumeRequest, 4 4 GetStreamVolumeResponse, PauseRequest, PauseResponse, PlayRequest, PlayResponse, 5 - PlayerState as ProtoState, ResumeRequest, ResumeResponse, SetStreamVolumeRequest, 6 - SetStreamVolumeResponse, StatusRequest, StatusResponse, StopRequest, StopResponse, 5 + PlaybackSource as ProtoSource, PlayerState as ProtoState, ResumeRequest, ResumeResponse, 6 + SetStreamVolumeRequest, SetStreamVolumeResponse, SpotifyStartRequest, SpotifyStartResponse, 7 + SpotifyStopRequest, SpotifyStopResponse, StatusRequest, StatusResponse, StopRequest, 8 + StopResponse, 7 9 }; 8 - use zerod_stream::{AudioOutput, PlayConfig, PlayerState}; 10 + use zerod_stream::{AudioOutput, LibrespotConfig, PlayConfig, PlaybackSource, PlayerState}; 9 11 10 - #[derive(Default)] 11 - pub struct StreamSvc; 12 + use crate::settings::LibrespotSettings; 12 13 13 - fn map_output(req: &PlayRequest) -> Result<AudioOutput, Status> { 14 - match ProtoOutput::try_from(req.output) { 14 + pub struct StreamSvc { 15 + librespot: LibrespotSettings, 16 + } 17 + 18 + impl StreamSvc { 19 + pub fn new(librespot: LibrespotSettings) -> Self { 20 + Self { librespot } 21 + } 22 + } 23 + 24 + fn map_output( 25 + output: i32, 26 + pipe_path: Option<String>, 27 + cpal_device: Option<String>, 28 + ) -> Result<AudioOutput, Status> { 29 + match ProtoOutput::try_from(output) { 15 30 Ok(ProtoOutput::Cpal) | Ok(ProtoOutput::Unspecified) => Ok(AudioOutput::Cpal { 16 - device: req.cpal_device.clone().filter(|s| !s.is_empty()), 31 + device: cpal_device.filter(|s| !s.is_empty()), 17 32 }), 18 33 Ok(ProtoOutput::Stdout) => Ok(AudioOutput::Stdout), 19 34 Ok(ProtoOutput::Pipe) => { 20 - let path = req 21 - .pipe_path 22 - .clone() 35 + let path = pipe_path 23 36 .filter(|s| !s.is_empty()) 24 37 .ok_or_else(|| Status::invalid_argument("pipe_path required for AUDIO_OUTPUT_PIPE"))?; 25 38 Ok(AudioOutput::Pipe { path }) ··· 46 59 } 47 60 } 48 61 62 + fn map_source(s: PlaybackSource) -> ProtoSource { 63 + match s { 64 + PlaybackSource::Unspecified => ProtoSource::Unspecified, 65 + PlaybackSource::Hls => ProtoSource::Hls, 66 + PlaybackSource::Dash => ProtoSource::Dash, 67 + PlaybackSource::Spotify => ProtoSource::Spotify, 68 + } 69 + } 70 + 49 71 #[tonic::async_trait] 50 72 impl StreamService for StreamSvc { 51 73 async fn play(&self, req: Request<PlayRequest>) -> Result<Response<PlayResponse>, Status> { 52 74 let req = req.into_inner(); 53 - let output = map_output(&req)?; 75 + let output = map_output(req.output, req.pipe_path, req.cpal_device)?; 54 76 tracing::info!("stream.Play url={} output={:?}", req.url, output); 55 77 zerod_stream::play(PlayConfig { 56 78 url: req.url, ··· 95 117 error: s.error, 96 118 output: map_output_back(&s.output) as i32, 97 119 volume_percent: s.volume_percent, 120 + source: map_source(s.source) as i32, 98 121 })) 99 122 } 100 123 ··· 114 137 Ok(Response::new(GetStreamVolumeResponse { 115 138 volume_percent: zerod_stream::volume(), 116 139 })) 140 + } 141 + 142 + async fn spotify_start( 143 + &self, 144 + req: Request<SpotifyStartRequest>, 145 + ) -> Result<Response<SpotifyStartResponse>, Status> { 146 + if !self.librespot.enabled { 147 + return Err(Status::failed_precondition( 148 + "librespot disabled in zerod.toml ([librespot].enabled = false)", 149 + )); 150 + } 151 + let req = req.into_inner(); 152 + let output = map_output(req.output, req.pipe_path, req.cpal_device)?; 153 + let cfg = LibrespotConfig { 154 + binary: self.librespot.binary.clone(), 155 + name: self.librespot.name.clone(), 156 + bitrate: self.librespot.bitrate, 157 + cache_path: self.librespot.cache_path.clone(), 158 + output, 159 + }; 160 + tracing::info!("stream.SpotifyStart name={}", cfg.name); 161 + zerod_stream::spotify_start(cfg).map_err(|e| Status::internal(format!("{e:#}")))?; 162 + Ok(Response::new(SpotifyStartResponse {})) 163 + } 164 + 165 + async fn spotify_stop( 166 + &self, 167 + _req: Request<SpotifyStopRequest>, 168 + ) -> Result<Response<SpotifyStopResponse>, Status> { 169 + tracing::info!("stream.SpotifyStop"); 170 + zerod_stream::spotify_stop(); 171 + Ok(Response::new(SpotifyStopResponse {})) 117 172 } 118 173 }
+4 -1
crates/stream/src/lib.rs
··· 20 20 mod output; 21 21 mod player; 22 22 mod sink; 23 + mod sources; 23 24 24 25 pub use manifest::{is_hls_or_dash_url, ManifestKind}; 25 26 pub use player::{ 26 - pause, play, resume, set_volume, status, stop, volume, PlayConfig, PlayerState, Status, 27 + pause, play, resume, set_volume, status, stop, volume, PlayConfig, PlaybackSource, 28 + PlayerState, Status, 27 29 }; 28 30 pub use sink::{AudioOutput, AudioSink}; 31 + pub use sources::{spotify_start, spotify_stop, LibrespotConfig};
+95 -31
crates/stream/src/player.rs
··· 42 42 Errored = 4, 43 43 } 44 44 45 + #[repr(u8)] 46 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 47 + pub enum PlaybackSource { 48 + Unspecified = 0, 49 + Hls = 1, 50 + Dash = 2, 51 + Spotify = 3, 52 + } 53 + 54 + impl PlaybackSource { 55 + fn from_u8(v: u8) -> Self { 56 + match v { 57 + 1 => Self::Hls, 58 + 2 => Self::Dash, 59 + 3 => Self::Spotify, 60 + _ => Self::Unspecified, 61 + } 62 + } 63 + } 64 + 45 65 pub struct PlayConfig { 46 66 pub url: String, 47 67 pub output: AudioOutput, ··· 56 76 pub error: Option<String>, 57 77 pub output: AudioOutput, 58 78 pub volume_percent: u32, 79 + pub source: PlaybackSource, 59 80 } 60 81 61 82 /// Player volume state shared with the runtime. We store volume as an ··· 63 84 /// happens at sample-apply time. 64 85 static GLOBAL_VOLUME: AtomicU32 = AtomicU32::new(100); 65 86 87 + /// Shared tokio runtime for any source spawned via [`install`]. 88 + #[cfg(target_os = "linux")] 89 + pub(crate) fn runtime() -> &'static Runtime { 90 + &RT 91 + } 92 + 93 + #[cfg(target_os = "linux")] 94 + pub(crate) fn global_volume() -> u32 { 95 + GLOBAL_VOLUME.load(Ordering::Relaxed) 96 + } 97 + 98 + #[cfg(target_os = "linux")] 99 + pub(crate) fn apply_gain_pub(samples: &mut [i16], vol: u32) { 100 + apply_gain(samples, vol); 101 + } 102 + 66 103 fn apply_gain(samples: &mut [i16], volume_percent: u32) { 67 104 if volume_percent >= 100 { 68 105 return; ··· 81 118 } 82 119 } 83 120 84 - struct Player { 85 - url: String, 86 - output: AudioOutput, 87 - sink: Arc<dyn AudioSink>, 121 + pub(crate) struct Player { 122 + pub(crate) url: String, 123 + pub(crate) output: AudioOutput, 124 + pub(crate) sink: Arc<dyn AudioSink>, 88 125 state: AtomicU8, 89 126 paused: AtomicBool, 90 - stop_flag: Arc<AtomicBool>, 127 + pub(crate) stop_flag: Arc<AtomicBool>, 91 128 position_ms: AtomicI64, 92 129 duration_ms: AtomicI64, 93 130 is_live: AtomicBool, 94 131 task: Mutex<Option<JoinHandle<()>>>, 95 132 last_error: Mutex<Option<String>>, 133 + source: AtomicU8, 96 134 } 97 135 98 136 impl Player { 99 - fn set_state(&self, s: PlayerState) { 137 + pub(crate) fn new( 138 + url: String, 139 + output: AudioOutput, 140 + sink: Arc<dyn AudioSink>, 141 + source: PlaybackSource, 142 + ) -> Self { 143 + Self { 144 + url, 145 + output, 146 + sink, 147 + state: AtomicU8::new(PlayerState::Stopped as u8), 148 + paused: AtomicBool::new(false), 149 + stop_flag: Arc::new(AtomicBool::new(false)), 150 + position_ms: AtomicI64::new(0), 151 + duration_ms: AtomicI64::new(-1), 152 + is_live: AtomicBool::new(false), 153 + task: Mutex::new(None), 154 + last_error: Mutex::new(None), 155 + source: AtomicU8::new(source as u8), 156 + } 157 + } 158 + 159 + pub(crate) fn set_state(&self, s: PlayerState) { 100 160 self.state.store(s as u8, Ordering::SeqCst); 101 161 let error = self.last_error.lock().unwrap().clone(); 102 162 zerod_events::publish(zerod_events::Event::StreamStateChanged { ··· 116 176 } 117 177 } 118 178 119 - fn record_error(&self, msg: String) { 179 + pub(crate) fn record_error(&self, msg: String) { 120 180 tracing::error!("stream: {msg}"); 121 181 *self.last_error.lock().unwrap() = Some(msg); 122 182 self.set_state(PlayerState::Errored); 123 183 } 124 184 125 - fn cancel(&self) { 185 + pub(crate) fn cancel(&self) { 126 186 self.stop_flag.store(true, Ordering::SeqCst); 127 187 self.sink.close(); 128 188 if let Some(h) = self.task.lock().unwrap().take() { 129 189 h.abort(); 130 190 } 131 191 } 192 + } 193 + 194 + /// Install `player` as the current singleton, attaching its driver task. 195 + /// Any previously-installed player is cancelled. Used by both the HLS/DASH 196 + /// path and `sources::librespot`. 197 + pub(crate) fn install(player: Arc<Player>, task: JoinHandle<()>) { 198 + *player.task.lock().unwrap() = Some(task); 199 + let mut g = PLAYER.lock().unwrap(); 200 + if let Some(old) = g.take() { 201 + old.cancel(); 202 + } 203 + *g = Some(player); 204 + } 205 + 206 + #[cfg(target_os = "linux")] 207 + pub(crate) fn build_sink_pub(out: &AudioOutput) -> Result<Arc<dyn AudioSink>> { 208 + build_sink(out) 132 209 } 133 210 134 211 fn build_sink(out: &AudioOutput) -> Result<Arc<dyn AudioSink>> { ··· 377 454 } 378 455 379 456 pub fn play(cfg: PlayConfig) -> Result<()> { 380 - if manifest::is_hls_or_dash_url(&cfg.url).is_none() { 381 - return Err(anyhow!("not an HLS or DASH URL: {}", cfg.url)); 382 - } 457 + let kind = manifest::is_hls_or_dash_url(&cfg.url) 458 + .ok_or_else(|| anyhow!("not an HLS or DASH URL: {}", cfg.url))?; 459 + let source = match kind { 460 + ManifestKind::Hls => PlaybackSource::Hls, 461 + ManifestKind::Dash => PlaybackSource::Dash, 462 + }; 383 463 let sink = build_sink(&cfg.output)?; 384 - let player = Arc::new(Player { 385 - url: cfg.url, 386 - output: cfg.output, 387 - sink, 388 - state: AtomicU8::new(PlayerState::Stopped as u8), 389 - paused: AtomicBool::new(false), 390 - stop_flag: Arc::new(AtomicBool::new(false)), 391 - position_ms: AtomicI64::new(0), 392 - duration_ms: AtomicI64::new(-1), 393 - is_live: AtomicBool::new(false), 394 - task: Mutex::new(None), 395 - last_error: Mutex::new(None), 396 - }); 464 + let player = Arc::new(Player::new(cfg.url, cfg.output, sink, source)); 397 465 let runner = player.clone(); 398 466 let task = RT.spawn(async move { run_player(runner).await }); 399 - *player.task.lock().unwrap() = Some(task); 400 - 401 - let mut g = PLAYER.lock().unwrap(); 402 - if let Some(old) = g.take() { 403 - old.cancel(); 404 - } 405 - *g = Some(player); 467 + install(player, task); 406 468 Ok(()) 407 469 } 408 470 ··· 449 511 error: p.last_error.lock().unwrap().clone(), 450 512 output: p.output.clone(), 451 513 volume_percent, 514 + source: PlaybackSource::from_u8(p.source.load(Ordering::SeqCst)), 452 515 } 453 516 } else { 454 517 Status { ··· 460 523 error: None, 461 524 output: AudioOutput::Cpal { device: None }, 462 525 volume_percent, 526 + source: PlaybackSource::Unspecified, 463 527 } 464 528 } 465 529 }
+130
crates/stream/src/sources/librespot.rs
··· 1 + //! Librespot subprocess source. 2 + //! 3 + //! Spawns `librespot --backend pipe --format S16 --device -` so the child 4 + //! writes raw interleaved S16LE PCM at 44.1 kHz / 2ch to stdout. A tokio 5 + //! task reads in 8 KiB chunks, converts byte pairs to `i16` via 6 + //! `from_le_bytes` (endian-safe across cross-compile targets), applies the 7 + //! existing per-stream gain, and forwards into the same `AudioSink` as 8 + //! HLS/DASH playback. Stderr is mirrored at TRACE so unsolicited 9 + //! librespot chatter doesn't dominate logs. 10 + //! 11 + //! The child is held via `tokio::process::Command::kill_on_drop(true)` 12 + //! so an `abort()` from `Player::cancel()` reliably reaps it. 13 + 14 + use anyhow::{anyhow, Context, Result}; 15 + use std::process::Stdio; 16 + use std::sync::atomic::Ordering; 17 + use std::sync::Arc; 18 + use tokio::io::AsyncReadExt; 19 + use tokio::process::{Child, Command}; 20 + 21 + use super::LibrespotConfig; 22 + use crate::player::{ 23 + apply_gain_pub, build_sink_pub, global_volume, install, runtime, PlaybackSource, Player, 24 + PlayerState, 25 + }; 26 + use crate::sink::PcmFormat; 27 + 28 + pub fn spotify_start(cfg: LibrespotConfig) -> Result<()> { 29 + let sink = build_sink_pub(&cfg.output)?; 30 + let player = Arc::new(Player::new( 31 + format!("spotify://{}", cfg.name), 32 + cfg.output.clone(), 33 + sink, 34 + PlaybackSource::Spotify, 35 + )); 36 + let runner = player.clone(); 37 + let task = runtime().spawn(async move { run_librespot(runner, cfg).await }); 38 + install(player, task); 39 + Ok(()) 40 + } 41 + 42 + pub fn spotify_stop() -> bool { 43 + crate::player::stop() 44 + } 45 + 46 + async fn run_librespot(player: Arc<Player>, cfg: LibrespotConfig) { 47 + if let Err(e) = run_librespot_inner(&player, cfg).await { 48 + player.record_error(format!("{e:#}")); 49 + } else { 50 + player.set_state(PlayerState::Stopped); 51 + } 52 + player.sink.close(); 53 + } 54 + 55 + async fn run_librespot_inner(player: &Arc<Player>, cfg: LibrespotConfig) -> Result<()> { 56 + player.set_state(PlayerState::Buffering); 57 + // librespot --backend pipe emits 44100 / 2ch S16LE regardless of 58 + // source quality. Fix the sink format up front. 59 + let fmt = PcmFormat { 60 + sample_rate: 44_100, 61 + channels: 2, 62 + }; 63 + player.sink.set_format(fmt)?; 64 + 65 + let mut child = spawn_librespot(&cfg)?; 66 + let mut stdout = child 67 + .stdout 68 + .take() 69 + .ok_or_else(|| anyhow!("librespot: no stdout"))?; 70 + if let Some(stderr) = child.stderr.take() { 71 + runtime().spawn(forward_stderr(stderr)); 72 + } 73 + 74 + player.set_state(PlayerState::Playing); 75 + 76 + let mut buf = vec![0u8; 8192]; 77 + loop { 78 + if player.stop_flag.load(Ordering::SeqCst) { 79 + break; 80 + } 81 + let n = match stdout.read(&mut buf).await { 82 + Ok(0) => break, // EOF — child exited 83 + Ok(n) => n, 84 + Err(e) => return Err(anyhow!("librespot stdout read: {e}")), 85 + }; 86 + let mut samples = Vec::with_capacity(n / 2); 87 + for pair in buf[..n].chunks_exact(2) { 88 + samples.push(i16::from_le_bytes([pair[0], pair[1]])); 89 + } 90 + let vol = global_volume(); 91 + if vol < 100 { 92 + apply_gain_pub(&mut samples, vol); 93 + } 94 + player.sink.write(&samples)?; 95 + } 96 + 97 + let _ = child.kill().await; 98 + Ok(()) 99 + } 100 + 101 + fn spawn_librespot(cfg: &LibrespotConfig) -> Result<Child> { 102 + let mut cmd = Command::new(&cfg.binary); 103 + cmd.kill_on_drop(true); 104 + // TODO verify these flags against the installed librespot version. 105 + // 0.4.x ships `--backend pipe` + `--format S16`; older builds used 106 + // `--backend pipe-stdout`. CI Pi smoke test should catch a regression. 107 + cmd.arg("--name").arg(&cfg.name); 108 + cmd.arg("--bitrate").arg(cfg.bitrate.to_string()); 109 + cmd.arg("--backend").arg("pipe"); 110 + cmd.arg("--device").arg("-"); 111 + cmd.arg("--format").arg("S16"); 112 + cmd.arg("--initial-volume").arg("100"); 113 + if !cfg.cache_path.is_empty() { 114 + cmd.arg("--cache").arg(&cfg.cache_path); 115 + // Cache credentials but not raw audio chunks — disk-cheap on Pi. 116 + cmd.arg("--disable-audio-cache"); 117 + } 118 + cmd.stdout(Stdio::piped()); 119 + cmd.stderr(Stdio::piped()); 120 + cmd.spawn() 121 + .with_context(|| format!("spawn librespot binary `{}`", cfg.binary)) 122 + } 123 + 124 + async fn forward_stderr(stderr: tokio::process::ChildStderr) { 125 + use tokio::io::{AsyncBufReadExt, BufReader}; 126 + let mut lines = BufReader::new(stderr).lines(); 127 + while let Ok(Some(line)) = lines.next_line().await { 128 + tracing::debug!("librespot: {line}"); 129 + } 130 + }
+15
crates/stream/src/sources/librespot_stub.rs
··· 1 + //! Non-Linux placeholder. librespot itself runs on macOS too, but the 2 + //! production target for zerod is Linux audio appliances — the stub 3 + //! keeps non-Linux dev builds compiling without pulling in a working 4 + //! librespot binary requirement. 5 + 6 + use super::LibrespotConfig; 7 + use anyhow::{bail, Result}; 8 + 9 + pub fn spotify_start(_cfg: LibrespotConfig) -> Result<()> { 10 + bail!("librespot: linux only") 11 + } 12 + 13 + pub fn spotify_stop() -> bool { 14 + false 15 + }
+33
crates/stream/src/sources/mod.rs
··· 1 + //! External playback sources. Each source eventually feeds the same 2 + //! [`AudioSink`](crate::sink::AudioSink) trait so the server only has to 3 + //! know about one sink-construction path. 4 + 5 + use crate::sink::AudioOutput; 6 + 7 + /// Configuration for the librespot subprocess source. Constructed by the 8 + /// server from `[librespot]` in zerod.toml plus the per-RPC output choice. 9 + #[derive(Debug, Clone)] 10 + pub struct LibrespotConfig { 11 + /// Path or name of the `librespot` binary. Resolved against `$PATH` 12 + /// when not absolute. 13 + pub binary: String, 14 + /// Spotify Connect device name advertised to phones. 15 + pub name: String, 16 + /// 96 / 160 / 320 (kbps). 17 + pub bitrate: u32, 18 + /// Directory librespot uses for credentials/cache. Empty → librespot 19 + /// default (current directory). Disables on-disk audio caching. 20 + pub cache_path: String, 21 + /// Where decoded PCM goes. 22 + pub output: AudioOutput, 23 + } 24 + 25 + #[cfg(target_os = "linux")] 26 + mod librespot; 27 + #[cfg(target_os = "linux")] 28 + pub use librespot::{spotify_start, spotify_stop}; 29 + 30 + #[cfg(not(target_os = "linux"))] 31 + mod librespot_stub; 32 + #[cfg(not(target_os = "linux"))] 33 + pub use librespot_stub::{spotify_start, spotify_stop};
+45 -1
src/main.rs
··· 204 204 /// Per-stream software gain (0..=100), independent of the system mixer. 205 205 #[command(subcommand)] 206 206 Volume(StreamVolumeCmd), 207 + /// Spotify Connect via librespot — daemon advertises itself; the 208 + /// phone pushes audio to it. 209 + #[command(subcommand)] 210 + Spotify(SpotifyCmd), 211 + } 212 + 213 + #[derive(Subcommand)] 214 + enum SpotifyCmd { 215 + /// Begin advertising as a Spotify Connect device. 216 + Start { 217 + #[arg(long, value_enum, default_value_t = OutputArg::Cpal)] 218 + output: OutputArg, 219 + #[arg(long)] 220 + pipe_path: Option<String>, 221 + }, 222 + /// Tear down the librespot subprocess. 223 + Stop, 207 224 } 208 225 209 226 #[derive(Subcommand)] ··· 580 597 .into_inner(); 581 598 let state = pb::PlayerState::try_from(r.state).unwrap_or(pb::PlayerState::Unspecified); 582 599 let out = pb::AudioOutput::try_from(r.output).unwrap_or(pb::AudioOutput::Unspecified); 600 + let src = pb::PlaybackSource::try_from(r.source).unwrap_or(pb::PlaybackSource::Unspecified); 583 601 println!( 584 - "state={state:?} url={} position_ms={} duration_ms={} is_live={} output={out:?} volume={}% error={:?}", 602 + "state={state:?} source={src:?} url={} position_ms={} duration_ms={} is_live={} output={out:?} volume={}% error={:?}", 585 603 r.url, r.position_ms, r.duration_ms, r.is_live, r.volume_percent, r.error, 586 604 ); 587 605 } ··· 599 617 Request::new(pb::SetStreamVolumeRequest { volume_percent: percent }), 600 618 &token, 601 619 )) 620 + .await?; 621 + println!("ok"); 622 + } 623 + }, 624 + StreamCmd::Spotify(cmd) => match cmd { 625 + SpotifyCmd::Start { output, pipe_path } => { 626 + let output = match output { 627 + OutputArg::Cpal => pb::AudioOutput::Cpal, 628 + OutputArg::Stdout => pb::AudioOutput::Stdout, 629 + OutputArg::Pipe => pb::AudioOutput::Pipe, 630 + } as i32; 631 + client 632 + .spotify_start(attach_token( 633 + Request::new(pb::SpotifyStartRequest { 634 + output, 635 + pipe_path, 636 + cpal_device: None, 637 + }), 638 + &token, 639 + )) 640 + .await?; 641 + println!("ok"); 642 + } 643 + SpotifyCmd::Stop => { 644 + client 645 + .spotify_stop(attach_token(Request::new(pb::SpotifyStopRequest {}), &token)) 602 646 .await?; 603 647 println!("ok"); 604 648 }
+16
zerod.toml.example
··· 63 63 # Forward snapserver push notifications onto the in-process event bus so 64 64 # `zerod events tail --filter snap` catches external `snapctl` changes. 65 65 forward_notifications = true 66 + 67 + [librespot] 68 + # When enabled, StreamService.SpotifyStart spawns a librespot subprocess 69 + # wired into the same audio sink as HLS/DASH. Install librespot via 70 + # `apt install librespot` (Debian/Ubuntu) or `brew install librespot`. 71 + enabled = false 72 + # Resolved against $PATH when not absolute. 73 + binary = "librespot" 74 + # Device name as it appears in the Spotify app's Devices list. 75 + name = "zerod" 76 + # 96 / 160 / 320 kbps. 77 + bitrate = 320 78 + # Where librespot stores credentials. Empty → librespot's default 79 + # (current directory). --disable-audio-cache is always passed, so this 80 + # is creds only — pick a persistent path so you don't re-pair every boot. 81 + cache_path = ""