diff --git a/Cargo.lock b/Cargo.lock index 72bcf3e..9a71bea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3160,6 +3160,7 @@ dependencies = [ "futures-core", "pin-project-lite", "tokio", + "tokio-util", ] [[package]] @@ -4271,6 +4272,7 @@ dependencies = [ "serde", "tokio", "tracing", + "zerod-events", ] [[package]] @@ -4293,6 +4295,14 @@ dependencies = [ "tracing", ] +[[package]] +name = "zerod-events" +version = "0.1.0" +dependencies = [ + "once_cell", + "tokio", +] + [[package]] name = "zerod-proto" version = "0.1.0" @@ -4308,11 +4318,13 @@ version = "0.1.0" dependencies = [ "anyhow", "async-trait", + "futures", "gethostname", "getrandom 0.2.17", "serde", "serde_json", "tokio", + "tokio-stream", "toml", "tonic", "tonic-reflection", @@ -4320,6 +4332,7 @@ dependencies = [ "zerod-bluetooth", "zerod-config", "zerod-discovery", + "zerod-events", "zerod-proto", "zerod-stream", "zerod-systemd", @@ -4344,6 +4357,7 @@ dependencies = [ "tokio", "tracing", "url", + "zerod-events", ] [[package]] @@ -4355,6 +4369,7 @@ dependencies = [ "tokio", "tracing", "zbus", + "zerod-events", ] [[package]] @@ -4365,6 +4380,7 @@ dependencies = [ "anyhow", "serde", "tracing", + "zerod-events", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index a704a47..e9bbe17 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,7 @@ members = [ "crates/systemd", "crates/config", "crates/discovery", + "crates/events", "crates/server", "crates/volume", ] diff --git a/crates/bluetooth/Cargo.toml b/crates/bluetooth/Cargo.toml index dcad4ff..84d80dc 100644 --- a/crates/bluetooth/Cargo.toml +++ b/crates/bluetooth/Cargo.toml @@ -7,6 +7,7 @@ edition = "2021" anyhow = { workspace = true } serde = { workspace = true } tracing = { workspace = true } +zerod-events = { path = "../events" } [target.'cfg(target_os = "linux")'.dependencies] bluer = { version = "0.17", features = ["bluetoothd"] } diff --git a/crates/bluetooth/src/lib.rs b/crates/bluetooth/src/lib.rs index 58cd76a..49f6cd7 100644 --- a/crates/bluetooth/src/lib.rs +++ b/crates/bluetooth/src/lib.rs @@ -24,6 +24,16 @@ mod imp { use std::time::Duration; use tracing::warn; + fn publish_device(d: &BluetoothDevice) { + zerod_events::publish(zerod_events::Event::BluetoothDeviceChanged { + address: d.address.clone(), + name: d.name.clone(), + paired: d.paired, + connected: d.connected, + trusted: d.trusted, + }); + } + async fn adapter() -> Result { let session = bluer::Session::new().await?; let adapter = session.default_adapter().await?; @@ -89,6 +99,9 @@ mod imp { if !device.is_trusted().await? { device.set_trusted(true).await?; } + if let Ok(info) = device_info(&adapter, addr).await { + publish_device(&info); + } Ok(()) } @@ -103,6 +116,9 @@ mod imp { device.set_trusted(true).await?; } device.connect().await?; + if let Ok(info) = device_info(&adapter, addr).await { + publish_device(&info); + } Ok(()) } @@ -110,6 +126,9 @@ mod imp { let adapter = adapter().await?; let addr = Address::from_str(address)?; adapter.device(addr)?.disconnect().await?; + if let Ok(info) = device_info(&adapter, addr).await { + publish_device(&info); + } Ok(()) } @@ -117,6 +136,13 @@ mod imp { let adapter = adapter().await?; let addr = Address::from_str(address)?; adapter.remove_device(addr).await?; + zerod_events::publish(zerod_events::Event::BluetoothDeviceChanged { + address: address.to_string(), + name: String::new(), + paired: false, + connected: false, + trusted: false, + }); Ok(()) } } diff --git a/crates/events/Cargo.toml b/crates/events/Cargo.toml new file mode 100644 index 0000000..d5115b9 --- /dev/null +++ b/crates/events/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "zerod-events" +version = "0.1.0" +edition = "2021" + +[dependencies] +once_cell = { workspace = true } +tokio = { workspace = true } diff --git a/crates/events/src/lib.rs b/crates/events/src/lib.rs new file mode 100644 index 0000000..026f75f --- /dev/null +++ b/crates/events/src/lib.rs @@ -0,0 +1,129 @@ +//! Lossy in-process event bus. Subsystems call [`publish`]; the server fans +//! out to gRPC subscribers via [`subscribe`]. +//! +//! Living in its own leaf crate sidesteps the dependency cycle: `stream`, +//! `bluetooth`, `systemd`, and `volume` can't depend on `zerod-server`, and +//! `zerod-proto` is codegen-only. + +use once_cell::sync::Lazy; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::sync::broadcast; + +/// Capacity of the broadcast ring. Slow subscribers see `Lagged(n)`; the +/// server maps that into a synthetic `LaggedEvent` rather than dropping +/// the stream. +const BUS_CAPACITY: usize = 512; + +static BUS: Lazy> = Lazy::new(|| broadcast::channel(BUS_CAPACITY).0); + +/// Publish-time timestamp + payload. Stamping at emission gives every +/// subscriber the same time even though `broadcast` may drop messages +/// between sender and receiver. +#[derive(Debug, Clone)] +pub struct Envelope { + pub timestamp_ms: i64, + pub event: Event, +} + +#[derive(Debug, Clone)] +pub enum Event { + StreamStateChanged { + state: StreamState, + url: String, + error: Option, + }, + StreamVolumeChanged { + volume_percent: u32, + }, + BluetoothDeviceChanged { + address: String, + name: String, + paired: bool, + connected: bool, + trusted: bool, + }, + BluetoothPairingRequest { + address: String, + passkey: Option, + }, + BluetoothA2dpConnected { + address: String, + name: String, + }, + BluetoothA2dpDisconnected { + address: String, + name: String, + }, + SystemdUnitState { + name: String, + active_state: String, + sub_state: String, + enabled: bool, + }, + VolumeChanged { + card: String, + control: String, + volume_percent: u32, + muted: bool, + }, + SnapcastClientChanged { + client_id: String, + name: String, + volume_percent: u32, + muted: bool, + }, + LibrespotStateChanged { + state: String, + track: Option, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StreamState { + Stopped, + Buffering, + Playing, + Paused, + Errored, +} + +impl Event { + /// Stable kind label used for client-side filtering. Keep these in sync + /// with the docs for `EventsService.Subscribe`. + pub fn kind(&self) -> &'static str { + match self { + Event::StreamStateChanged { .. } => "stream.state", + Event::StreamVolumeChanged { .. } => "stream.volume", + Event::BluetoothDeviceChanged { .. } => "bt.device", + Event::BluetoothPairingRequest { .. } => "bt.pairing", + Event::BluetoothA2dpConnected { .. } => "bt.a2dp.connected", + Event::BluetoothA2dpDisconnected { .. } => "bt.a2dp.disconnected", + Event::SystemdUnitState { .. } => "systemd.unit", + Event::VolumeChanged { .. } => "volume", + Event::SnapcastClientChanged { .. } => "snap.client", + Event::LibrespotStateChanged { .. } => "librespot.state", + } + } +} + +/// Non-blocking emit. Safe to call while holding any Mutex — `broadcast::Sender::send` +/// never blocks. Silently drops if there are no subscribers (which is the +/// common case at startup). +pub fn publish(event: Event) { + let env = Envelope { + timestamp_ms: now_ms(), + event, + }; + let _ = BUS.send(env); +} + +pub fn subscribe() -> broadcast::Receiver { + BUS.subscribe() +} + +fn now_ms() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} diff --git a/crates/proto/build.rs b/crates/proto/build.rs index 5bfb09c..40a1ae5 100644 --- a/crates/proto/build.rs +++ b/crates/proto/build.rs @@ -17,6 +17,7 @@ fn main() -> Result<(), Box> { "proto/zerod/v1alpha1/config.proto", "proto/zerod/v1alpha1/system.proto", "proto/zerod/v1alpha1/volume.proto", + "proto/zerod/v1alpha1/events.proto", ], &["proto"], )?; diff --git a/crates/proto/proto/zerod/v1alpha1/events.proto b/crates/proto/proto/zerod/v1alpha1/events.proto new file mode 100644 index 0000000..87d4e8a --- /dev/null +++ b/crates/proto/proto/zerod/v1alpha1/events.proto @@ -0,0 +1,100 @@ +syntax = "proto3"; +package zerod.v1alpha1; + +import "zerod/v1alpha1/stream.proto"; + +// Single envelope for every event the daemon emits. New event kinds go +// into the `payload` oneof; the field tag is the wire-stable identity. +message Event { + int64 timestamp_ms = 1; + oneof payload { + StreamStateChanged stream_state = 10; + StreamVolumeChanged stream_volume = 11; + BluetoothDeviceChanged bt_device = 20; + BluetoothPairingRequest bt_pairing_request = 21; + BluetoothA2dpConnected bt_a2dp_connected = 22; + BluetoothA2dpDisconnected bt_a2dp_disconnected = 23; + SystemdUnitState systemd_unit = 30; + VolumeChanged volume = 40; + SnapcastClientChanged snap_client = 50; // reserved for PR2 + LibrespotStateChanged librespot_state = 60; // reserved for PR3 + LaggedEvent lagged = 99; + } +} + +message StreamStateChanged { + PlayerState state = 1; + string url = 2; + optional string error = 3; +} + +message StreamVolumeChanged { + uint32 volume_percent = 1; +} + +message BluetoothDeviceChanged { + string address = 1; + string name = 2; + bool paired = 3; + bool connected = 4; + bool trusted = 5; +} + +message BluetoothPairingRequest { + string address = 1; + optional uint32 passkey = 2; +} + +message BluetoothA2dpConnected { + string address = 1; + string name = 2; +} + +message BluetoothA2dpDisconnected { + string address = 1; + string name = 2; +} + +message SystemdUnitState { + string name = 1; + string active_state = 2; + string sub_state = 3; + bool enabled = 4; +} + +message VolumeChanged { + string card = 1; + string control = 2; + uint32 volume_percent = 3; + bool muted = 4; +} + +message SnapcastClientChanged { + string client_id = 1; + string name = 2; + uint32 volume_percent = 3; + bool muted = 4; +} + +message LibrespotStateChanged { + string state = 1; + optional string track = 2; +} + +// Synthetic event emitted by the server when a subscriber falls behind and +// `broadcast::Receiver::recv` returns `Lagged(n)`. Clients can treat this +// as a hint to re-fetch state. +message LaggedEvent { + uint64 dropped = 1; +} + +message SubscribeRequest { + // Filter labels. Empty → receive every event. Supports exact match + // (`stream.state`) or a `.*` wildcard suffix (`bt.*` matches every + // bluetooth event). See `Event.kind()` in the server for the label list. + repeated string kinds = 1; +} + +service EventsService { + rpc Subscribe(SubscribeRequest) returns (stream Event); +} diff --git a/crates/proto/src/generated/zerod.v1alpha1.rs b/crates/proto/src/generated/zerod.v1alpha1.rs index 83d1a56..abc44d4 100644 --- a/crates/proto/src/generated/zerod.v1alpha1.rs +++ b/crates/proto/src/generated/zerod.v1alpha1.rs @@ -3969,3 +3969,450 @@ pub mod volume_service_server { const NAME: &'static str = SERVICE_NAME; } } +/// Single envelope for every event the daemon emits. New event kinds go +/// into the `payload` oneof; the field tag is the wire-stable identity. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct Event { + #[prost(int64, tag = "1")] + pub timestamp_ms: i64, + #[prost( + oneof = "event::Payload", + tags = "10, 11, 20, 21, 22, 23, 30, 40, 50, 60, 99" + )] + pub payload: ::core::option::Option, +} +/// Nested message and enum types in `Event`. +pub mod event { + #[derive(Clone, PartialEq, ::prost::Oneof)] + pub enum Payload { + #[prost(message, tag = "10")] + StreamState(super::StreamStateChanged), + #[prost(message, tag = "11")] + StreamVolume(super::StreamVolumeChanged), + #[prost(message, tag = "20")] + BtDevice(super::BluetoothDeviceChanged), + #[prost(message, tag = "21")] + BtPairingRequest(super::BluetoothPairingRequest), + #[prost(message, tag = "22")] + BtA2dpConnected(super::BluetoothA2dpConnected), + #[prost(message, tag = "23")] + BtA2dpDisconnected(super::BluetoothA2dpDisconnected), + #[prost(message, tag = "30")] + SystemdUnit(super::SystemdUnitState), + #[prost(message, tag = "40")] + Volume(super::VolumeChanged), + /// reserved for PR2 + #[prost(message, tag = "50")] + SnapClient(super::SnapcastClientChanged), + /// reserved for PR3 + #[prost(message, tag = "60")] + LibrespotState(super::LibrespotStateChanged), + #[prost(message, tag = "99")] + Lagged(super::LaggedEvent), + } +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct StreamStateChanged { + #[prost(enumeration = "PlayerState", tag = "1")] + pub state: i32, + #[prost(string, tag = "2")] + pub url: ::prost::alloc::string::String, + #[prost(string, optional, tag = "3")] + pub error: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct StreamVolumeChanged { + #[prost(uint32, tag = "1")] + pub volume_percent: u32, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct BluetoothDeviceChanged { + #[prost(string, tag = "1")] + pub address: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub name: ::prost::alloc::string::String, + #[prost(bool, tag = "3")] + pub paired: bool, + #[prost(bool, tag = "4")] + pub connected: bool, + #[prost(bool, tag = "5")] + pub trusted: bool, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct BluetoothPairingRequest { + #[prost(string, tag = "1")] + pub address: ::prost::alloc::string::String, + #[prost(uint32, optional, tag = "2")] + pub passkey: ::core::option::Option, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct BluetoothA2dpConnected { + #[prost(string, tag = "1")] + pub address: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub name: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct BluetoothA2dpDisconnected { + #[prost(string, tag = "1")] + pub address: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub name: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct SystemdUnitState { + #[prost(string, tag = "1")] + pub name: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub active_state: ::prost::alloc::string::String, + #[prost(string, tag = "3")] + pub sub_state: ::prost::alloc::string::String, + #[prost(bool, tag = "4")] + pub enabled: bool, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct VolumeChanged { + #[prost(string, tag = "1")] + pub card: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub control: ::prost::alloc::string::String, + #[prost(uint32, tag = "3")] + pub volume_percent: u32, + #[prost(bool, tag = "4")] + pub muted: bool, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct SnapcastClientChanged { + #[prost(string, tag = "1")] + pub client_id: ::prost::alloc::string::String, + #[prost(string, tag = "2")] + pub name: ::prost::alloc::string::String, + #[prost(uint32, tag = "3")] + pub volume_percent: u32, + #[prost(bool, tag = "4")] + pub muted: bool, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct LibrespotStateChanged { + #[prost(string, tag = "1")] + pub state: ::prost::alloc::string::String, + #[prost(string, optional, tag = "2")] + pub track: ::core::option::Option<::prost::alloc::string::String>, +} +/// Synthetic event emitted by the server when a subscriber falls behind and +/// `broadcast::Receiver::recv` returns `Lagged(n)`. Clients can treat this +/// as a hint to re-fetch state. +#[derive(Clone, Copy, PartialEq, ::prost::Message)] +pub struct LaggedEvent { + #[prost(uint64, tag = "1")] + pub dropped: u64, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct SubscribeRequest { + /// Filter labels. Empty → receive every event. Supports exact match + /// (`stream.state`) or a `.*` wildcard suffix (`bt.*` matches every + /// bluetooth event). See `Event.kind()` in the server for the label list. + #[prost(string, repeated, tag = "1")] + pub kinds: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, +} +/// Generated client implementations. +pub mod events_service_client { + #![allow( + unused_variables, + dead_code, + missing_docs, + clippy::wildcard_imports, + clippy::let_unit_value, + )] + use tonic::codegen::*; + use tonic::codegen::http::Uri; + #[derive(Debug, Clone)] + pub struct EventsServiceClient { + inner: tonic::client::Grpc, + } + impl EventsServiceClient { + /// Attempt to create a new client by connecting to a given endpoint. + pub async fn connect(dst: D) -> Result + where + D: TryInto, + D::Error: Into, + { + let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; + Ok(Self::new(conn)) + } + } + impl EventsServiceClient + where + T: tonic::client::GrpcService, + T::Error: Into, + T::ResponseBody: Body + std::marker::Send + 'static, + ::Error: Into + std::marker::Send, + { + pub fn new(inner: T) -> Self { + let inner = tonic::client::Grpc::new(inner); + Self { inner } + } + pub fn with_origin(inner: T, origin: Uri) -> Self { + let inner = tonic::client::Grpc::with_origin(inner, origin); + Self { inner } + } + pub fn with_interceptor( + inner: T, + interceptor: F, + ) -> EventsServiceClient> + where + F: tonic::service::Interceptor, + T::ResponseBody: Default, + T: tonic::codegen::Service< + http::Request, + Response = http::Response< + >::ResponseBody, + >, + >, + , + >>::Error: Into + std::marker::Send + std::marker::Sync, + { + EventsServiceClient::new(InterceptedService::new(inner, interceptor)) + } + /// Compress requests with the given encoding. + /// + /// This requires the server to support it otherwise it might respond with an + /// error. + #[must_use] + pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.inner = self.inner.send_compressed(encoding); + self + } + /// Enable decompressing responses. + #[must_use] + pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.inner = self.inner.accept_compressed(encoding); + self + } + /// Limits the maximum size of a decoded message. + /// + /// Default: `4MB` + #[must_use] + pub fn max_decoding_message_size(mut self, limit: usize) -> Self { + self.inner = self.inner.max_decoding_message_size(limit); + self + } + /// Limits the maximum size of an encoded message. + /// + /// Default: `usize::MAX` + #[must_use] + pub fn max_encoding_message_size(mut self, limit: usize) -> Self { + self.inner = self.inner.max_encoding_message_size(limit); + self + } + pub async fn subscribe( + &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.EventsService/Subscribe", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("zerod.v1alpha1.EventsService", "Subscribe")); + self.inner.server_streaming(req, path, codec).await + } + } +} +/// Generated server implementations. +pub mod events_service_server { + #![allow( + unused_variables, + dead_code, + missing_docs, + clippy::wildcard_imports, + clippy::let_unit_value, + )] + use tonic::codegen::*; + /// Generated trait containing gRPC methods that should be implemented for use with EventsServiceServer. + #[async_trait] + pub trait EventsService: std::marker::Send + std::marker::Sync + 'static { + /// Server streaming response type for the Subscribe method. + type SubscribeStream: tonic::codegen::tokio_stream::Stream< + Item = std::result::Result, + > + + std::marker::Send + + 'static; + async fn subscribe( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; + } + #[derive(Debug)] + pub struct EventsServiceServer { + inner: Arc, + accept_compression_encodings: EnabledCompressionEncodings, + send_compression_encodings: EnabledCompressionEncodings, + max_decoding_message_size: Option, + max_encoding_message_size: Option, + } + impl EventsServiceServer { + pub fn new(inner: T) -> Self { + Self::from_arc(Arc::new(inner)) + } + pub fn from_arc(inner: Arc) -> Self { + Self { + inner, + accept_compression_encodings: Default::default(), + send_compression_encodings: Default::default(), + max_decoding_message_size: None, + max_encoding_message_size: None, + } + } + pub fn with_interceptor( + inner: T, + interceptor: F, + ) -> InterceptedService + where + F: tonic::service::Interceptor, + { + InterceptedService::new(Self::new(inner), interceptor) + } + /// Enable decompressing requests with the given encoding. + #[must_use] + pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.accept_compression_encodings.enable(encoding); + self + } + /// Compress responses with the given encoding, if the client supports it. + #[must_use] + pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.send_compression_encodings.enable(encoding); + self + } + /// Limits the maximum size of a decoded message. + /// + /// Default: `4MB` + #[must_use] + pub fn max_decoding_message_size(mut self, limit: usize) -> Self { + self.max_decoding_message_size = Some(limit); + self + } + /// Limits the maximum size of an encoded message. + /// + /// Default: `usize::MAX` + #[must_use] + pub fn max_encoding_message_size(mut self, limit: usize) -> Self { + self.max_encoding_message_size = Some(limit); + self + } + } + impl tonic::codegen::Service> for EventsServiceServer + where + T: EventsService, + B: Body + std::marker::Send + 'static, + B::Error: Into + std::marker::Send + 'static, + { + type Response = http::Response; + type Error = std::convert::Infallible; + type Future = BoxFuture; + fn poll_ready( + &mut self, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + fn call(&mut self, req: http::Request) -> Self::Future { + match req.uri().path() { + "/zerod.v1alpha1.EventsService/Subscribe" => { + #[allow(non_camel_case_types)] + struct SubscribeSvc(pub Arc); + impl< + T: EventsService, + > tonic::server::ServerStreamingService + for SubscribeSvc { + type Response = super::Event; + type ResponseStream = T::SubscribeStream; + 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 { + ::subscribe(&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 = SubscribeSvc(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.server_streaming(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + _ => { + Box::pin(async move { + let mut response = http::Response::new(empty_body()); + let headers = response.headers_mut(); + headers + .insert( + tonic::Status::GRPC_STATUS, + (tonic::Code::Unimplemented as i32).into(), + ); + headers + .insert( + http::header::CONTENT_TYPE, + tonic::metadata::GRPC_CONTENT_TYPE, + ); + Ok(response) + }) + } + } + } + } + impl Clone for EventsServiceServer { + fn clone(&self) -> Self { + let inner = self.inner.clone(); + Self { + inner, + accept_compression_encodings: self.accept_compression_encodings, + send_compression_encodings: self.send_compression_encodings, + max_decoding_message_size: self.max_decoding_message_size, + max_encoding_message_size: self.max_encoding_message_size, + } + } + } + /// Generated gRPC service name + pub const SERVICE_NAME: &str = "zerod.v1alpha1.EventsService"; + impl tonic::server::NamedService for EventsServiceServer { + const NAME: &'static str = SERVICE_NAME; + } +} diff --git a/crates/proto/src/generated/zerod_descriptor.bin b/crates/proto/src/generated/zerod_descriptor.bin index 630a96376336d1eb5eda0c26a483d290cf8d09f7..2474ce364df95701d50ba3f14666795266a11c4c 100644 GIT binary patch delta 5167 zcmZpuTeXUDfxQ&havd(|s??(V6#X*8#GHbRL__`5vedki;>ijIVvJIg*K*0!D}7?) zVs!OA<>`CAbtA zHMotG^uV_0!EDhB2CE4MDRIt7%u7#A@n8tzhUf%~3VCtyg1Hg7@n!irrMamR++d@W zu^SZzR)a7L>I#UkkSiBwQb~MDYFTD-s)PvG9JR?djPiyE4ID`&E?{|~a4w;wlK6te z%%aS^^!TFG!qU{@5(!bTF7?SajFR>N5V;_Tlm|nQKvD^+m{16pU{XnZqESjgd~$wX zUTShlYKnvy*bI=(k`9hWDFx0j84rda{-hEVA)#0<5ts=nnZ*c0#KDGWf*ev0(&2(6 z>A?^r1ThC$RLGT!r?|4XBsDiBzBDhhL_!X1gaSrLLezxjWtM>B030O+pl^Phi1cDK%3ml4*56Xy$fr^>LKjI^{BI6#9qic6Eg z%0V7Y%}Y$mNllS}1Ohvht5n6siyE+RjrE09x%eOnIKCjYC>fM0nI{|a3i}F)aIxl= zf`SWT3WTH7%E%=ej4gc%Npo>Rl68D$3b+*H4B~(Yc&1ED6vbsxJr0Y4xJ0ogT##}| zF$yYaIfGci93e?A){>&c&(Ob3}1)Y7wX?6>{U^ zgliSTEKkr(l~NK#&t_ob3>dw7q`BCHSQr?RC)XNE)-y8*F*7r$a?NDqVq#$sVqs?x zW0dM)R0z(@OV3GFNX;ut&B-rFRY=P(Qb;XJEvf{Oc_j)Z8L0{>iK)5yc?zkynI*+~ z3Vx~OP&u#<6w>p#6f*Nl^1(_Ipm{k#AwMrQKg}8}n3kDZpOd0cl9;ZLSqv5_&n!yS zEiM5iGlk5Q)Vz|+l1e=;E-o%+76wKkW_Au_0(!g$O=%-kAUP;JaY%sebw z^-wuxHV{_|#$yy>=8;x`@jyz|HKF=ggqV3*OrTOMY#^RAjK?U%%qyb@<1q^{^BNjM zwXq5@^RXB~rC8ZOJdkbmpzvd56k_I+)`3Yf3o-NQ8$$K52{H4tSU{!Nz&u$Pk5P!3 zUk+v_8?z8Izo|J?8@mv*0E;?QAv+t02a6GQMj>VaQALPcJrg^#5VL@a8dM*L5VIhQ z97K+ZgAK$Jgz*@Km<5GEF$r=r2eS~fptLNPJSbk7IE9#nShSUd6pB)dQ;W({Q^4sp zAjpUds+p4wqym;;I2nbQg``070`@;Avk!jA3SRv4K>;%;I7c zViuNyB^xegA!cD6J*aQEg_uQHWS~yqW&`o~VLV15W)T4~Sp0J{3o(mGN^@~>F|jZT zu}Cn;FoIJ(Bcl+r1cwyZJVsEemt=y303#Cvn-H@k6EE02MkY}9kmM7F@tB2}CB?*{ z+8BkHrI;YW$jHRV3gU4<71o0+kmBTlav7P0n5Fmyp!%4En5CH@&Szv|Vgd0WVZg}5 z#45xr%?&e?iBX7IS_qQr8JU=vg_xzKAbw#IVv%D|fcOQZN{&MsY&a9hFY-(>5dSkV zF|Z0T%QHb-#RT?@JSW6FCME`EA!d1~dCWpAN(>qh^OzZhn3Xu>z~(W7%u{9(1^a^; zY@9OGIA*YM%A5i)Ic6bdWnmGBwtA4GRG5UIa-hIbf$IZ>jS43ugqfK@VWT1-2+_yH zB*d%=*9Hm=RVFs5As{1F*@qX4unIA&!?m$63Nfp*^Fb|VVHRRm7Zd^~Bo-kSEe2gkV1ZO=ama&R&jJc8 zZLl-ySy@=Xp`{J=H48Yjw4tHJ0uFp_s54kVKGR`>WKR|*MiwDv9VSSMU;(Eq9cYSR zVFCr6ju0e|uz-V3M@AOxC{`gBeFh_lqgWa1g_!lBQOF8%l>yjQtSqcxM;SmJ#R_(m z0n|~fU`H819mNWAlp!=KSi#OPglhvi!w?#StYBvt3J8Lo!6w9F!eGu=56)$5j6%#N z9E#w;VFNkC6zU8%uro}d=COgDVG4By8`v49P-n1#oM8ra1{>HJW^ipFXP7~q!3K7Q z8Ppl|>_RM-40e#<0F{iE91>t>u!Edo1vQTy>|keD!?l5&Va*B2BkW*jSl3HPLak#GVzyy|WI1+l_O^lR1DR>V3CSbu;OuQ9 zDhAcYEW~UJO_=OV%&Z_DBp``R1oYfiAQy`jG51pC?>>T50`7C(kSh_69)r5}eZII6iop7yVY zrX4P@ul=FU-~#*FpA%B3ae;m9FDL}n$0)=c0CheWxDX0}7D8O$FbIGaLR{cNC;(b5 zae>1iKuVfxAE>3vEyNPW5Wy%_zmZWPxH7LKBef(m8QOdWwP#CGQxuXa!EI7-gQ7^G zJR>zvAyJ_i*3d3eNK4GgDON~I&B)A4QAo^7;ZjIQD#}kx0rj!1tb$ULQ!~p_i>$1Q zQj^OP6pB(yN{jM}6%rs_1&ut-1U&`kdT?L7SRpwvPXRPe18P`j7IP^i7AqtwWMqOG z+W87ask&*YCCM2I;64eoJC8a&3QDPX57b8JQ=s6%^%{> + Send + 'static>>; + + async fn subscribe( + &self, + req: Request, + ) -> Result, Status> { + let kinds = req.into_inner().kinds; + let rx = zerod_events::subscribe(); + let stream = BroadcastStream::new(rx).filter_map(move |item| match item { + Ok(env) => { + if kinds_match(&kinds, env.event.kind()) { + Some(Ok(envelope_to_proto(env))) + } else { + None + } + } + Err(BroadcastStreamRecvError::Lagged(n)) => Some(Ok(lagged_proto(n))), + }); + Ok(Response::new(Box::pin(stream))) + } +} + +fn kinds_match(filters: &[String], kind: &str) -> bool { + if filters.is_empty() { + return true; + } + filters.iter().any(|f| { + if let Some(prefix) = f.strip_suffix(".*") { + kind.starts_with(prefix) + } else { + f == kind + } + }) +} + +fn envelope_to_proto(env: Envelope) -> pb::Event { + pb::Event { + timestamp_ms: env.timestamp_ms, + payload: Some(event_to_payload(env.event)), + } +} + +fn event_to_payload(event: Event) -> pb::event::Payload { + use pb::event::Payload; + match event { + Event::StreamStateChanged { state, url, error } => { + Payload::StreamState(pb::StreamStateChanged { + state: stream_state_to_pb(state) as i32, + url, + error, + }) + } + Event::StreamVolumeChanged { volume_percent } => { + Payload::StreamVolume(pb::StreamVolumeChanged { volume_percent }) + } + Event::BluetoothDeviceChanged { + address, + name, + paired, + connected, + trusted, + } => Payload::BtDevice(pb::BluetoothDeviceChanged { + address, + name, + paired, + connected, + trusted, + }), + Event::BluetoothPairingRequest { address, passkey } => { + Payload::BtPairingRequest(pb::BluetoothPairingRequest { address, passkey }) + } + Event::BluetoothA2dpConnected { address, name } => { + Payload::BtA2dpConnected(pb::BluetoothA2dpConnected { address, name }) + } + Event::BluetoothA2dpDisconnected { address, name } => { + Payload::BtA2dpDisconnected(pb::BluetoothA2dpDisconnected { address, name }) + } + Event::SystemdUnitState { + name, + active_state, + sub_state, + enabled, + } => Payload::SystemdUnit(pb::SystemdUnitState { + name, + active_state, + sub_state, + enabled, + }), + Event::VolumeChanged { + card, + control, + volume_percent, + muted, + } => Payload::Volume(pb::VolumeChanged { + card, + control, + volume_percent, + muted, + }), + Event::SnapcastClientChanged { + client_id, + name, + volume_percent, + muted, + } => Payload::SnapClient(pb::SnapcastClientChanged { + client_id, + name, + volume_percent, + muted, + }), + Event::LibrespotStateChanged { state, track } => { + Payload::LibrespotState(pb::LibrespotStateChanged { state, track }) + } + } +} + +fn stream_state_to_pb(s: StreamState) -> pb::PlayerState { + match s { + StreamState::Stopped => pb::PlayerState::Stopped, + StreamState::Buffering => pb::PlayerState::Buffering, + StreamState::Playing => pb::PlayerState::Playing, + StreamState::Paused => pb::PlayerState::Paused, + StreamState::Errored => pb::PlayerState::Errored, + } +} + +fn lagged_proto(n: u64) -> pb::Event { + pb::Event { + timestamp_ms: 0, + payload: Some(pb::event::Payload::Lagged(pb::LaggedEvent { dropped: n })), + } +} diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 80d1f87..93df97e 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -2,6 +2,7 @@ mod bluetooth; mod config; +mod events; mod settings; mod stream; mod system; @@ -16,9 +17,9 @@ use std::sync::Arc; use tonic::transport::Server; use zerod_proto::v1alpha1::{ bluetooth_service_server::BluetoothServiceServer, - config_service_server::ConfigServiceServer, stream_service_server::StreamServiceServer, - system_service_server::SystemServiceServer, systemd_service_server::SystemdServiceServer, - volume_service_server::VolumeServiceServer, + config_service_server::ConfigServiceServer, events_service_server::EventsServiceServer, + stream_service_server::StreamServiceServer, system_service_server::SystemServiceServer, + systemd_service_server::SystemdServiceServer, volume_service_server::VolumeServiceServer, }; pub async fn serve(settings: Settings) -> Result<()> { @@ -77,6 +78,10 @@ pub async fn serve(settings: Settings) -> Result<()> { )) .add_service(VolumeServiceServer::with_interceptor( volume::VolumeSvc::default(), + interceptor.clone(), + )) + .add_service(EventsServiceServer::with_interceptor( + events::EventsSvc::default(), interceptor, )) .serve(addr) diff --git a/crates/stream/Cargo.toml b/crates/stream/Cargo.toml index e8baa33..06c2fd0 100644 --- a/crates/stream/Cargo.toml +++ b/crates/stream/Cargo.toml @@ -16,6 +16,7 @@ symphonia = { version = "0.5", default-features = false, features = ["isomp4", " tokio = { workspace = true } tracing = { workspace = true } url = { workspace = true } +zerod-events = { path = "../events" } # On Linux we go straight to libasound via the `alsa` crate (the # `Access::RWInterleaved` / `snd_pcm_writei` path), bypassing cpal's diff --git a/crates/stream/src/player.rs b/crates/stream/src/player.rs index b9dd2fe..f2f3733 100644 --- a/crates/stream/src/player.rs +++ b/crates/stream/src/player.rs @@ -98,6 +98,12 @@ struct Player { impl Player { 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 { + state: to_event_state(s), + url: self.url.clone(), + error, + }); } fn state(&self) -> PlayerState { @@ -462,9 +468,22 @@ pub fn status() -> Status { pub fn set_volume(percent: u32) { let p = percent.min(100); GLOBAL_VOLUME.store(p, Ordering::Relaxed); + zerod_events::publish(zerod_events::Event::StreamVolumeChanged { + volume_percent: p, + }); tracing::info!("stream: set volume to {}%", p); } +fn to_event_state(s: PlayerState) -> zerod_events::StreamState { + match s { + PlayerState::Stopped => zerod_events::StreamState::Stopped, + PlayerState::Buffering => zerod_events::StreamState::Buffering, + PlayerState::Playing => zerod_events::StreamState::Playing, + PlayerState::Paused => zerod_events::StreamState::Paused, + PlayerState::Errored => zerod_events::StreamState::Errored, + } +} + pub fn volume() -> u32 { GLOBAL_VOLUME.load(Ordering::Relaxed) } diff --git a/crates/systemd/Cargo.toml b/crates/systemd/Cargo.toml index 9946652..fd99052 100644 --- a/crates/systemd/Cargo.toml +++ b/crates/systemd/Cargo.toml @@ -7,6 +7,7 @@ edition = "2021" anyhow = { workspace = true } serde = { workspace = true } tracing = { workspace = true } +zerod-events = { path = "../events" } [target.'cfg(target_os = "linux")'.dependencies] zbus = { version = "4", default-features = false, features = ["tokio"] } diff --git a/crates/systemd/src/lib.rs b/crates/systemd/src/lib.rs index 42935c9..c8f86d4 100644 --- a/crates/systemd/src/lib.rs +++ b/crates/systemd/src/lib.rs @@ -63,6 +63,21 @@ mod imp { use super::{ActiveState, UnitAllowlist, UnitStatus}; use anyhow::{Context, Result}; + fn publish_unit(u: &UnitStatus) { + zerod_events::publish(zerod_events::Event::SystemdUnitState { + name: u.name.clone(), + active_state: u.active_state.clone(), + sub_state: u.sub_state.clone(), + enabled: u.enabled, + }); + } + + async fn publish_after(allow: &UnitAllowlist, name: &str) { + if let Ok(s) = status(allow, name).await { + publish_unit(&s); + } + } + #[zbus::proxy( interface = "org.freedesktop.systemd1.Manager", default_service = "org.freedesktop.systemd1", @@ -170,24 +185,28 @@ mod imp { allow.check(name)?; let (_c, mgr) = connect().await?; mgr.start_unit(name, "replace").await.with_context(|| format!("start {name}"))?; + publish_after(allow, name).await; Ok(()) } pub async fn stop(allow: &UnitAllowlist, name: &str) -> Result<()> { allow.check(name)?; let (_c, mgr) = connect().await?; mgr.stop_unit(name, "replace").await.with_context(|| format!("stop {name}"))?; + publish_after(allow, name).await; Ok(()) } pub async fn restart(allow: &UnitAllowlist, name: &str) -> Result<()> { allow.check(name)?; let (_c, mgr) = connect().await?; mgr.restart_unit(name, "replace").await.with_context(|| format!("restart {name}"))?; + publish_after(allow, name).await; Ok(()) } pub async fn reload(allow: &UnitAllowlist, name: &str) -> Result<()> { allow.check(name)?; let (_c, mgr) = connect().await?; mgr.reload_unit(name, "replace").await.with_context(|| format!("reload {name}"))?; + publish_after(allow, name).await; Ok(()) } pub async fn enable(allow: &UnitAllowlist, name: &str) -> Result<()> { @@ -198,6 +217,7 @@ mod imp { .with_context(|| format!("enable {name}"))?; // EnableUnitFiles doesn't apply changes until daemon-reload. mgr.reload().await.ok(); + publish_after(allow, name).await; Ok(()) } pub async fn disable(allow: &UnitAllowlist, name: &str) -> Result<()> { @@ -207,6 +227,7 @@ mod imp { .await .with_context(|| format!("disable {name}"))?; mgr.reload().await.ok(); + publish_after(allow, name).await; Ok(()) } diff --git a/crates/volume/Cargo.toml b/crates/volume/Cargo.toml index b32a3dc..dc36c0c 100644 --- a/crates/volume/Cargo.toml +++ b/crates/volume/Cargo.toml @@ -7,6 +7,7 @@ edition = "2021" anyhow = { workspace = true } serde = { workspace = true } tracing = { workspace = true } +zerod-events = { path = "../events" } [target.'cfg(target_os = "linux")'.dependencies] alsa = "0.9" diff --git a/crates/volume/src/lib.rs b/crates/volume/src/lib.rs index c6402f3..68e04dc 100644 --- a/crates/volume/src/lib.rs +++ b/crates/volume/src/lib.rs @@ -149,6 +149,7 @@ mod imp { selem .set_playback_volume_all(target) .with_context(|| format!("set_playback_volume_all({target})"))?; + publish_volume(sel); Ok(()) } @@ -164,8 +165,20 @@ mod imp { selem .set_playback_switch_all(if muted { 0 } else { 1 }) .with_context(|| format!("set_playback_switch_all(muted={muted})"))?; + publish_volume(sel); Ok(()) } + + fn publish_volume(sel: &Selector) { + if let Ok(st) = get(sel) { + zerod_events::publish(zerod_events::Event::VolumeChanged { + card: sel.card.clone(), + control: sel.control.clone(), + volume_percent: st.volume_percent, + muted: st.muted, + }); + } + } } #[cfg(not(target_os = "linux"))] diff --git a/src/main.rs b/src/main.rs index 1115b88..03cac67 100644 --- a/src/main.rs +++ b/src/main.rs @@ -103,10 +103,24 @@ enum Command { /// Manage zerod's own systemd --user unit (Linux only). #[command(subcommand)] Service(ServiceCmd), + /// Subscribe to server-side events (server-streaming). + #[command(subcommand)] + Events(EventsCmd), /// Browse mDNS for zerod servers on the LAN. Discover, } +#[derive(Subcommand)] +enum EventsCmd { + /// Print events as JSON, one per line, until interrupted. + Tail { + /// Kind filter(s). Empty → every event. Exact match (`stream.state`) + /// or `.*` suffix wildcard (`bt.*`). Repeat the flag for OR. + #[arg(long)] + filter: Vec, + }, +} + #[derive(Subcommand)] enum ServiceCmd { /// Write ~/.config/systemd/user/zerod.service pointing at this binary. @@ -293,6 +307,7 @@ async fn main() -> Result<()> { Some(Command::Config(cmd)) => run_config(&endpoint.unwrap(), bearer_token, cmd).await, Some(Command::System(cmd)) => run_system(&endpoint.unwrap(), bearer_token, cmd).await, Some(Command::Volume(cmd)) => run_volume(&endpoint.unwrap(), bearer_token, cmd).await, + Some(Command::Events(cmd)) => run_events(&endpoint.unwrap(), bearer_token, cmd).await, } } @@ -720,6 +735,99 @@ async fn run_config(ep: &Endpoint, token: Option, cmd: ConfigCmd) -> Res Ok(()) } +// --- events ---------------------------------------------------------------- + +async fn run_events(ep: &Endpoint, token: Option, cmd: EventsCmd) -> Result<()> { + let ch = channel(ep).await?; + let mut client = pb::events_service_client::EventsServiceClient::new(ch); + match cmd { + EventsCmd::Tail { filter } => { + let mut stream = client + .subscribe(attach_token( + Request::new(pb::SubscribeRequest { kinds: filter }), + &token, + )) + .await? + .into_inner(); + while let Some(ev) = stream.message().await? { + println!("{}", serde_json::to_string(&event_to_json(&ev))?); + } + } + } + Ok(()) +} + +fn event_to_json(ev: &pb::Event) -> serde_json::Value { + use serde_json::json; + let (kind, payload) = match &ev.payload { + Some(pb::event::Payload::StreamState(p)) => { + let state = pb::PlayerState::try_from(p.state) + .unwrap_or(pb::PlayerState::Unspecified); + ( + "stream.state", + json!({"state": format!("{state:?}"), "url": p.url, "error": p.error}), + ) + } + Some(pb::event::Payload::StreamVolume(p)) => ( + "stream.volume", + json!({"volume_percent": p.volume_percent}), + ), + Some(pb::event::Payload::BtDevice(p)) => ( + "bt.device", + json!({ + "address": p.address, "name": p.name, + "paired": p.paired, "connected": p.connected, "trusted": p.trusted, + }), + ), + Some(pb::event::Payload::BtPairingRequest(p)) => ( + "bt.pairing", + json!({"address": p.address, "passkey": p.passkey}), + ), + Some(pb::event::Payload::BtA2dpConnected(p)) => ( + "bt.a2dp.connected", + json!({"address": p.address, "name": p.name}), + ), + Some(pb::event::Payload::BtA2dpDisconnected(p)) => ( + "bt.a2dp.disconnected", + json!({"address": p.address, "name": p.name}), + ), + Some(pb::event::Payload::SystemdUnit(p)) => ( + "systemd.unit", + json!({ + "name": p.name, "active_state": p.active_state, + "sub_state": p.sub_state, "enabled": p.enabled, + }), + ), + Some(pb::event::Payload::Volume(p)) => ( + "volume", + json!({ + "card": p.card, "control": p.control, + "volume_percent": p.volume_percent, "muted": p.muted, + }), + ), + Some(pb::event::Payload::SnapClient(p)) => ( + "snap.client", + json!({ + "client_id": p.client_id, "name": p.name, + "volume_percent": p.volume_percent, "muted": p.muted, + }), + ), + Some(pb::event::Payload::LibrespotState(p)) => ( + "librespot.state", + json!({"state": p.state, "track": p.track}), + ), + Some(pb::event::Payload::Lagged(p)) => { + ("lagged", json!({"dropped": p.dropped})) + } + None => ("unknown", json!({})), + }; + json!({ + "timestamp_ms": ev.timestamp_ms, + "kind": kind, + "payload": payload, + }) +} + // --- system ---------------------------------------------------------------- async fn run_system(ep: &Endpoint, token: Option, cmd: SystemCmd) -> Result<()> {