diff --git a/crates/proto/proto/zerod/v1alpha1/stream.proto b/crates/proto/proto/zerod/v1alpha1/stream.proto index 68a1709..225086a 100644 --- a/crates/proto/proto/zerod/v1alpha1/stream.proto +++ b/crates/proto/proto/zerod/v1alpha1/stream.proto @@ -17,6 +17,13 @@ enum PlayerState { PLAYER_STATE_ERRORED = 5; } +enum PlaybackSource { + PLAYBACK_SOURCE_UNSPECIFIED = 0; + PLAYBACK_SOURCE_HLS = 1; + PLAYBACK_SOURCE_DASH = 2; + PLAYBACK_SOURCE_SPOTIFY = 3; +} + message PlayRequest { // HLS (.m3u8) or DASH (.mpd) URL. string url = 1; @@ -47,7 +54,19 @@ message StatusResponse { optional string error = 6; AudioOutput output = 7; uint32 volume_percent = 8; // per-stream gain, 0..=100 + PlaybackSource source = 9; +} + +message SpotifyStartRequest { + AudioOutput output = 1; + // Required when output == AUDIO_OUTPUT_PIPE. + optional string pipe_path = 2; + optional string cpal_device = 3; } +message SpotifyStartResponse {} + +message SpotifyStopRequest {} +message SpotifyStopResponse {} message SetStreamVolumeRequest { uint32 volume_percent = 1; // 0..=100 @@ -69,4 +88,10 @@ service StreamService { // Independent of the system ALSA mixer (see VolumeService). rpc SetStreamVolume(SetStreamVolumeRequest) returns (SetStreamVolumeResponse); rpc GetStreamVolume(GetStreamVolumeRequest) returns (GetStreamVolumeResponse); + // Start a Spotify Connect session via librespot. The daemon advertises + // itself on the LAN; the phone picks it from the Devices list. Returns + // FAILED_PRECONDITION when `[librespot].enabled = false` or the binary + // is missing. + rpc SpotifyStart(SpotifyStartRequest) returns (SpotifyStartResponse); + rpc SpotifyStop(SpotifyStopRequest) returns (SpotifyStopResponse); } diff --git a/crates/proto/src/generated/zerod.v1alpha1.rs b/crates/proto/src/generated/zerod.v1alpha1.rs index 130f9e6..953248c 100644 --- a/crates/proto/src/generated/zerod.v1alpha1.rs +++ b/crates/proto/src/generated/zerod.v1alpha1.rs @@ -774,7 +774,25 @@ pub struct StatusResponse { /// per-stream gain, 0..=100 #[prost(uint32, tag = "8")] pub volume_percent: u32, + #[prost(enumeration = "PlaybackSource", tag = "9")] + pub source: i32, } +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct SpotifyStartRequest { + #[prost(enumeration = "AudioOutput", tag = "1")] + pub output: i32, + /// Required when output == AUDIO_OUTPUT_PIPE. + #[prost(string, optional, tag = "2")] + pub pipe_path: ::core::option::Option<::prost::alloc::string::String>, + #[prost(string, optional, tag = "3")] + pub cpal_device: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct SpotifyStartResponse {} +#[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct SpotifyStopRequest {} +#[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct SpotifyStopResponse {} #[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct SetStreamVolumeRequest { /// 0..=100 @@ -863,6 +881,38 @@ impl PlayerState { } } } +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] +pub enum PlaybackSource { + Unspecified = 0, + Hls = 1, + Dash = 2, + Spotify = 3, +} +impl PlaybackSource { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Self::Unspecified => "PLAYBACK_SOURCE_UNSPECIFIED", + Self::Hls => "PLAYBACK_SOURCE_HLS", + Self::Dash => "PLAYBACK_SOURCE_DASH", + Self::Spotify => "PLAYBACK_SOURCE_SPOTIFY", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "PLAYBACK_SOURCE_UNSPECIFIED" => Some(Self::Unspecified), + "PLAYBACK_SOURCE_HLS" => Some(Self::Hls), + "PLAYBACK_SOURCE_DASH" => Some(Self::Dash), + "PLAYBACK_SOURCE_SPOTIFY" => Some(Self::Spotify), + _ => None, + } + } +} /// Generated client implementations. pub mod stream_service_client { #![allow( @@ -1113,6 +1163,58 @@ pub mod stream_service_client { ); self.inner.unary(req, path, codec).await } + /// Start a Spotify Connect session via librespot. The daemon advertises + /// itself on the LAN; the phone picks it from the Devices list. Returns + /// FAILED_PRECONDITION when `[librespot].enabled = false` or the binary + /// is missing. + pub async fn spotify_start( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic::codec::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/zerod.v1alpha1.StreamService/SpotifyStart", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("zerod.v1alpha1.StreamService", "SpotifyStart")); + self.inner.unary(req, path, codec).await + } + pub async fn spotify_stop( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic::codec::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/zerod.v1alpha1.StreamService/SpotifyStop", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("zerod.v1alpha1.StreamService", "SpotifyStop")); + self.inner.unary(req, path, codec).await + } } } /// Generated server implementations. @@ -1164,6 +1266,24 @@ pub mod stream_service_server { tonic::Response, tonic::Status, >; + /// Start a Spotify Connect session via librespot. The daemon advertises + /// itself on the LAN; the phone picks it from the Devices list. Returns + /// FAILED_PRECONDITION when `[librespot].enabled = false` or the binary + /// is missing. + async fn spotify_start( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; + async fn spotify_stop( + &self, + request: tonic::Request, + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; } #[derive(Debug)] pub struct StreamServiceServer { @@ -1555,6 +1675,96 @@ pub mod stream_service_server { }; Box::pin(fut) } + "/zerod.v1alpha1.StreamService/SpotifyStart" => { + #[allow(non_camel_case_types)] + struct SpotifyStartSvc(pub Arc); + impl< + T: StreamService, + > tonic::server::UnaryService + for SpotifyStartSvc { + type Response = super::SpotifyStartResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::spotify_start(&inner, request).await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = SpotifyStartSvc(inner); + let codec = tonic::codec::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/zerod.v1alpha1.StreamService/SpotifyStop" => { + #[allow(non_camel_case_types)] + struct SpotifyStopSvc(pub Arc); + impl< + T: StreamService, + > tonic::server::UnaryService + for SpotifyStopSvc { + type Response = super::SpotifyStopResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::spotify_stop(&inner, request).await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = SpotifyStopSvc(inner); + let codec = tonic::codec::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } _ => { Box::pin(async move { let mut response = http::Response::new(empty_body()); diff --git a/crates/proto/src/generated/zerod_descriptor.bin b/crates/proto/src/generated/zerod_descriptor.bin index 83861de4859826b5150f371e32f351fe02c6a3e0..3cca3198d5665d17fb3b53f18f681d0cff00ae91 100644 GIT binary patch delta 2957 zcmaEIgYotg#to~um{03)O*XWb*?f{Kh>`K-W*Ht0#uOnlF1F(Q(xT*42~Gt@4L&0| zy{gor{1m-1!^E6|j6_4dfSkn2q{QUxV6dtnsA?w;F1Gm8qN4mFr45W+!odakC7EfJ z!6k`BB|)i$rK!awLdINd`K2WVr6rReI!R2HHmXauv6(CWF;>~`%cNy!o8X39xP+ch`&m|q;;~44W=>nEB z>>3~H7aZW~?CIv|>LS1(B+ey_uE4`5Sb$MTf=dKl!o@MzLx4$0noAsACOE)9gu~M< zQh?d$1?y%#ep9A;p-3(sxCau8N`#cr0~A>n>OUzJ{E8sn6$<0xhHK6*5K_XTi31W1 z`2~{=V@&VFdAHVLT=wW)THNs6s{|W>F?7s1zfZCko>+2{DUGNJ14d z2{DT?NkgSTmWjnMiNUxaNl7WFN@gKuaV9ya6f+};Ckf**2{DVy$Z~OUF|jZRu}Co} zfgR7nz$nBl#Uae4$Hl_J0CK#{VyP97MKS%_JNUjSqu3j@e4 zvP@tbKpZw9W?3d)s6vp3Wcfs3JZ2$gS#b$2ORz0WLd^1$V@0ejlnLoWYuzg_sp5Ul4JvSA=V3W))&qglh)*TTuuc<}3_M z%pmm&id>vrOe~B-%*qVnpg3azIZv5G1T4%X#Hp<;fXB1-Af!fI-#HjbZvw@g6IOaG({xoEQ1UDxWBdZXzAvD4{ znLt5f2#s(~CProp+oaBqq)!1$HH~ z5VJ9pgc7fUuAxF&evv{>W?3pE>N%M}v1bf71r*fAa8p3BXAE~Vix9I36U>5o7FH0K z4Qe|mDooh2NZEBFn zqY$$-2gKD}pyXi#akU`}7dU;`KwZrRP9HWDr?4TKy3!GN$*g^P)aMTprRZU`vh*uxD0 zS!NH-uw3BWZV%0{+(PxtjtuhP+{4W%#O%l+1I|4>Ld?z#pt6jag@;jy*_lHMEX*s! z?7|=iR?W*O#O%TW2^~Hm7B>bDNa*k}3NgEJ$Z?5rv9R!gLdTs+Mu}6wKu^!s(7*ta zD)^WfScRC~!O>OE3R1=>#Ow}@FFtTUxl2idZR8hX_GC~3+sMx-#O%o-4;B^>V)kZG z1Pco=3Nd?gfE>ldA}GY-!{7&Th#;d7vkwO(tOP+0@rA~^AlP#)y6Eu90GSGDC38~UCGQO#2msc#udQB#mWXL z#Y5oEWD#NxgKLLi}H%O6xJVQMF{S?YG zQu7oNqTz0c)l1DwOv*`3QLt4=OUx-wO;E@$0-KtYnU`2p$)%84tdN^oT%4Jgt_Mp} zpz`X$;al&da9z RkREm)e#N&nWKyrK{ zFdnlIvzWL9mnB#mlMu7S6;vj&G4S3M{SI6*O{$s_^xEGH8Kn-H@m6C?^anLtsi z$tMhxV-{l86cY!>94E-1T1=4O=44`I6=K$cMmQ%EC}^~x5zfiP$SlOH1&we{CMF?f zZK!SaoJ^od(`I6aS^zRrn*(MY6SEMrwwO4V6c-B%Cy1lNB%#Esplhg*mS3chlUbGu ziF!^ZQ0(czO#ub94%`$_?CHQA%_79C%LKEao`n^}WrNxdiV9tJJ{XT#h*?)q2Cd8}{P3xRYY-~c>%=++< z2F0vCG_i9su`vrV>x)Zpse`@FF2rmw`JR+!y#YLgK{0N?3A2WsS%}#{N*Wx|TtY0y z45pBX=3*3LHs+Ay660cF;R40836qQxr-Ff=o~@yQ0mK7bObo0-%qCEixxi6w0yUY7 ziGf*&*+fbjoZz_Yg_zA4l)wp*n^B0_j6)tQ&LhNZ!Jr5h=3x|Kw&0KhJA_w=#gf4q z;t*a&A!bVsd2l%Jf*fK64NP9J?N(6RdBL_@L2c&++inH%7aM~ROT8_FJJ>unP`b3` z5CxgX1}<{!nDjwG4&pEhG25|${RZMO2{GGoi^1fWg_!LWb)niAg_!M`jG$7WBx}#c z4dXEhG28Qi^8_0MI04%$8-Vi*8v`f>J206-Rf3YN0~@$(14%InF+1=|!Q_~Qm>pD& zq1G`AF+0Lt2}+}ma91)j2{AjuT?tCDj&N792r)ZBZ8Ky57eG#M>sXkCn4RF(u`mlU zJEFN`X?N a8(bS3lMu5TTpJs+5VM=P-{$uUIe`EM<8h$? diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 503be08..fd0c1e5 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -68,7 +68,7 @@ pub async fn serve(settings: Settings) -> Result<()> { interceptor.clone(), )) .add_service(StreamServiceServer::with_interceptor( - stream::StreamSvc::default(), + stream::StreamSvc::new(settings.librespot.clone()), interceptor.clone(), )) .add_service(SystemdServiceServer::with_interceptor( diff --git a/crates/server/src/settings.rs b/crates/server/src/settings.rs index 8571aac..f94e8cb 100644 --- a/crates/server/src/settings.rs +++ b/crates/server/src/settings.rs @@ -31,6 +31,56 @@ pub struct Settings { pub configs: Vec, #[serde(default)] pub snapcast: SnapcastSettings, + #[serde(default)] + pub librespot: LibrespotSettings, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct LibrespotSettings { + /// Allow `StreamService.SpotifyStart` to spawn librespot. When false, + /// the RPC returns `FAILED_PRECONDITION`. + #[serde(default)] + pub enabled: bool, + /// `librespot` binary path or name. Resolved against $PATH when not + /// absolute. + #[serde(default = "default_librespot_binary")] + pub binary: String, + /// Spotify Connect device name as it appears in the phone's Devices + /// list. Empty → "zerod". + #[serde(default = "default_librespot_name")] + pub name: String, + /// 96 / 160 / 320 kbps. Defaults to 320. + #[serde(default = "default_librespot_bitrate")] + pub bitrate: u32, + /// Directory where librespot stores credentials. Empty → librespot's + /// own default. `--disable-audio-cache` is always set so this is + /// credentials only. + #[serde(default)] + pub cache_path: String, +} + +impl Default for LibrespotSettings { + fn default() -> Self { + Self { + enabled: false, + binary: default_librespot_binary(), + name: default_librespot_name(), + bitrate: default_librespot_bitrate(), + cache_path: String::new(), + } + } +} + +fn default_librespot_binary() -> String { + "librespot".to_string() +} + +fn default_librespot_name() -> String { + "zerod".to_string() +} + +fn default_librespot_bitrate() -> u32 { + 320 } #[derive(Debug, Clone, Deserialize, Serialize)] @@ -127,6 +177,7 @@ impl Default for Settings { mdns: MdnsSettings::default(), configs: Vec::new(), snapcast: SnapcastSettings::default(), + librespot: LibrespotSettings::default(), } } } diff --git a/crates/server/src/stream.rs b/crates/server/src/stream.rs index ebfe7e1..1324ef3 100644 --- a/crates/server/src/stream.rs +++ b/crates/server/src/stream.rs @@ -2,24 +2,37 @@ use tonic::{Request, Response, Status}; use zerod_proto::v1alpha1::{ stream_service_server::StreamService, AudioOutput as ProtoOutput, GetStreamVolumeRequest, GetStreamVolumeResponse, PauseRequest, PauseResponse, PlayRequest, PlayResponse, - PlayerState as ProtoState, ResumeRequest, ResumeResponse, SetStreamVolumeRequest, - SetStreamVolumeResponse, StatusRequest, StatusResponse, StopRequest, StopResponse, + PlaybackSource as ProtoSource, PlayerState as ProtoState, ResumeRequest, ResumeResponse, + SetStreamVolumeRequest, SetStreamVolumeResponse, SpotifyStartRequest, SpotifyStartResponse, + SpotifyStopRequest, SpotifyStopResponse, StatusRequest, StatusResponse, StopRequest, + StopResponse, }; -use zerod_stream::{AudioOutput, PlayConfig, PlayerState}; +use zerod_stream::{AudioOutput, LibrespotConfig, PlayConfig, PlaybackSource, PlayerState}; -#[derive(Default)] -pub struct StreamSvc; +use crate::settings::LibrespotSettings; -fn map_output(req: &PlayRequest) -> Result { - match ProtoOutput::try_from(req.output) { +pub struct StreamSvc { + librespot: LibrespotSettings, +} + +impl StreamSvc { + pub fn new(librespot: LibrespotSettings) -> Self { + Self { librespot } + } +} + +fn map_output( + output: i32, + pipe_path: Option, + cpal_device: Option, +) -> Result { + match ProtoOutput::try_from(output) { Ok(ProtoOutput::Cpal) | Ok(ProtoOutput::Unspecified) => Ok(AudioOutput::Cpal { - device: req.cpal_device.clone().filter(|s| !s.is_empty()), + device: cpal_device.filter(|s| !s.is_empty()), }), Ok(ProtoOutput::Stdout) => Ok(AudioOutput::Stdout), Ok(ProtoOutput::Pipe) => { - let path = req - .pipe_path - .clone() + let path = pipe_path .filter(|s| !s.is_empty()) .ok_or_else(|| Status::invalid_argument("pipe_path required for AUDIO_OUTPUT_PIPE"))?; Ok(AudioOutput::Pipe { path }) @@ -46,11 +59,20 @@ fn map_state(s: PlayerState) -> ProtoState { } } +fn map_source(s: PlaybackSource) -> ProtoSource { + match s { + PlaybackSource::Unspecified => ProtoSource::Unspecified, + PlaybackSource::Hls => ProtoSource::Hls, + PlaybackSource::Dash => ProtoSource::Dash, + PlaybackSource::Spotify => ProtoSource::Spotify, + } +} + #[tonic::async_trait] impl StreamService for StreamSvc { async fn play(&self, req: Request) -> Result, Status> { let req = req.into_inner(); - let output = map_output(&req)?; + let output = map_output(req.output, req.pipe_path, req.cpal_device)?; tracing::info!("stream.Play url={} output={:?}", req.url, output); zerod_stream::play(PlayConfig { url: req.url, @@ -95,6 +117,7 @@ impl StreamService for StreamSvc { error: s.error, output: map_output_back(&s.output) as i32, volume_percent: s.volume_percent, + source: map_source(s.source) as i32, })) } @@ -115,4 +138,36 @@ impl StreamService for StreamSvc { volume_percent: zerod_stream::volume(), })) } + + async fn spotify_start( + &self, + req: Request, + ) -> Result, Status> { + if !self.librespot.enabled { + return Err(Status::failed_precondition( + "librespot disabled in zerod.toml ([librespot].enabled = false)", + )); + } + let req = req.into_inner(); + let output = map_output(req.output, req.pipe_path, req.cpal_device)?; + let cfg = LibrespotConfig { + binary: self.librespot.binary.clone(), + name: self.librespot.name.clone(), + bitrate: self.librespot.bitrate, + cache_path: self.librespot.cache_path.clone(), + output, + }; + tracing::info!("stream.SpotifyStart name={}", cfg.name); + zerod_stream::spotify_start(cfg).map_err(|e| Status::internal(format!("{e:#}")))?; + Ok(Response::new(SpotifyStartResponse {})) + } + + async fn spotify_stop( + &self, + _req: Request, + ) -> Result, Status> { + tracing::info!("stream.SpotifyStop"); + zerod_stream::spotify_stop(); + Ok(Response::new(SpotifyStopResponse {})) + } } diff --git a/crates/stream/src/lib.rs b/crates/stream/src/lib.rs index 39b177e..f01403f 100644 --- a/crates/stream/src/lib.rs +++ b/crates/stream/src/lib.rs @@ -20,9 +20,12 @@ mod manifest; mod output; mod player; mod sink; +mod sources; pub use manifest::{is_hls_or_dash_url, ManifestKind}; pub use player::{ - pause, play, resume, set_volume, status, stop, volume, PlayConfig, PlayerState, Status, + pause, play, resume, set_volume, status, stop, volume, PlayConfig, PlaybackSource, + PlayerState, Status, }; pub use sink::{AudioOutput, AudioSink}; +pub use sources::{spotify_start, spotify_stop, LibrespotConfig}; diff --git a/crates/stream/src/player.rs b/crates/stream/src/player.rs index f2f3733..3e914d0 100644 --- a/crates/stream/src/player.rs +++ b/crates/stream/src/player.rs @@ -42,6 +42,26 @@ pub enum PlayerState { Errored = 4, } +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PlaybackSource { + Unspecified = 0, + Hls = 1, + Dash = 2, + Spotify = 3, +} + +impl PlaybackSource { + fn from_u8(v: u8) -> Self { + match v { + 1 => Self::Hls, + 2 => Self::Dash, + 3 => Self::Spotify, + _ => Self::Unspecified, + } + } +} + pub struct PlayConfig { pub url: String, pub output: AudioOutput, @@ -56,6 +76,7 @@ pub struct Status { pub error: Option, pub output: AudioOutput, pub volume_percent: u32, + pub source: PlaybackSource, } /// Player volume state shared with the runtime. We store volume as an @@ -63,6 +84,22 @@ pub struct Status { /// happens at sample-apply time. static GLOBAL_VOLUME: AtomicU32 = AtomicU32::new(100); +/// Shared tokio runtime for any source spawned via [`install`]. +#[cfg(target_os = "linux")] +pub(crate) fn runtime() -> &'static Runtime { + &RT +} + +#[cfg(target_os = "linux")] +pub(crate) fn global_volume() -> u32 { + GLOBAL_VOLUME.load(Ordering::Relaxed) +} + +#[cfg(target_os = "linux")] +pub(crate) fn apply_gain_pub(samples: &mut [i16], vol: u32) { + apply_gain(samples, vol); +} + fn apply_gain(samples: &mut [i16], volume_percent: u32) { if volume_percent >= 100 { return; @@ -81,22 +118,45 @@ fn apply_gain(samples: &mut [i16], volume_percent: u32) { } } -struct Player { - url: String, - output: AudioOutput, - sink: Arc, +pub(crate) struct Player { + pub(crate) url: String, + pub(crate) output: AudioOutput, + pub(crate) sink: Arc, state: AtomicU8, paused: AtomicBool, - stop_flag: Arc, + pub(crate) stop_flag: Arc, position_ms: AtomicI64, duration_ms: AtomicI64, is_live: AtomicBool, task: Mutex>>, last_error: Mutex>, + source: AtomicU8, } impl Player { - fn set_state(&self, s: PlayerState) { + pub(crate) fn new( + url: String, + output: AudioOutput, + sink: Arc, + source: PlaybackSource, + ) -> Self { + Self { + url, + output, + sink, + state: AtomicU8::new(PlayerState::Stopped as u8), + paused: AtomicBool::new(false), + stop_flag: Arc::new(AtomicBool::new(false)), + position_ms: AtomicI64::new(0), + duration_ms: AtomicI64::new(-1), + is_live: AtomicBool::new(false), + task: Mutex::new(None), + last_error: Mutex::new(None), + source: AtomicU8::new(source as u8), + } + } + + pub(crate) fn set_state(&self, s: PlayerState) { self.state.store(s as u8, Ordering::SeqCst); let error = self.last_error.lock().unwrap().clone(); zerod_events::publish(zerod_events::Event::StreamStateChanged { @@ -116,13 +176,13 @@ impl Player { } } - fn record_error(&self, msg: String) { + pub(crate) fn record_error(&self, msg: String) { tracing::error!("stream: {msg}"); *self.last_error.lock().unwrap() = Some(msg); self.set_state(PlayerState::Errored); } - fn cancel(&self) { + pub(crate) fn cancel(&self) { self.stop_flag.store(true, Ordering::SeqCst); self.sink.close(); if let Some(h) = self.task.lock().unwrap().take() { @@ -131,6 +191,23 @@ impl Player { } } +/// Install `player` as the current singleton, attaching its driver task. +/// Any previously-installed player is cancelled. Used by both the HLS/DASH +/// path and `sources::librespot`. +pub(crate) fn install(player: Arc, task: JoinHandle<()>) { + *player.task.lock().unwrap() = Some(task); + let mut g = PLAYER.lock().unwrap(); + if let Some(old) = g.take() { + old.cancel(); + } + *g = Some(player); +} + +#[cfg(target_os = "linux")] +pub(crate) fn build_sink_pub(out: &AudioOutput) -> Result> { + build_sink(out) +} + fn build_sink(out: &AudioOutput) -> Result> { match out { AudioOutput::Cpal { device } => { @@ -377,32 +454,17 @@ fn current() -> Option> { } pub fn play(cfg: PlayConfig) -> Result<()> { - if manifest::is_hls_or_dash_url(&cfg.url).is_none() { - return Err(anyhow!("not an HLS or DASH URL: {}", cfg.url)); - } + let kind = manifest::is_hls_or_dash_url(&cfg.url) + .ok_or_else(|| anyhow!("not an HLS or DASH URL: {}", cfg.url))?; + let source = match kind { + ManifestKind::Hls => PlaybackSource::Hls, + ManifestKind::Dash => PlaybackSource::Dash, + }; let sink = build_sink(&cfg.output)?; - let player = Arc::new(Player { - url: cfg.url, - output: cfg.output, - sink, - state: AtomicU8::new(PlayerState::Stopped as u8), - paused: AtomicBool::new(false), - stop_flag: Arc::new(AtomicBool::new(false)), - position_ms: AtomicI64::new(0), - duration_ms: AtomicI64::new(-1), - is_live: AtomicBool::new(false), - task: Mutex::new(None), - last_error: Mutex::new(None), - }); + let player = Arc::new(Player::new(cfg.url, cfg.output, sink, source)); let runner = player.clone(); let task = RT.spawn(async move { run_player(runner).await }); - *player.task.lock().unwrap() = Some(task); - - let mut g = PLAYER.lock().unwrap(); - if let Some(old) = g.take() { - old.cancel(); - } - *g = Some(player); + install(player, task); Ok(()) } @@ -449,6 +511,7 @@ pub fn status() -> Status { error: p.last_error.lock().unwrap().clone(), output: p.output.clone(), volume_percent, + source: PlaybackSource::from_u8(p.source.load(Ordering::SeqCst)), } } else { Status { @@ -460,6 +523,7 @@ pub fn status() -> Status { error: None, output: AudioOutput::Cpal { device: None }, volume_percent, + source: PlaybackSource::Unspecified, } } } diff --git a/crates/stream/src/sources/librespot.rs b/crates/stream/src/sources/librespot.rs new file mode 100644 index 0000000..7b4dbd4 --- /dev/null +++ b/crates/stream/src/sources/librespot.rs @@ -0,0 +1,130 @@ +//! Librespot subprocess source. +//! +//! Spawns `librespot --backend pipe --format S16 --device -` so the child +//! writes raw interleaved S16LE PCM at 44.1 kHz / 2ch to stdout. A tokio +//! task reads in 8 KiB chunks, converts byte pairs to `i16` via +//! `from_le_bytes` (endian-safe across cross-compile targets), applies the +//! existing per-stream gain, and forwards into the same `AudioSink` as +//! HLS/DASH playback. Stderr is mirrored at TRACE so unsolicited +//! librespot chatter doesn't dominate logs. +//! +//! The child is held via `tokio::process::Command::kill_on_drop(true)` +//! so an `abort()` from `Player::cancel()` reliably reaps it. + +use anyhow::{anyhow, Context, Result}; +use std::process::Stdio; +use std::sync::atomic::Ordering; +use std::sync::Arc; +use tokio::io::AsyncReadExt; +use tokio::process::{Child, Command}; + +use super::LibrespotConfig; +use crate::player::{ + apply_gain_pub, build_sink_pub, global_volume, install, runtime, PlaybackSource, Player, + PlayerState, +}; +use crate::sink::PcmFormat; + +pub fn spotify_start(cfg: LibrespotConfig) -> Result<()> { + let sink = build_sink_pub(&cfg.output)?; + let player = Arc::new(Player::new( + format!("spotify://{}", cfg.name), + cfg.output.clone(), + sink, + PlaybackSource::Spotify, + )); + let runner = player.clone(); + let task = runtime().spawn(async move { run_librespot(runner, cfg).await }); + install(player, task); + Ok(()) +} + +pub fn spotify_stop() -> bool { + crate::player::stop() +} + +async fn run_librespot(player: Arc, cfg: LibrespotConfig) { + if let Err(e) = run_librespot_inner(&player, cfg).await { + player.record_error(format!("{e:#}")); + } else { + player.set_state(PlayerState::Stopped); + } + player.sink.close(); +} + +async fn run_librespot_inner(player: &Arc, cfg: LibrespotConfig) -> Result<()> { + player.set_state(PlayerState::Buffering); + // librespot --backend pipe emits 44100 / 2ch S16LE regardless of + // source quality. Fix the sink format up front. + let fmt = PcmFormat { + sample_rate: 44_100, + channels: 2, + }; + player.sink.set_format(fmt)?; + + let mut child = spawn_librespot(&cfg)?; + let mut stdout = child + .stdout + .take() + .ok_or_else(|| anyhow!("librespot: no stdout"))?; + if let Some(stderr) = child.stderr.take() { + runtime().spawn(forward_stderr(stderr)); + } + + player.set_state(PlayerState::Playing); + + let mut buf = vec![0u8; 8192]; + loop { + if player.stop_flag.load(Ordering::SeqCst) { + break; + } + let n = match stdout.read(&mut buf).await { + Ok(0) => break, // EOF — child exited + Ok(n) => n, + Err(e) => return Err(anyhow!("librespot stdout read: {e}")), + }; + let mut samples = Vec::with_capacity(n / 2); + for pair in buf[..n].chunks_exact(2) { + samples.push(i16::from_le_bytes([pair[0], pair[1]])); + } + let vol = global_volume(); + if vol < 100 { + apply_gain_pub(&mut samples, vol); + } + player.sink.write(&samples)?; + } + + let _ = child.kill().await; + Ok(()) +} + +fn spawn_librespot(cfg: &LibrespotConfig) -> Result { + let mut cmd = Command::new(&cfg.binary); + cmd.kill_on_drop(true); + // TODO verify these flags against the installed librespot version. + // 0.4.x ships `--backend pipe` + `--format S16`; older builds used + // `--backend pipe-stdout`. CI Pi smoke test should catch a regression. + cmd.arg("--name").arg(&cfg.name); + cmd.arg("--bitrate").arg(cfg.bitrate.to_string()); + cmd.arg("--backend").arg("pipe"); + cmd.arg("--device").arg("-"); + cmd.arg("--format").arg("S16"); + cmd.arg("--initial-volume").arg("100"); + if !cfg.cache_path.is_empty() { + cmd.arg("--cache").arg(&cfg.cache_path); + // Cache credentials but not raw audio chunks — disk-cheap on Pi. + cmd.arg("--disable-audio-cache"); + } + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + cmd.spawn() + .with_context(|| format!("spawn librespot binary `{}`", cfg.binary)) +} + +async fn forward_stderr(stderr: tokio::process::ChildStderr) { + use tokio::io::{AsyncBufReadExt, BufReader}; + let mut lines = BufReader::new(stderr).lines(); + while let Ok(Some(line)) = lines.next_line().await { + tracing::debug!("librespot: {line}"); + } +} diff --git a/crates/stream/src/sources/librespot_stub.rs b/crates/stream/src/sources/librespot_stub.rs new file mode 100644 index 0000000..4233549 --- /dev/null +++ b/crates/stream/src/sources/librespot_stub.rs @@ -0,0 +1,15 @@ +//! Non-Linux placeholder. librespot itself runs on macOS too, but the +//! production target for zerod is Linux audio appliances — the stub +//! keeps non-Linux dev builds compiling without pulling in a working +//! librespot binary requirement. + +use super::LibrespotConfig; +use anyhow::{bail, Result}; + +pub fn spotify_start(_cfg: LibrespotConfig) -> Result<()> { + bail!("librespot: linux only") +} + +pub fn spotify_stop() -> bool { + false +} diff --git a/crates/stream/src/sources/mod.rs b/crates/stream/src/sources/mod.rs new file mode 100644 index 0000000..454e8c7 --- /dev/null +++ b/crates/stream/src/sources/mod.rs @@ -0,0 +1,33 @@ +//! External playback sources. Each source eventually feeds the same +//! [`AudioSink`](crate::sink::AudioSink) trait so the server only has to +//! know about one sink-construction path. + +use crate::sink::AudioOutput; + +/// Configuration for the librespot subprocess source. Constructed by the +/// server from `[librespot]` in zerod.toml plus the per-RPC output choice. +#[derive(Debug, Clone)] +pub struct LibrespotConfig { + /// Path or name of the `librespot` binary. Resolved against `$PATH` + /// when not absolute. + pub binary: String, + /// Spotify Connect device name advertised to phones. + pub name: String, + /// 96 / 160 / 320 (kbps). + pub bitrate: u32, + /// Directory librespot uses for credentials/cache. Empty → librespot + /// default (current directory). Disables on-disk audio caching. + pub cache_path: String, + /// Where decoded PCM goes. + pub output: AudioOutput, +} + +#[cfg(target_os = "linux")] +mod librespot; +#[cfg(target_os = "linux")] +pub use librespot::{spotify_start, spotify_stop}; + +#[cfg(not(target_os = "linux"))] +mod librespot_stub; +#[cfg(not(target_os = "linux"))] +pub use librespot_stub::{spotify_start, spotify_stop}; diff --git a/src/main.rs b/src/main.rs index e614017..7381f86 100644 --- a/src/main.rs +++ b/src/main.rs @@ -204,6 +204,23 @@ enum StreamCmd { /// Per-stream software gain (0..=100), independent of the system mixer. #[command(subcommand)] Volume(StreamVolumeCmd), + /// Spotify Connect via librespot — daemon advertises itself; the + /// phone pushes audio to it. + #[command(subcommand)] + Spotify(SpotifyCmd), +} + +#[derive(Subcommand)] +enum SpotifyCmd { + /// Begin advertising as a Spotify Connect device. + Start { + #[arg(long, value_enum, default_value_t = OutputArg::Cpal)] + output: OutputArg, + #[arg(long)] + pipe_path: Option, + }, + /// Tear down the librespot subprocess. + Stop, } #[derive(Subcommand)] @@ -580,8 +597,9 @@ async fn run_stream(ep: &Endpoint, token: Option, cmd: StreamCmd) -> Res .into_inner(); let state = pb::PlayerState::try_from(r.state).unwrap_or(pb::PlayerState::Unspecified); let out = pb::AudioOutput::try_from(r.output).unwrap_or(pb::AudioOutput::Unspecified); + let src = pb::PlaybackSource::try_from(r.source).unwrap_or(pb::PlaybackSource::Unspecified); println!( - "state={state:?} url={} position_ms={} duration_ms={} is_live={} output={out:?} volume={}% error={:?}", + "state={state:?} source={src:?} url={} position_ms={} duration_ms={} is_live={} output={out:?} volume={}% error={:?}", r.url, r.position_ms, r.duration_ms, r.is_live, r.volume_percent, r.error, ); } @@ -603,6 +621,32 @@ async fn run_stream(ep: &Endpoint, token: Option, cmd: StreamCmd) -> Res println!("ok"); } }, + StreamCmd::Spotify(cmd) => match cmd { + SpotifyCmd::Start { output, pipe_path } => { + let output = match output { + OutputArg::Cpal => pb::AudioOutput::Cpal, + OutputArg::Stdout => pb::AudioOutput::Stdout, + OutputArg::Pipe => pb::AudioOutput::Pipe, + } as i32; + client + .spotify_start(attach_token( + Request::new(pb::SpotifyStartRequest { + output, + pipe_path, + cpal_device: None, + }), + &token, + )) + .await?; + println!("ok"); + } + SpotifyCmd::Stop => { + client + .spotify_stop(attach_token(Request::new(pb::SpotifyStopRequest {}), &token)) + .await?; + println!("ok"); + } + }, } Ok(()) } diff --git a/zerod.toml.example b/zerod.toml.example index 7b7ea04..cbaff95 100644 --- a/zerod.toml.example +++ b/zerod.toml.example @@ -63,3 +63,19 @@ port = 1705 # Forward snapserver push notifications onto the in-process event bus so # `zerod events tail --filter snap` catches external `snapctl` changes. forward_notifications = true + +[librespot] +# When enabled, StreamService.SpotifyStart spawns a librespot subprocess +# wired into the same audio sink as HLS/DASH. Install librespot via +# `apt install librespot` (Debian/Ubuntu) or `brew install librespot`. +enabled = false +# Resolved against $PATH when not absolute. +binary = "librespot" +# Device name as it appears in the Spotify app's Devices list. +name = "zerod" +# 96 / 160 / 320 kbps. +bitrate = 320 +# Where librespot stores credentials. Empty → librespot's default +# (current directory). --disable-audio-cache is always passed, so this +# is creds only — pick a persistent path so you don't re-pair every boot. +cache_path = ""