+13

Cargo.lock

View file Reviewed

··· 4334 4334 "zerod-discovery", 4335 4335 "zerod-events", 4336 4336 "zerod-proto", 4337 4337 + "zerod-snapcast", 4337 4338 "zerod-stream", 4338 4339 "zerod-systemd", 4339 4340 "zerod-volume", 4341 4341 + ] 4342 4342 + 4343 4343 + [[package]] 4344 4344 + name = "zerod-snapcast" 4345 4345 + version = "0.1.0" 4346 4346 + dependencies = [ 4347 4347 + "anyhow", 4348 4348 + "serde", 4349 4349 + "serde_json", 4350 4350 + "tokio", 4351 4351 + "tracing", 4352 4352 + "zerod-events", 4340 4353 ] 4341 4354 4342 4355 [[package]]

+1

Cargo.toml

View file Reviewed

··· 17 17 "crates/discovery", 18 18 "crates/events", 19 19 "crates/server", 20 20 + "crates/snapcast", 20 21 "crates/volume", 21 22 ] 22 23

+1

crates/proto/build.rs

View file Reviewed

··· 18 18 "proto/zerod/v1alpha1/system.proto", 19 19 "proto/zerod/v1alpha1/volume.proto", 20 20 "proto/zerod/v1alpha1/events.proto", 21 21 + "proto/zerod/v1alpha1/snapcast.proto", 21 22 ], 22 23 &["proto"], 23 24 )?;

+104

crates/proto/proto/zerod/v1alpha1/snapcast.proto

View file Reviewed

··· 1 1 + syntax = "proto3"; 2 2 + package zerod.v1alpha1; 3 3 + 4 4 + // MVP wrapper around snapserver's JSON-RPC 2.0 API. See: 5 5 + // https://github.com/badaix/snapcast/blob/master/doc/json_rpc_api/control.md 6 6 + 7 7 + message SnapVolume { 8 8 + uint32 percent = 1; 9 9 + bool muted = 2; 10 10 + } 11 11 + 12 12 + message SnapClient { 13 13 + string id = 1; 14 14 + string name = 2; // config.name (best-effort display name) 15 15 + bool connected = 3; 16 16 + uint32 volume_percent = 4; 17 17 + bool muted = 5; 18 18 + int32 latency_ms = 6; 19 19 + string mac = 7; 20 20 + string host = 8; // host.name 21 21 + string version = 9; // snapclient.version 22 22 + } 23 23 + 24 24 + message SnapGroup { 25 25 + string id = 1; 26 26 + string name = 2; 27 27 + string stream_id = 3; 28 28 + bool muted = 4; 29 29 + repeated string client_ids = 5; 30 30 + } 31 31 + 32 32 + message SnapStream { 33 33 + string id = 1; 34 34 + string status = 2; // "playing" / "idle" / etc. 35 35 + } 36 36 + 37 37 + message SnapServer { 38 38 + repeated SnapGroup groups = 1; 39 39 + repeated SnapStream streams = 2; 40 40 + } 41 41 + 42 42 + message GetServerStatusRequest {} 43 43 + message GetServerStatusResponse { 44 44 + SnapServer server = 1; 45 45 + } 46 46 + 47 47 + message ListClientsRequest {} 48 48 + message ListClientsResponse { 49 49 + repeated SnapClient clients = 1; 50 50 + } 51 51 + 52 52 + message ListSnapStreamsRequest {} 53 53 + message ListSnapStreamsResponse { 54 54 + repeated SnapStream streams = 1; 55 55 + } 56 56 + 57 57 + message SetClientVolumeRequest { 58 58 + string client_id = 1; 59 59 + uint32 volume_percent = 2; // 0..=100 60 60 + bool muted = 3; 61 61 + } 62 62 + message SetClientVolumeResponse {} 63 63 + 64 64 + message SetClientLatencyRequest { 65 65 + string client_id = 1; 66 66 + int32 latency_ms = 2; 67 67 + } 68 68 + message SetClientLatencyResponse {} 69 69 + 70 70 + message SetClientNameRequest { 71 71 + string client_id = 1; 72 72 + string name = 2; 73 73 + } 74 74 + message SetClientNameResponse {} 75 75 + 76 76 + message SetGroupStreamRequest { 77 77 + string group_id = 1; 78 78 + string stream_id = 2; 79 79 + } 80 80 + message SetGroupStreamResponse {} 81 81 + 82 82 + message SetGroupMuteRequest { 83 83 + string group_id = 1; 84 84 + bool muted = 2; 85 85 + } 86 86 + message SetGroupMuteResponse {} 87 87 + 88 88 + message SetGroupClientsRequest { 89 89 + string group_id = 1; 90 90 + repeated string client_ids = 2; 91 91 + } 92 92 + message SetGroupClientsResponse {} 93 93 + 94 94 + service SnapcastService { 95 95 + rpc GetServerStatus(GetServerStatusRequest) returns (GetServerStatusResponse); 96 96 + rpc ListClients(ListClientsRequest) returns (ListClientsResponse); 97 97 + rpc ListSnapStreams(ListSnapStreamsRequest) returns (ListSnapStreamsResponse); 98 98 + rpc SetClientVolume(SetClientVolumeRequest) returns (SetClientVolumeResponse); 99 99 + rpc SetClientLatency(SetClientLatencyRequest) returns (SetClientLatencyResponse); 100 100 + rpc SetClientName(SetClientNameRequest) returns (SetClientNameResponse); 101 101 + rpc SetGroupStream(SetGroupStreamRequest) returns (SetGroupStreamResponse); 102 102 + rpc SetGroupMute(SetGroupMuteRequest) returns (SetGroupMuteResponse); 103 103 + rpc SetGroupClients(SetGroupClientsRequest) returns (SetGroupClientsResponse); 104 104 + }

+1068

crates/proto/src/generated/zerod.v1alpha1.rs

View file Reviewed

··· 4416 4416 const NAME: &'static str = SERVICE_NAME; 4417 4417 } 4418 4418 } 4419 4419 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4420 4420 + pub struct SnapVolume { 4421 4421 + #[prost(uint32, tag = "1")] 4422 4422 + pub percent: u32, 4423 4423 + #[prost(bool, tag = "2")] 4424 4424 + pub muted: bool, 4425 4425 + } 4426 4426 + #[derive(Clone, PartialEq, ::prost::Message)] 4427 4427 + pub struct SnapClient { 4428 4428 + #[prost(string, tag = "1")] 4429 4429 + pub id: ::prost::alloc::string::String, 4430 4430 + /// config.name (best-effort display name) 4431 4431 + #[prost(string, tag = "2")] 4432 4432 + pub name: ::prost::alloc::string::String, 4433 4433 + #[prost(bool, tag = "3")] 4434 4434 + pub connected: bool, 4435 4435 + #[prost(uint32, tag = "4")] 4436 4436 + pub volume_percent: u32, 4437 4437 + #[prost(bool, tag = "5")] 4438 4438 + pub muted: bool, 4439 4439 + #[prost(int32, tag = "6")] 4440 4440 + pub latency_ms: i32, 4441 4441 + #[prost(string, tag = "7")] 4442 4442 + pub mac: ::prost::alloc::string::String, 4443 4443 + /// host.name 4444 4444 + #[prost(string, tag = "8")] 4445 4445 + pub host: ::prost::alloc::string::String, 4446 4446 + /// snapclient.version 4447 4447 + #[prost(string, tag = "9")] 4448 4448 + pub version: ::prost::alloc::string::String, 4449 4449 + } 4450 4450 + #[derive(Clone, PartialEq, ::prost::Message)] 4451 4451 + pub struct SnapGroup { 4452 4452 + #[prost(string, tag = "1")] 4453 4453 + pub id: ::prost::alloc::string::String, 4454 4454 + #[prost(string, tag = "2")] 4455 4455 + pub name: ::prost::alloc::string::String, 4456 4456 + #[prost(string, tag = "3")] 4457 4457 + pub stream_id: ::prost::alloc::string::String, 4458 4458 + #[prost(bool, tag = "4")] 4459 4459 + pub muted: bool, 4460 4460 + #[prost(string, repeated, tag = "5")] 4461 4461 + pub client_ids: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, 4462 4462 + } 4463 4463 + #[derive(Clone, PartialEq, ::prost::Message)] 4464 4464 + pub struct SnapStream { 4465 4465 + #[prost(string, tag = "1")] 4466 4466 + pub id: ::prost::alloc::string::String, 4467 4467 + /// "playing" / "idle" / etc. 4468 4468 + #[prost(string, tag = "2")] 4469 4469 + pub status: ::prost::alloc::string::String, 4470 4470 + } 4471 4471 + #[derive(Clone, PartialEq, ::prost::Message)] 4472 4472 + pub struct SnapServer { 4473 4473 + #[prost(message, repeated, tag = "1")] 4474 4474 + pub groups: ::prost::alloc::vec::Vec<SnapGroup>, 4475 4475 + #[prost(message, repeated, tag = "2")] 4476 4476 + pub streams: ::prost::alloc::vec::Vec<SnapStream>, 4477 4477 + } 4478 4478 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4479 4479 + pub struct GetServerStatusRequest {} 4480 4480 + #[derive(Clone, PartialEq, ::prost::Message)] 4481 4481 + pub struct GetServerStatusResponse { 4482 4482 + #[prost(message, optional, tag = "1")] 4483 4483 + pub server: ::core::option::Option<SnapServer>, 4484 4484 + } 4485 4485 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4486 4486 + pub struct ListClientsRequest {} 4487 4487 + #[derive(Clone, PartialEq, ::prost::Message)] 4488 4488 + pub struct ListClientsResponse { 4489 4489 + #[prost(message, repeated, tag = "1")] 4490 4490 + pub clients: ::prost::alloc::vec::Vec<SnapClient>, 4491 4491 + } 4492 4492 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4493 4493 + pub struct ListSnapStreamsRequest {} 4494 4494 + #[derive(Clone, PartialEq, ::prost::Message)] 4495 4495 + pub struct ListSnapStreamsResponse { 4496 4496 + #[prost(message, repeated, tag = "1")] 4497 4497 + pub streams: ::prost::alloc::vec::Vec<SnapStream>, 4498 4498 + } 4499 4499 + #[derive(Clone, PartialEq, ::prost::Message)] 4500 4500 + pub struct SetClientVolumeRequest { 4501 4501 + #[prost(string, tag = "1")] 4502 4502 + pub client_id: ::prost::alloc::string::String, 4503 4503 + /// 0..=100 4504 4504 + #[prost(uint32, tag = "2")] 4505 4505 + pub volume_percent: u32, 4506 4506 + #[prost(bool, tag = "3")] 4507 4507 + pub muted: bool, 4508 4508 + } 4509 4509 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4510 4510 + pub struct SetClientVolumeResponse {} 4511 4511 + #[derive(Clone, PartialEq, ::prost::Message)] 4512 4512 + pub struct SetClientLatencyRequest { 4513 4513 + #[prost(string, tag = "1")] 4514 4514 + pub client_id: ::prost::alloc::string::String, 4515 4515 + #[prost(int32, tag = "2")] 4516 4516 + pub latency_ms: i32, 4517 4517 + } 4518 4518 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4519 4519 + pub struct SetClientLatencyResponse {} 4520 4520 + #[derive(Clone, PartialEq, ::prost::Message)] 4521 4521 + pub struct SetClientNameRequest { 4522 4522 + #[prost(string, tag = "1")] 4523 4523 + pub client_id: ::prost::alloc::string::String, 4524 4524 + #[prost(string, tag = "2")] 4525 4525 + pub name: ::prost::alloc::string::String, 4526 4526 + } 4527 4527 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4528 4528 + pub struct SetClientNameResponse {} 4529 4529 + #[derive(Clone, PartialEq, ::prost::Message)] 4530 4530 + pub struct SetGroupStreamRequest { 4531 4531 + #[prost(string, tag = "1")] 4532 4532 + pub group_id: ::prost::alloc::string::String, 4533 4533 + #[prost(string, tag = "2")] 4534 4534 + pub stream_id: ::prost::alloc::string::String, 4535 4535 + } 4536 4536 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4537 4537 + pub struct SetGroupStreamResponse {} 4538 4538 + #[derive(Clone, PartialEq, ::prost::Message)] 4539 4539 + pub struct SetGroupMuteRequest { 4540 4540 + #[prost(string, tag = "1")] 4541 4541 + pub group_id: ::prost::alloc::string::String, 4542 4542 + #[prost(bool, tag = "2")] 4543 4543 + pub muted: bool, 4544 4544 + } 4545 4545 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4546 4546 + pub struct SetGroupMuteResponse {} 4547 4547 + #[derive(Clone, PartialEq, ::prost::Message)] 4548 4548 + pub struct SetGroupClientsRequest { 4549 4549 + #[prost(string, tag = "1")] 4550 4550 + pub group_id: ::prost::alloc::string::String, 4551 4551 + #[prost(string, repeated, tag = "2")] 4552 4552 + pub client_ids: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, 4553 4553 + } 4554 4554 + #[derive(Clone, Copy, PartialEq, ::prost::Message)] 4555 4555 + pub struct SetGroupClientsResponse {} 4556 4556 + /// Generated client implementations. 4557 4557 + pub mod snapcast_service_client { 4558 4558 + #![allow( 4559 4559 + unused_variables, 4560 4560 + dead_code, 4561 4561 + missing_docs, 4562 4562 + clippy::wildcard_imports, 4563 4563 + clippy::let_unit_value, 4564 4564 + )] 4565 4565 + use tonic::codegen::*; 4566 4566 + use tonic::codegen::http::Uri; 4567 4567 + #[derive(Debug, Clone)] 4568 4568 + pub struct SnapcastServiceClient<T> { 4569 4569 + inner: tonic::client::Grpc<T>, 4570 4570 + } 4571 4571 + impl SnapcastServiceClient<tonic::transport::Channel> { 4572 4572 + /// Attempt to create a new client by connecting to a given endpoint. 4573 4573 + pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error> 4574 4574 + where 4575 4575 + D: TryInto<tonic::transport::Endpoint>, 4576 4576 + D::Error: Into<StdError>, 4577 4577 + { 4578 4578 + let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; 4579 4579 + Ok(Self::new(conn)) 4580 4580 + } 4581 4581 + } 4582 4582 + impl<T> SnapcastServiceClient<T> 4583 4583 + where 4584 4584 + T: tonic::client::GrpcService<tonic::body::BoxBody>, 4585 4585 + T::Error: Into<StdError>, 4586 4586 + T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static, 4587 4587 + <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send, 4588 4588 + { 4589 4589 + pub fn new(inner: T) -> Self { 4590 4590 + let inner = tonic::client::Grpc::new(inner); 4591 4591 + Self { inner } 4592 4592 + } 4593 4593 + pub fn with_origin(inner: T, origin: Uri) -> Self { 4594 4594 + let inner = tonic::client::Grpc::with_origin(inner, origin); 4595 4595 + Self { inner } 4596 4596 + } 4597 4597 + pub fn with_interceptor<F>( 4598 4598 + inner: T, 4599 4599 + interceptor: F, 4600 4600 + ) -> SnapcastServiceClient<InterceptedService<T, F>> 4601 4601 + where 4602 4602 + F: tonic::service::Interceptor, 4603 4603 + T::ResponseBody: Default, 4604 4604 + T: tonic::codegen::Service< 4605 4605 + http::Request<tonic::body::BoxBody>, 4606 4606 + Response = http::Response< 4607 4607 + <T as tonic::client::GrpcService<tonic::body::BoxBody>>::ResponseBody, 4608 4608 + >, 4609 4609 + >, 4610 4610 + <T as tonic::codegen::Service< 4611 4611 + http::Request<tonic::body::BoxBody>, 4612 4612 + >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync, 4613 4613 + { 4614 4614 + SnapcastServiceClient::new(InterceptedService::new(inner, interceptor)) 4615 4615 + } 4616 4616 + /// Compress requests with the given encoding. 4617 4617 + /// 4618 4618 + /// This requires the server to support it otherwise it might respond with an 4619 4619 + /// error. 4620 4620 + #[must_use] 4621 4621 + pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { 4622 4622 + self.inner = self.inner.send_compressed(encoding); 4623 4623 + self 4624 4624 + } 4625 4625 + /// Enable decompressing responses. 4626 4626 + #[must_use] 4627 4627 + pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { 4628 4628 + self.inner = self.inner.accept_compressed(encoding); 4629 4629 + self 4630 4630 + } 4631 4631 + /// Limits the maximum size of a decoded message. 4632 4632 + /// 4633 4633 + /// Default: `4MB` 4634 4634 + #[must_use] 4635 4635 + pub fn max_decoding_message_size(mut self, limit: usize) -> Self { 4636 4636 + self.inner = self.inner.max_decoding_message_size(limit); 4637 4637 + self 4638 4638 + } 4639 4639 + /// Limits the maximum size of an encoded message. 4640 4640 + /// 4641 4641 + /// Default: `usize::MAX` 4642 4642 + #[must_use] 4643 4643 + pub fn max_encoding_message_size(mut self, limit: usize) -> Self { 4644 4644 + self.inner = self.inner.max_encoding_message_size(limit); 4645 4645 + self 4646 4646 + } 4647 4647 + pub async fn get_server_status( 4648 4648 + &mut self, 4649 4649 + request: impl tonic::IntoRequest<super::GetServerStatusRequest>, 4650 4650 + ) -> std::result::Result< 4651 4651 + tonic::Response<super::GetServerStatusResponse>, 4652 4652 + tonic::Status, 4653 4653 + > { 4654 4654 + self.inner 4655 4655 + .ready() 4656 4656 + .await 4657 4657 + .map_err(|e| { 4658 4658 + tonic::Status::unknown( 4659 4659 + format!("Service was not ready: {}", e.into()), 4660 4660 + ) 4661 4661 + })?; 4662 4662 + let codec = tonic::codec::ProstCodec::default(); 4663 4663 + let path = http::uri::PathAndQuery::from_static( 4664 4664 + "/zerod.v1alpha1.SnapcastService/GetServerStatus", 4665 4665 + ); 4666 4666 + let mut req = request.into_request(); 4667 4667 + req.extensions_mut() 4668 4668 + .insert( 4669 4669 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "GetServerStatus"), 4670 4670 + ); 4671 4671 + self.inner.unary(req, path, codec).await 4672 4672 + } 4673 4673 + pub async fn list_clients( 4674 4674 + &mut self, 4675 4675 + request: impl tonic::IntoRequest<super::ListClientsRequest>, 4676 4676 + ) -> std::result::Result< 4677 4677 + tonic::Response<super::ListClientsResponse>, 4678 4678 + tonic::Status, 4679 4679 + > { 4680 4680 + self.inner 4681 4681 + .ready() 4682 4682 + .await 4683 4683 + .map_err(|e| { 4684 4684 + tonic::Status::unknown( 4685 4685 + format!("Service was not ready: {}", e.into()), 4686 4686 + ) 4687 4687 + })?; 4688 4688 + let codec = tonic::codec::ProstCodec::default(); 4689 4689 + let path = http::uri::PathAndQuery::from_static( 4690 4690 + "/zerod.v1alpha1.SnapcastService/ListClients", 4691 4691 + ); 4692 4692 + let mut req = request.into_request(); 4693 4693 + req.extensions_mut() 4694 4694 + .insert( 4695 4695 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "ListClients"), 4696 4696 + ); 4697 4697 + self.inner.unary(req, path, codec).await 4698 4698 + } 4699 4699 + pub async fn list_snap_streams( 4700 4700 + &mut self, 4701 4701 + request: impl tonic::IntoRequest<super::ListSnapStreamsRequest>, 4702 4702 + ) -> std::result::Result< 4703 4703 + tonic::Response<super::ListSnapStreamsResponse>, 4704 4704 + tonic::Status, 4705 4705 + > { 4706 4706 + self.inner 4707 4707 + .ready() 4708 4708 + .await 4709 4709 + .map_err(|e| { 4710 4710 + tonic::Status::unknown( 4711 4711 + format!("Service was not ready: {}", e.into()), 4712 4712 + ) 4713 4713 + })?; 4714 4714 + let codec = tonic::codec::ProstCodec::default(); 4715 4715 + let path = http::uri::PathAndQuery::from_static( 4716 4716 + "/zerod.v1alpha1.SnapcastService/ListSnapStreams", 4717 4717 + ); 4718 4718 + let mut req = request.into_request(); 4719 4719 + req.extensions_mut() 4720 4720 + .insert( 4721 4721 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "ListSnapStreams"), 4722 4722 + ); 4723 4723 + self.inner.unary(req, path, codec).await 4724 4724 + } 4725 4725 + pub async fn set_client_volume( 4726 4726 + &mut self, 4727 4727 + request: impl tonic::IntoRequest<super::SetClientVolumeRequest>, 4728 4728 + ) -> std::result::Result< 4729 4729 + tonic::Response<super::SetClientVolumeResponse>, 4730 4730 + tonic::Status, 4731 4731 + > { 4732 4732 + self.inner 4733 4733 + .ready() 4734 4734 + .await 4735 4735 + .map_err(|e| { 4736 4736 + tonic::Status::unknown( 4737 4737 + format!("Service was not ready: {}", e.into()), 4738 4738 + ) 4739 4739 + })?; 4740 4740 + let codec = tonic::codec::ProstCodec::default(); 4741 4741 + let path = http::uri::PathAndQuery::from_static( 4742 4742 + "/zerod.v1alpha1.SnapcastService/SetClientVolume", 4743 4743 + ); 4744 4744 + let mut req = request.into_request(); 4745 4745 + req.extensions_mut() 4746 4746 + .insert( 4747 4747 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "SetClientVolume"), 4748 4748 + ); 4749 4749 + self.inner.unary(req, path, codec).await 4750 4750 + } 4751 4751 + pub async fn set_client_latency( 4752 4752 + &mut self, 4753 4753 + request: impl tonic::IntoRequest<super::SetClientLatencyRequest>, 4754 4754 + ) -> std::result::Result< 4755 4755 + tonic::Response<super::SetClientLatencyResponse>, 4756 4756 + tonic::Status, 4757 4757 + > { 4758 4758 + self.inner 4759 4759 + .ready() 4760 4760 + .await 4761 4761 + .map_err(|e| { 4762 4762 + tonic::Status::unknown( 4763 4763 + format!("Service was not ready: {}", e.into()), 4764 4764 + ) 4765 4765 + })?; 4766 4766 + let codec = tonic::codec::ProstCodec::default(); 4767 4767 + let path = http::uri::PathAndQuery::from_static( 4768 4768 + "/zerod.v1alpha1.SnapcastService/SetClientLatency", 4769 4769 + ); 4770 4770 + let mut req = request.into_request(); 4771 4771 + req.extensions_mut() 4772 4772 + .insert( 4773 4773 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "SetClientLatency"), 4774 4774 + ); 4775 4775 + self.inner.unary(req, path, codec).await 4776 4776 + } 4777 4777 + pub async fn set_client_name( 4778 4778 + &mut self, 4779 4779 + request: impl tonic::IntoRequest<super::SetClientNameRequest>, 4780 4780 + ) -> std::result::Result< 4781 4781 + tonic::Response<super::SetClientNameResponse>, 4782 4782 + tonic::Status, 4783 4783 + > { 4784 4784 + self.inner 4785 4785 + .ready() 4786 4786 + .await 4787 4787 + .map_err(|e| { 4788 4788 + tonic::Status::unknown( 4789 4789 + format!("Service was not ready: {}", e.into()), 4790 4790 + ) 4791 4791 + })?; 4792 4792 + let codec = tonic::codec::ProstCodec::default(); 4793 4793 + let path = http::uri::PathAndQuery::from_static( 4794 4794 + "/zerod.v1alpha1.SnapcastService/SetClientName", 4795 4795 + ); 4796 4796 + let mut req = request.into_request(); 4797 4797 + req.extensions_mut() 4798 4798 + .insert( 4799 4799 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "SetClientName"), 4800 4800 + ); 4801 4801 + self.inner.unary(req, path, codec).await 4802 4802 + } 4803 4803 + pub async fn set_group_stream( 4804 4804 + &mut self, 4805 4805 + request: impl tonic::IntoRequest<super::SetGroupStreamRequest>, 4806 4806 + ) -> std::result::Result< 4807 4807 + tonic::Response<super::SetGroupStreamResponse>, 4808 4808 + tonic::Status, 4809 4809 + > { 4810 4810 + self.inner 4811 4811 + .ready() 4812 4812 + .await 4813 4813 + .map_err(|e| { 4814 4814 + tonic::Status::unknown( 4815 4815 + format!("Service was not ready: {}", e.into()), 4816 4816 + ) 4817 4817 + })?; 4818 4818 + let codec = tonic::codec::ProstCodec::default(); 4819 4819 + let path = http::uri::PathAndQuery::from_static( 4820 4820 + "/zerod.v1alpha1.SnapcastService/SetGroupStream", 4821 4821 + ); 4822 4822 + let mut req = request.into_request(); 4823 4823 + req.extensions_mut() 4824 4824 + .insert( 4825 4825 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "SetGroupStream"), 4826 4826 + ); 4827 4827 + self.inner.unary(req, path, codec).await 4828 4828 + } 4829 4829 + pub async fn set_group_mute( 4830 4830 + &mut self, 4831 4831 + request: impl tonic::IntoRequest<super::SetGroupMuteRequest>, 4832 4832 + ) -> std::result::Result< 4833 4833 + tonic::Response<super::SetGroupMuteResponse>, 4834 4834 + tonic::Status, 4835 4835 + > { 4836 4836 + self.inner 4837 4837 + .ready() 4838 4838 + .await 4839 4839 + .map_err(|e| { 4840 4840 + tonic::Status::unknown( 4841 4841 + format!("Service was not ready: {}", e.into()), 4842 4842 + ) 4843 4843 + })?; 4844 4844 + let codec = tonic::codec::ProstCodec::default(); 4845 4845 + let path = http::uri::PathAndQuery::from_static( 4846 4846 + "/zerod.v1alpha1.SnapcastService/SetGroupMute", 4847 4847 + ); 4848 4848 + let mut req = request.into_request(); 4849 4849 + req.extensions_mut() 4850 4850 + .insert( 4851 4851 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "SetGroupMute"), 4852 4852 + ); 4853 4853 + self.inner.unary(req, path, codec).await 4854 4854 + } 4855 4855 + pub async fn set_group_clients( 4856 4856 + &mut self, 4857 4857 + request: impl tonic::IntoRequest<super::SetGroupClientsRequest>, 4858 4858 + ) -> std::result::Result< 4859 4859 + tonic::Response<super::SetGroupClientsResponse>, 4860 4860 + tonic::Status, 4861 4861 + > { 4862 4862 + self.inner 4863 4863 + .ready() 4864 4864 + .await 4865 4865 + .map_err(|e| { 4866 4866 + tonic::Status::unknown( 4867 4867 + format!("Service was not ready: {}", e.into()), 4868 4868 + ) 4869 4869 + })?; 4870 4870 + let codec = tonic::codec::ProstCodec::default(); 4871 4871 + let path = http::uri::PathAndQuery::from_static( 4872 4872 + "/zerod.v1alpha1.SnapcastService/SetGroupClients", 4873 4873 + ); 4874 4874 + let mut req = request.into_request(); 4875 4875 + req.extensions_mut() 4876 4876 + .insert( 4877 4877 + GrpcMethod::new("zerod.v1alpha1.SnapcastService", "SetGroupClients"), 4878 4878 + ); 4879 4879 + self.inner.unary(req, path, codec).await 4880 4880 + } 4881 4881 + } 4882 4882 + } 4883 4883 + /// Generated server implementations. 4884 4884 + pub mod snapcast_service_server { 4885 4885 + #![allow( 4886 4886 + unused_variables, 4887 4887 + dead_code, 4888 4888 + missing_docs, 4889 4889 + clippy::wildcard_imports, 4890 4890 + clippy::let_unit_value, 4891 4891 + )] 4892 4892 + use tonic::codegen::*; 4893 4893 + /// Generated trait containing gRPC methods that should be implemented for use with SnapcastServiceServer. 4894 4894 + #[async_trait] 4895 4895 + pub trait SnapcastService: std::marker::Send + std::marker::Sync + 'static { 4896 4896 + async fn get_server_status( 4897 4897 + &self, 4898 4898 + request: tonic::Request<super::GetServerStatusRequest>, 4899 4899 + ) -> std::result::Result< 4900 4900 + tonic::Response<super::GetServerStatusResponse>, 4901 4901 + tonic::Status, 4902 4902 + >; 4903 4903 + async fn list_clients( 4904 4904 + &self, 4905 4905 + request: tonic::Request<super::ListClientsRequest>, 4906 4906 + ) -> std::result::Result< 4907 4907 + tonic::Response<super::ListClientsResponse>, 4908 4908 + tonic::Status, 4909 4909 + >; 4910 4910 + async fn list_snap_streams( 4911 4911 + &self, 4912 4912 + request: tonic::Request<super::ListSnapStreamsRequest>, 4913 4913 + ) -> std::result::Result< 4914 4914 + tonic::Response<super::ListSnapStreamsResponse>, 4915 4915 + tonic::Status, 4916 4916 + >; 4917 4917 + async fn set_client_volume( 4918 4918 + &self, 4919 4919 + request: tonic::Request<super::SetClientVolumeRequest>, 4920 4920 + ) -> std::result::Result< 4921 4921 + tonic::Response<super::SetClientVolumeResponse>, 4922 4922 + tonic::Status, 4923 4923 + >; 4924 4924 + async fn set_client_latency( 4925 4925 + &self, 4926 4926 + request: tonic::Request<super::SetClientLatencyRequest>, 4927 4927 + ) -> std::result::Result< 4928 4928 + tonic::Response<super::SetClientLatencyResponse>, 4929 4929 + tonic::Status, 4930 4930 + >; 4931 4931 + async fn set_client_name( 4932 4932 + &self, 4933 4933 + request: tonic::Request<super::SetClientNameRequest>, 4934 4934 + ) -> std::result::Result< 4935 4935 + tonic::Response<super::SetClientNameResponse>, 4936 4936 + tonic::Status, 4937 4937 + >; 4938 4938 + async fn set_group_stream( 4939 4939 + &self, 4940 4940 + request: tonic::Request<super::SetGroupStreamRequest>, 4941 4941 + ) -> std::result::Result< 4942 4942 + tonic::Response<super::SetGroupStreamResponse>, 4943 4943 + tonic::Status, 4944 4944 + >; 4945 4945 + async fn set_group_mute( 4946 4946 + &self, 4947 4947 + request: tonic::Request<super::SetGroupMuteRequest>, 4948 4948 + ) -> std::result::Result< 4949 4949 + tonic::Response<super::SetGroupMuteResponse>, 4950 4950 + tonic::Status, 4951 4951 + >; 4952 4952 + async fn set_group_clients( 4953 4953 + &self, 4954 4954 + request: tonic::Request<super::SetGroupClientsRequest>, 4955 4955 + ) -> std::result::Result< 4956 4956 + tonic::Response<super::SetGroupClientsResponse>, 4957 4957 + tonic::Status, 4958 4958 + >; 4959 4959 + } 4960 4960 + #[derive(Debug)] 4961 4961 + pub struct SnapcastServiceServer<T> { 4962 4962 + inner: Arc<T>, 4963 4963 + accept_compression_encodings: EnabledCompressionEncodings, 4964 4964 + send_compression_encodings: EnabledCompressionEncodings, 4965 4965 + max_decoding_message_size: Option<usize>, 4966 4966 + max_encoding_message_size: Option<usize>, 4967 4967 + } 4968 4968 + impl<T> SnapcastServiceServer<T> { 4969 4969 + pub fn new(inner: T) -> Self { 4970 4970 + Self::from_arc(Arc::new(inner)) 4971 4971 + } 4972 4972 + pub fn from_arc(inner: Arc<T>) -> Self { 4973 4973 + Self { 4974 4974 + inner, 4975 4975 + accept_compression_encodings: Default::default(), 4976 4976 + send_compression_encodings: Default::default(), 4977 4977 + max_decoding_message_size: None, 4978 4978 + max_encoding_message_size: None, 4979 4979 + } 4980 4980 + } 4981 4981 + pub fn with_interceptor<F>( 4982 4982 + inner: T, 4983 4983 + interceptor: F, 4984 4984 + ) -> InterceptedService<Self, F> 4985 4985 + where 4986 4986 + F: tonic::service::Interceptor, 4987 4987 + { 4988 4988 + InterceptedService::new(Self::new(inner), interceptor) 4989 4989 + } 4990 4990 + /// Enable decompressing requests with the given encoding. 4991 4991 + #[must_use] 4992 4992 + pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { 4993 4993 + self.accept_compression_encodings.enable(encoding); 4994 4994 + self 4995 4995 + } 4996 4996 + /// Compress responses with the given encoding, if the client supports it. 4997 4997 + #[must_use] 4998 4998 + pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { 4999 4999 + self.send_compression_encodings.enable(encoding); 5000 5000 + self 5001 5001 + } 5002 5002 + /// Limits the maximum size of a decoded message. 5003 5003 + /// 5004 5004 + /// Default: `4MB` 5005 5005 + #[must_use] 5006 5006 + pub fn max_decoding_message_size(mut self, limit: usize) -> Self { 5007 5007 + self.max_decoding_message_size = Some(limit); 5008 5008 + self 5009 5009 + } 5010 5010 + /// Limits the maximum size of an encoded message. 5011 5011 + /// 5012 5012 + /// Default: `usize::MAX` 5013 5013 + #[must_use] 5014 5014 + pub fn max_encoding_message_size(mut self, limit: usize) -> Self { 5015 5015 + self.max_encoding_message_size = Some(limit); 5016 5016 + self 5017 5017 + } 5018 5018 + } 5019 5019 + impl<T, B> tonic::codegen::Service<http::Request<B>> for SnapcastServiceServer<T> 5020 5020 + where 5021 5021 + T: SnapcastService, 5022 5022 + B: Body + std::marker::Send + 'static, 5023 5023 + B::Error: Into<StdError> + std::marker::Send + 'static, 5024 5024 + { 5025 5025 + type Response = http::Response<tonic::body::BoxBody>; 5026 5026 + type Error = std::convert::Infallible; 5027 5027 + type Future = BoxFuture<Self::Response, Self::Error>; 5028 5028 + fn poll_ready( 5029 5029 + &mut self, 5030 5030 + _cx: &mut Context<'_>, 5031 5031 + ) -> Poll<std::result::Result<(), Self::Error>> { 5032 5032 + Poll::Ready(Ok(())) 5033 5033 + } 5034 5034 + fn call(&mut self, req: http::Request<B>) -> Self::Future { 5035 5035 + match req.uri().path() { 5036 5036 + "/zerod.v1alpha1.SnapcastService/GetServerStatus" => { 5037 5037 + #[allow(non_camel_case_types)] 5038 5038 + struct GetServerStatusSvc<T: SnapcastService>(pub Arc<T>); 5039 5039 + impl< 5040 5040 + T: SnapcastService, 5041 5041 + > tonic::server::UnaryService<super::GetServerStatusRequest> 5042 5042 + for GetServerStatusSvc<T> { 5043 5043 + type Response = super::GetServerStatusResponse; 5044 5044 + type Future = BoxFuture< 5045 5045 + tonic::Response<Self::Response>, 5046 5046 + tonic::Status, 5047 5047 + >; 5048 5048 + fn call( 5049 5049 + &mut self, 5050 5050 + request: tonic::Request<super::GetServerStatusRequest>, 5051 5051 + ) -> Self::Future { 5052 5052 + let inner = Arc::clone(&self.0); 5053 5053 + let fut = async move { 5054 5054 + <T as SnapcastService>::get_server_status(&inner, request) 5055 5055 + .await 5056 5056 + }; 5057 5057 + Box::pin(fut) 5058 5058 + } 5059 5059 + } 5060 5060 + let accept_compression_encodings = self.accept_compression_encodings; 5061 5061 + let send_compression_encodings = self.send_compression_encodings; 5062 5062 + let max_decoding_message_size = self.max_decoding_message_size; 5063 5063 + let max_encoding_message_size = self.max_encoding_message_size; 5064 5064 + let inner = self.inner.clone(); 5065 5065 + let fut = async move { 5066 5066 + let method = GetServerStatusSvc(inner); 5067 5067 + let codec = tonic::codec::ProstCodec::default(); 5068 5068 + let mut grpc = tonic::server::Grpc::new(codec) 5069 5069 + .apply_compression_config( 5070 5070 + accept_compression_encodings, 5071 5071 + send_compression_encodings, 5072 5072 + ) 5073 5073 + .apply_max_message_size_config( 5074 5074 + max_decoding_message_size, 5075 5075 + max_encoding_message_size, 5076 5076 + ); 5077 5077 + let res = grpc.unary(method, req).await; 5078 5078 + Ok(res) 5079 5079 + }; 5080 5080 + Box::pin(fut) 5081 5081 + } 5082 5082 + "/zerod.v1alpha1.SnapcastService/ListClients" => { 5083 5083 + #[allow(non_camel_case_types)] 5084 5084 + struct ListClientsSvc<T: SnapcastService>(pub Arc<T>); 5085 5085 + impl< 5086 5086 + T: SnapcastService, 5087 5087 + > tonic::server::UnaryService<super::ListClientsRequest> 5088 5088 + for ListClientsSvc<T> { 5089 5089 + type Response = super::ListClientsResponse; 5090 5090 + type Future = BoxFuture< 5091 5091 + tonic::Response<Self::Response>, 5092 5092 + tonic::Status, 5093 5093 + >; 5094 5094 + fn call( 5095 5095 + &mut self, 5096 5096 + request: tonic::Request<super::ListClientsRequest>, 5097 5097 + ) -> Self::Future { 5098 5098 + let inner = Arc::clone(&self.0); 5099 5099 + let fut = async move { 5100 5100 + <T as SnapcastService>::list_clients(&inner, request).await 5101 5101 + }; 5102 5102 + Box::pin(fut) 5103 5103 + } 5104 5104 + } 5105 5105 + let accept_compression_encodings = self.accept_compression_encodings; 5106 5106 + let send_compression_encodings = self.send_compression_encodings; 5107 5107 + let max_decoding_message_size = self.max_decoding_message_size; 5108 5108 + let max_encoding_message_size = self.max_encoding_message_size; 5109 5109 + let inner = self.inner.clone(); 5110 5110 + let fut = async move { 5111 5111 + let method = ListClientsSvc(inner); 5112 5112 + let codec = tonic::codec::ProstCodec::default(); 5113 5113 + let mut grpc = tonic::server::Grpc::new(codec) 5114 5114 + .apply_compression_config( 5115 5115 + accept_compression_encodings, 5116 5116 + send_compression_encodings, 5117 5117 + ) 5118 5118 + .apply_max_message_size_config( 5119 5119 + max_decoding_message_size, 5120 5120 + max_encoding_message_size, 5121 5121 + ); 5122 5122 + let res = grpc.unary(method, req).await; 5123 5123 + Ok(res) 5124 5124 + }; 5125 5125 + Box::pin(fut) 5126 5126 + } 5127 5127 + "/zerod.v1alpha1.SnapcastService/ListSnapStreams" => { 5128 5128 + #[allow(non_camel_case_types)] 5129 5129 + struct ListSnapStreamsSvc<T: SnapcastService>(pub Arc<T>); 5130 5130 + impl< 5131 5131 + T: SnapcastService, 5132 5132 + > tonic::server::UnaryService<super::ListSnapStreamsRequest> 5133 5133 + for ListSnapStreamsSvc<T> { 5134 5134 + type Response = super::ListSnapStreamsResponse; 5135 5135 + type Future = BoxFuture< 5136 5136 + tonic::Response<Self::Response>, 5137 5137 + tonic::Status, 5138 5138 + >; 5139 5139 + fn call( 5140 5140 + &mut self, 5141 5141 + request: tonic::Request<super::ListSnapStreamsRequest>, 5142 5142 + ) -> Self::Future { 5143 5143 + let inner = Arc::clone(&self.0); 5144 5144 + let fut = async move { 5145 5145 + <T as SnapcastService>::list_snap_streams(&inner, request) 5146 5146 + .await 5147 5147 + }; 5148 5148 + Box::pin(fut) 5149 5149 + } 5150 5150 + } 5151 5151 + let accept_compression_encodings = self.accept_compression_encodings; 5152 5152 + let send_compression_encodings = self.send_compression_encodings; 5153 5153 + let max_decoding_message_size = self.max_decoding_message_size; 5154 5154 + let max_encoding_message_size = self.max_encoding_message_size; 5155 5155 + let inner = self.inner.clone(); 5156 5156 + let fut = async move { 5157 5157 + let method = ListSnapStreamsSvc(inner); 5158 5158 + let codec = tonic::codec::ProstCodec::default(); 5159 5159 + let mut grpc = tonic::server::Grpc::new(codec) 5160 5160 + .apply_compression_config( 5161 5161 + accept_compression_encodings, 5162 5162 + send_compression_encodings, 5163 5163 + ) 5164 5164 + .apply_max_message_size_config( 5165 5165 + max_decoding_message_size, 5166 5166 + max_encoding_message_size, 5167 5167 + ); 5168 5168 + let res = grpc.unary(method, req).await; 5169 5169 + Ok(res) 5170 5170 + }; 5171 5171 + Box::pin(fut) 5172 5172 + } 5173 5173 + "/zerod.v1alpha1.SnapcastService/SetClientVolume" => { 5174 5174 + #[allow(non_camel_case_types)] 5175 5175 + struct SetClientVolumeSvc<T: SnapcastService>(pub Arc<T>); 5176 5176 + impl< 5177 5177 + T: SnapcastService, 5178 5178 + > tonic::server::UnaryService<super::SetClientVolumeRequest> 5179 5179 + for SetClientVolumeSvc<T> { 5180 5180 + type Response = super::SetClientVolumeResponse; 5181 5181 + type Future = BoxFuture< 5182 5182 + tonic::Response<Self::Response>, 5183 5183 + tonic::Status, 5184 5184 + >; 5185 5185 + fn call( 5186 5186 + &mut self, 5187 5187 + request: tonic::Request<super::SetClientVolumeRequest>, 5188 5188 + ) -> Self::Future { 5189 5189 + let inner = Arc::clone(&self.0); 5190 5190 + let fut = async move { 5191 5191 + <T as SnapcastService>::set_client_volume(&inner, request) 5192 5192 + .await 5193 5193 + }; 5194 5194 + Box::pin(fut) 5195 5195 + } 5196 5196 + } 5197 5197 + let accept_compression_encodings = self.accept_compression_encodings; 5198 5198 + let send_compression_encodings = self.send_compression_encodings; 5199 5199 + let max_decoding_message_size = self.max_decoding_message_size; 5200 5200 + let max_encoding_message_size = self.max_encoding_message_size; 5201 5201 + let inner = self.inner.clone(); 5202 5202 + let fut = async move { 5203 5203 + let method = SetClientVolumeSvc(inner); 5204 5204 + let codec = tonic::codec::ProstCodec::default(); 5205 5205 + let mut grpc = tonic::server::Grpc::new(codec) 5206 5206 + .apply_compression_config( 5207 5207 + accept_compression_encodings, 5208 5208 + send_compression_encodings, 5209 5209 + ) 5210 5210 + .apply_max_message_size_config( 5211 5211 + max_decoding_message_size, 5212 5212 + max_encoding_message_size, 5213 5213 + ); 5214 5214 + let res = grpc.unary(method, req).await; 5215 5215 + Ok(res) 5216 5216 + }; 5217 5217 + Box::pin(fut) 5218 5218 + } 5219 5219 + "/zerod.v1alpha1.SnapcastService/SetClientLatency" => { 5220 5220 + #[allow(non_camel_case_types)] 5221 5221 + struct SetClientLatencySvc<T: SnapcastService>(pub Arc<T>); 5222 5222 + impl< 5223 5223 + T: SnapcastService, 5224 5224 + > tonic::server::UnaryService<super::SetClientLatencyRequest> 5225 5225 + for SetClientLatencySvc<T> { 5226 5226 + type Response = super::SetClientLatencyResponse; 5227 5227 + type Future = BoxFuture< 5228 5228 + tonic::Response<Self::Response>, 5229 5229 + tonic::Status, 5230 5230 + >; 5231 5231 + fn call( 5232 5232 + &mut self, 5233 5233 + request: tonic::Request<super::SetClientLatencyRequest>, 5234 5234 + ) -> Self::Future { 5235 5235 + let inner = Arc::clone(&self.0); 5236 5236 + let fut = async move { 5237 5237 + <T as SnapcastService>::set_client_latency(&inner, request) 5238 5238 + .await 5239 5239 + }; 5240 5240 + Box::pin(fut) 5241 5241 + } 5242 5242 + } 5243 5243 + let accept_compression_encodings = self.accept_compression_encodings; 5244 5244 + let send_compression_encodings = self.send_compression_encodings; 5245 5245 + let max_decoding_message_size = self.max_decoding_message_size; 5246 5246 + let max_encoding_message_size = self.max_encoding_message_size; 5247 5247 + let inner = self.inner.clone(); 5248 5248 + let fut = async move { 5249 5249 + let method = SetClientLatencySvc(inner); 5250 5250 + let codec = tonic::codec::ProstCodec::default(); 5251 5251 + let mut grpc = tonic::server::Grpc::new(codec) 5252 5252 + .apply_compression_config( 5253 5253 + accept_compression_encodings, 5254 5254 + send_compression_encodings, 5255 5255 + ) 5256 5256 + .apply_max_message_size_config( 5257 5257 + max_decoding_message_size, 5258 5258 + max_encoding_message_size, 5259 5259 + ); 5260 5260 + let res = grpc.unary(method, req).await; 5261 5261 + Ok(res) 5262 5262 + }; 5263 5263 + Box::pin(fut) 5264 5264 + } 5265 5265 + "/zerod.v1alpha1.SnapcastService/SetClientName" => { 5266 5266 + #[allow(non_camel_case_types)] 5267 5267 + struct SetClientNameSvc<T: SnapcastService>(pub Arc<T>); 5268 5268 + impl< 5269 5269 + T: SnapcastService, 5270 5270 + > tonic::server::UnaryService<super::SetClientNameRequest> 5271 5271 + for SetClientNameSvc<T> { 5272 5272 + type Response = super::SetClientNameResponse; 5273 5273 + type Future = BoxFuture< 5274 5274 + tonic::Response<Self::Response>, 5275 5275 + tonic::Status, 5276 5276 + >; 5277 5277 + fn call( 5278 5278 + &mut self, 5279 5279 + request: tonic::Request<super::SetClientNameRequest>, 5280 5280 + ) -> Self::Future { 5281 5281 + let inner = Arc::clone(&self.0); 5282 5282 + let fut = async move { 5283 5283 + <T as SnapcastService>::set_client_name(&inner, request) 5284 5284 + .await 5285 5285 + }; 5286 5286 + Box::pin(fut) 5287 5287 + } 5288 5288 + } 5289 5289 + let accept_compression_encodings = self.accept_compression_encodings; 5290 5290 + let send_compression_encodings = self.send_compression_encodings; 5291 5291 + let max_decoding_message_size = self.max_decoding_message_size; 5292 5292 + let max_encoding_message_size = self.max_encoding_message_size; 5293 5293 + let inner = self.inner.clone(); 5294 5294 + let fut = async move { 5295 5295 + let method = SetClientNameSvc(inner); 5296 5296 + let codec = tonic::codec::ProstCodec::default(); 5297 5297 + let mut grpc = tonic::server::Grpc::new(codec) 5298 5298 + .apply_compression_config( 5299 5299 + accept_compression_encodings, 5300 5300 + send_compression_encodings, 5301 5301 + ) 5302 5302 + .apply_max_message_size_config( 5303 5303 + max_decoding_message_size, 5304 5304 + max_encoding_message_size, 5305 5305 + ); 5306 5306 + let res = grpc.unary(method, req).await; 5307 5307 + Ok(res) 5308 5308 + }; 5309 5309 + Box::pin(fut) 5310 5310 + } 5311 5311 + "/zerod.v1alpha1.SnapcastService/SetGroupStream" => { 5312 5312 + #[allow(non_camel_case_types)] 5313 5313 + struct SetGroupStreamSvc<T: SnapcastService>(pub Arc<T>); 5314 5314 + impl< 5315 5315 + T: SnapcastService, 5316 5316 + > tonic::server::UnaryService<super::SetGroupStreamRequest> 5317 5317 + for SetGroupStreamSvc<T> { 5318 5318 + type Response = super::SetGroupStreamResponse; 5319 5319 + type Future = BoxFuture< 5320 5320 + tonic::Response<Self::Response>, 5321 5321 + tonic::Status, 5322 5322 + >; 5323 5323 + fn call( 5324 5324 + &mut self, 5325 5325 + request: tonic::Request<super::SetGroupStreamRequest>, 5326 5326 + ) -> Self::Future { 5327 5327 + let inner = Arc::clone(&self.0); 5328 5328 + let fut = async move { 5329 5329 + <T as SnapcastService>::set_group_stream(&inner, request) 5330 5330 + .await 5331 5331 + }; 5332 5332 + Box::pin(fut) 5333 5333 + } 5334 5334 + } 5335 5335 + let accept_compression_encodings = self.accept_compression_encodings; 5336 5336 + let send_compression_encodings = self.send_compression_encodings; 5337 5337 + let max_decoding_message_size = self.max_decoding_message_size; 5338 5338 + let max_encoding_message_size = self.max_encoding_message_size; 5339 5339 + let inner = self.inner.clone(); 5340 5340 + let fut = async move { 5341 5341 + let method = SetGroupStreamSvc(inner); 5342 5342 + let codec = tonic::codec::ProstCodec::default(); 5343 5343 + let mut grpc = tonic::server::Grpc::new(codec) 5344 5344 + .apply_compression_config( 5345 5345 + accept_compression_encodings, 5346 5346 + send_compression_encodings, 5347 5347 + ) 5348 5348 + .apply_max_message_size_config( 5349 5349 + max_decoding_message_size, 5350 5350 + max_encoding_message_size, 5351 5351 + ); 5352 5352 + let res = grpc.unary(method, req).await; 5353 5353 + Ok(res) 5354 5354 + }; 5355 5355 + Box::pin(fut) 5356 5356 + } 5357 5357 + "/zerod.v1alpha1.SnapcastService/SetGroupMute" => { 5358 5358 + #[allow(non_camel_case_types)] 5359 5359 + struct SetGroupMuteSvc<T: SnapcastService>(pub Arc<T>); 5360 5360 + impl< 5361 5361 + T: SnapcastService, 5362 5362 + > tonic::server::UnaryService<super::SetGroupMuteRequest> 5363 5363 + for SetGroupMuteSvc<T> { 5364 5364 + type Response = super::SetGroupMuteResponse; 5365 5365 + type Future = BoxFuture< 5366 5366 + tonic::Response<Self::Response>, 5367 5367 + tonic::Status, 5368 5368 + >; 5369 5369 + fn call( 5370 5370 + &mut self, 5371 5371 + request: tonic::Request<super::SetGroupMuteRequest>, 5372 5372 + ) -> Self::Future { 5373 5373 + let inner = Arc::clone(&self.0); 5374 5374 + let fut = async move { 5375 5375 + <T as SnapcastService>::set_group_mute(&inner, request) 5376 5376 + .await 5377 5377 + }; 5378 5378 + Box::pin(fut) 5379 5379 + } 5380 5380 + } 5381 5381 + let accept_compression_encodings = self.accept_compression_encodings; 5382 5382 + let send_compression_encodings = self.send_compression_encodings; 5383 5383 + let max_decoding_message_size = self.max_decoding_message_size; 5384 5384 + let max_encoding_message_size = self.max_encoding_message_size; 5385 5385 + let inner = self.inner.clone(); 5386 5386 + let fut = async move { 5387 5387 + let method = SetGroupMuteSvc(inner); 5388 5388 + let codec = tonic::codec::ProstCodec::default(); 5389 5389 + let mut grpc = tonic::server::Grpc::new(codec) 5390 5390 + .apply_compression_config( 5391 5391 + accept_compression_encodings, 5392 5392 + send_compression_encodings, 5393 5393 + ) 5394 5394 + .apply_max_message_size_config( 5395 5395 + max_decoding_message_size, 5396 5396 + max_encoding_message_size, 5397 5397 + ); 5398 5398 + let res = grpc.unary(method, req).await; 5399 5399 + Ok(res) 5400 5400 + }; 5401 5401 + Box::pin(fut) 5402 5402 + } 5403 5403 + "/zerod.v1alpha1.SnapcastService/SetGroupClients" => { 5404 5404 + #[allow(non_camel_case_types)] 5405 5405 + struct SetGroupClientsSvc<T: SnapcastService>(pub Arc<T>); 5406 5406 + impl< 5407 5407 + T: SnapcastService, 5408 5408 + > tonic::server::UnaryService<super::SetGroupClientsRequest> 5409 5409 + for SetGroupClientsSvc<T> { 5410 5410 + type Response = super::SetGroupClientsResponse; 5411 5411 + type Future = BoxFuture< 5412 5412 + tonic::Response<Self::Response>, 5413 5413 + tonic::Status, 5414 5414 + >; 5415 5415 + fn call( 5416 5416 + &mut self, 5417 5417 + request: tonic::Request<super::SetGroupClientsRequest>, 5418 5418 + ) -> Self::Future { 5419 5419 + let inner = Arc::clone(&self.0); 5420 5420 + let fut = async move { 5421 5421 + <T as SnapcastService>::set_group_clients(&inner, request) 5422 5422 + .await 5423 5423 + }; 5424 5424 + Box::pin(fut) 5425 5425 + } 5426 5426 + } 5427 5427 + let accept_compression_encodings = self.accept_compression_encodings; 5428 5428 + let send_compression_encodings = self.send_compression_encodings; 5429 5429 + let max_decoding_message_size = self.max_decoding_message_size; 5430 5430 + let max_encoding_message_size = self.max_encoding_message_size; 5431 5431 + let inner = self.inner.clone(); 5432 5432 + let fut = async move { 5433 5433 + let method = SetGroupClientsSvc(inner); 5434 5434 + let codec = tonic::codec::ProstCodec::default(); 5435 5435 + let mut grpc = tonic::server::Grpc::new(codec) 5436 5436 + .apply_compression_config( 5437 5437 + accept_compression_encodings, 5438 5438 + send_compression_encodings, 5439 5439 + ) 5440 5440 + .apply_max_message_size_config( 5441 5441 + max_decoding_message_size, 5442 5442 + max_encoding_message_size, 5443 5443 + ); 5444 5444 + let res = grpc.unary(method, req).await; 5445 5445 + Ok(res) 5446 5446 + }; 5447 5447 + Box::pin(fut) 5448 5448 + } 5449 5449 + _ => { 5450 5450 + Box::pin(async move { 5451 5451 + let mut response = http::Response::new(empty_body()); 5452 5452 + let headers = response.headers_mut(); 5453 5453 + headers 5454 5454 + .insert( 5455 5455 + tonic::Status::GRPC_STATUS, 5456 5456 + (tonic::Code::Unimplemented as i32).into(), 5457 5457 + ); 5458 5458 + headers 5459 5459 + .insert( 5460 5460 + http::header::CONTENT_TYPE, 5461 5461 + tonic::metadata::GRPC_CONTENT_TYPE, 5462 5462 + ); 5463 5463 + Ok(response) 5464 5464 + }) 5465 5465 + } 5466 5466 + } 5467 5467 + } 5468 5468 + } 5469 5469 + impl<T> Clone for SnapcastServiceServer<T> { 5470 5470 + fn clone(&self) -> Self { 5471 5471 + let inner = self.inner.clone(); 5472 5472 + Self { 5473 5473 + inner, 5474 5474 + accept_compression_encodings: self.accept_compression_encodings, 5475 5475 + send_compression_encodings: self.send_compression_encodings, 5476 5476 + max_decoding_message_size: self.max_decoding_message_size, 5477 5477 + max_encoding_message_size: self.max_encoding_message_size, 5478 5478 + } 5479 5479 + } 5480 5480 + } 5481 5481 + /// Generated gRPC service name 5482 5482 + pub const SERVICE_NAME: &str = "zerod.v1alpha1.SnapcastService"; 5483 5483 + impl<T> tonic::server::NamedService for SnapcastServiceServer<T> { 5484 5484 + const NAME: &'static str = SERVICE_NAME; 5485 5485 + } 5486 5486 + }

crates/proto/src/generated/zerod_descriptor.bin

View file Reviewed

This is a binary file and will not be displayed.

+1

crates/server/Cargo.toml

View file Reviewed

··· 22 22 zerod-discovery = { path = "../discovery" } 23 23 zerod-events = { path = "../events" } 24 24 zerod-proto = { path = "../proto" } 25 25 + zerod-snapcast = { path = "../snapcast" } 25 26 zerod-stream = { path = "../stream" } 26 27 zerod-systemd = { path = "../systemd" } 27 28 zerod-volume = { path = "../volume" }

+29

crates/server/src/lib.rs

View file Reviewed

··· 4 4 mod config; 5 5 mod events; 6 6 mod settings; 7 7 + mod snapcast; 7 8 mod stream; 8 9 mod system; 9 10 mod systemd; ··· 18 19 use zerod_proto::v1alpha1::{ 19 20 bluetooth_service_server::BluetoothServiceServer, 20 21 config_service_server::ConfigServiceServer, events_service_server::EventsServiceServer, 22 22 + snapcast_service_server::SnapcastServiceServer, 21 23 stream_service_server::StreamServiceServer, system_service_server::SystemServiceServer, 22 24 systemd_service_server::SystemdServiceServer, volume_service_server::VolumeServiceServer, 23 25 }; ··· 32 34 let allow = Arc::new(settings.systemd_allowlist()); 33 35 let registry = Arc::new(settings.config_registry()); 34 36 let bearer = resolve_bearer_token(&settings.server.bearer_token)?; 37 37 + let snap_svc = build_snapcast_svc(&settings.snapcast); 35 38 36 39 let _mdns = maybe_advertise(&settings.mdns, addr.port()); 37 40 ··· 82 85 )) 83 86 .add_service(EventsServiceServer::with_interceptor( 84 87 events::EventsSvc::default(), 88 88 + interceptor.clone(), 89 89 + )) 90 90 + .add_service(SnapcastServiceServer::with_interceptor( 91 91 + snap_svc, 85 92 interceptor, 86 93 )) 87 94 .serve(addr) ··· 132 139 ); 133 140 tracing::warn!("auth: BEARER TOKEN = {}", token); 134 141 Ok(token) 142 142 + } 143 143 + 144 144 + /// Spawn the snapcast supervisor task if `[snapcast].enabled = true`, and 145 145 + /// hand it to the gRPC service. When disabled, the service still 146 146 + /// registers but every RPC returns `FAILED_PRECONDITION`. 147 147 + fn build_snapcast_svc(cfg: &settings::SnapcastSettings) -> snapcast::SnapcastSvc { 148 148 + if !cfg.enabled { 149 149 + tracing::info!("snapcast: disabled in zerod.toml"); 150 150 + return snapcast::SnapcastSvc::disabled(); 151 151 + } 152 152 + tracing::info!( 153 153 + "snapcast: connecting to {}:{} (notifications={})", 154 154 + cfg.host, 155 155 + cfg.port, 156 156 + cfg.forward_notifications, 157 157 + ); 158 158 + let client = zerod_snapcast::SnapcastClient::spawn( 159 159 + cfg.host.clone(), 160 160 + cfg.port, 161 161 + cfg.forward_notifications, 162 162 + ); 163 163 + snapcast::SnapcastSvc::enabled(Arc::new(client)) 135 164 } 136 165 137 166 /// Returns the live [`Advertisement`] so the caller keeps it alive for the

+39

crates/server/src/settings.rs

View file Reviewed

··· 29 29 pub mdns: MdnsSettings, 30 30 #[serde(default)] 31 31 pub configs: Vec<ManagedConfig>, 32 32 + #[serde(default)] 33 33 + pub snapcast: SnapcastSettings, 34 34 + } 35 35 + 36 36 + #[derive(Debug, Clone, Deserialize, Serialize)] 37 37 + pub struct SnapcastSettings { 38 38 + /// Connect to a snapserver on startup and expose `SnapcastService`. 39 39 + /// When `false`, the service is still reachable but every RPC returns 40 40 + /// `FAILED_PRECONDITION` so reflection-based clients can see it exists. 41 41 + #[serde(default)] 42 42 + pub enabled: bool, 43 43 + #[serde(default = "default_snap_host")] 44 44 + pub host: String, 45 45 + #[serde(default = "default_snap_port")] 46 46 + pub port: u16, 47 47 + /// Forward snapserver push notifications onto the in-process event bus. 48 48 + /// Useful so external `snapctl` changes still surface to `events tail`. 49 49 + #[serde(default = "default_true")] 50 50 + pub forward_notifications: bool, 51 51 + } 52 52 + 53 53 + impl Default for SnapcastSettings { 54 54 + fn default() -> Self { 55 55 + Self { 56 56 + enabled: false, 57 57 + host: default_snap_host(), 58 58 + port: default_snap_port(), 59 59 + forward_notifications: true, 60 60 + } 61 61 + } 62 62 + } 63 63 + 64 64 + fn default_snap_host() -> String { 65 65 + "127.0.0.1".to_string() 66 66 + } 67 67 + 68 68 + fn default_snap_port() -> u16 { 69 69 + 1705 32 70 } 33 71 34 72 #[derive(Debug, Clone, Deserialize, Serialize)] ··· 88 126 systemd: SystemdSettings::default(), 89 127 mdns: MdnsSettings::default(), 90 128 configs: Vec::new(), 129 129 + snapcast: SnapcastSettings::default(), 91 130 } 92 131 } 93 132 }

+179

crates/server/src/snapcast.rs

View file Reviewed

··· 1 1 + //! `SnapcastService` — wraps the `zerod_snapcast::SnapcastClient` if 2 2 + //! enabled in zerod.toml; otherwise every RPC returns 3 3 + //! `FAILED_PRECONDITION` so reflection still lists the service. 4 4 + 5 5 + use std::sync::Arc; 6 6 + use tonic::{Request, Response, Status}; 7 7 + use zerod_proto::v1alpha1 as pb; 8 8 + use zerod_proto::v1alpha1::snapcast_service_server::SnapcastService; 9 9 + use zerod_snapcast::{SnapcastClient, SnapClient as SnapClientT, SnapGroup as SnapGroupT, SnapStream as SnapStreamT, SnapServer as SnapServerT}; 10 10 + 11 11 + pub struct SnapcastSvc { 12 12 + inner: Option<Arc<SnapcastClient>>, 13 13 + } 14 14 + 15 15 + impl SnapcastSvc { 16 16 + pub fn enabled(client: Arc<SnapcastClient>) -> Self { 17 17 + Self { inner: Some(client) } 18 18 + } 19 19 + 20 20 + pub fn disabled() -> Self { 21 21 + Self { inner: None } 22 22 + } 23 23 + 24 24 + fn client(&self) -> Result<&SnapcastClient, Status> { 25 25 + self.inner 26 26 + .as_deref() 27 27 + .ok_or_else(|| Status::failed_precondition("snapcast disabled in zerod.toml")) 28 28 + } 29 29 + } 30 30 + 31 31 + fn rpc_err(e: anyhow::Error) -> Status { 32 32 + Status::unavailable(format!("snapcast: {e:#}")) 33 33 + } 34 34 + 35 35 + #[tonic::async_trait] 36 36 + impl SnapcastService for SnapcastSvc { 37 37 + async fn get_server_status( 38 38 + &self, 39 39 + _req: Request<pb::GetServerStatusRequest>, 40 40 + ) -> Result<Response<pb::GetServerStatusResponse>, Status> { 41 41 + let s = self.client()?.get_server_status().await.map_err(rpc_err)?; 42 42 + Ok(Response::new(pb::GetServerStatusResponse { 43 43 + server: Some(to_pb_server(&s)), 44 44 + })) 45 45 + } 46 46 + 47 47 + async fn list_clients( 48 48 + &self, 49 49 + _req: Request<pb::ListClientsRequest>, 50 50 + ) -> Result<Response<pb::ListClientsResponse>, Status> { 51 51 + let s = self.client()?.get_server_status().await.map_err(rpc_err)?; 52 52 + let mut clients = Vec::new(); 53 53 + for g in &s.groups { 54 54 + for c in &g.clients { 55 55 + clients.push(to_pb_client(c)); 56 56 + } 57 57 + } 58 58 + Ok(Response::new(pb::ListClientsResponse { clients })) 59 59 + } 60 60 + 61 61 + async fn list_snap_streams( 62 62 + &self, 63 63 + _req: Request<pb::ListSnapStreamsRequest>, 64 64 + ) -> Result<Response<pb::ListSnapStreamsResponse>, Status> { 65 65 + let s = self.client()?.get_server_status().await.map_err(rpc_err)?; 66 66 + let streams = s.streams.iter().map(to_pb_stream).collect(); 67 67 + Ok(Response::new(pb::ListSnapStreamsResponse { streams })) 68 68 + } 69 69 + 70 70 + async fn set_client_volume( 71 71 + &self, 72 72 + req: Request<pb::SetClientVolumeRequest>, 73 73 + ) -> Result<Response<pb::SetClientVolumeResponse>, Status> { 74 74 + let r = req.into_inner(); 75 75 + self.client()? 76 76 + .set_client_volume(&r.client_id, r.volume_percent, r.muted) 77 77 + .await 78 78 + .map_err(rpc_err)?; 79 79 + Ok(Response::new(pb::SetClientVolumeResponse {})) 80 80 + } 81 81 + 82 82 + async fn set_client_latency( 83 83 + &self, 84 84 + req: Request<pb::SetClientLatencyRequest>, 85 85 + ) -> Result<Response<pb::SetClientLatencyResponse>, Status> { 86 86 + let r = req.into_inner(); 87 87 + self.client()? 88 88 + .set_client_latency(&r.client_id, r.latency_ms) 89 89 + .await 90 90 + .map_err(rpc_err)?; 91 91 + Ok(Response::new(pb::SetClientLatencyResponse {})) 92 92 + } 93 93 + 94 94 + async fn set_client_name( 95 95 + &self, 96 96 + req: Request<pb::SetClientNameRequest>, 97 97 + ) -> Result<Response<pb::SetClientNameResponse>, Status> { 98 98 + let r = req.into_inner(); 99 99 + self.client()? 100 100 + .set_client_name(&r.client_id, &r.name) 101 101 + .await 102 102 + .map_err(rpc_err)?; 103 103 + Ok(Response::new(pb::SetClientNameResponse {})) 104 104 + } 105 105 + 106 106 + async fn set_group_stream( 107 107 + &self, 108 108 + req: Request<pb::SetGroupStreamRequest>, 109 109 + ) -> Result<Response<pb::SetGroupStreamResponse>, Status> { 110 110 + let r = req.into_inner(); 111 111 + self.client()? 112 112 + .set_group_stream(&r.group_id, &r.stream_id) 113 113 + .await 114 114 + .map_err(rpc_err)?; 115 115 + Ok(Response::new(pb::SetGroupStreamResponse {})) 116 116 + } 117 117 + 118 118 + async fn set_group_mute( 119 119 + &self, 120 120 + req: Request<pb::SetGroupMuteRequest>, 121 121 + ) -> Result<Response<pb::SetGroupMuteResponse>, Status> { 122 122 + let r = req.into_inner(); 123 123 + self.client()? 124 124 + .set_group_mute(&r.group_id, r.muted) 125 125 + .await 126 126 + .map_err(rpc_err)?; 127 127 + Ok(Response::new(pb::SetGroupMuteResponse {})) 128 128 + } 129 129 + 130 130 + async fn set_group_clients( 131 131 + &self, 132 132 + req: Request<pb::SetGroupClientsRequest>, 133 133 + ) -> Result<Response<pb::SetGroupClientsResponse>, Status> { 134 134 + let r = req.into_inner(); 135 135 + self.client()? 136 136 + .set_group_clients(&r.group_id, r.client_ids) 137 137 + .await 138 138 + .map_err(rpc_err)?; 139 139 + Ok(Response::new(pb::SetGroupClientsResponse {})) 140 140 + } 141 141 + } 142 142 + 143 143 + fn to_pb_server(s: &SnapServerT) -> pb::SnapServer { 144 144 + pb::SnapServer { 145 145 + groups: s.groups.iter().map(to_pb_group).collect(), 146 146 + streams: s.streams.iter().map(to_pb_stream).collect(), 147 147 + } 148 148 + } 149 149 + 150 150 + fn to_pb_group(g: &SnapGroupT) -> pb::SnapGroup { 151 151 + pb::SnapGroup { 152 152 + id: g.id.clone(), 153 153 + name: g.name.clone(), 154 154 + stream_id: g.stream_id.clone(), 155 155 + muted: g.muted, 156 156 + client_ids: g.clients.iter().map(|c| c.id.clone()).collect(), 157 157 + } 158 158 + } 159 159 + 160 160 + fn to_pb_client(c: &SnapClientT) -> pb::SnapClient { 161 161 + pb::SnapClient { 162 162 + id: c.id.clone(), 163 163 + name: c.display_name().to_string(), 164 164 + connected: c.connected, 165 165 + volume_percent: c.config.volume.percent, 166 166 + muted: c.config.volume.muted, 167 167 + latency_ms: c.config.latency, 168 168 + mac: c.host.mac.clone(), 169 169 + host: c.host.name.clone(), 170 170 + version: c.snapclient.version.clone(), 171 171 + } 172 172 + } 173 173 + 174 174 + fn to_pb_stream(s: &SnapStreamT) -> pb::SnapStream { 175 175 + pb::SnapStream { 176 176 + id: s.id.clone(), 177 177 + status: s.status.clone(), 178 178 + } 179 179 + }

+12

crates/snapcast/Cargo.toml

View file Reviewed

··· 1 1 + [package] 2 2 + name = "zerod-snapcast" 3 3 + version = "0.1.0" 4 4 + edition = "2021" 5 5 + 6 6 + [dependencies] 7 7 + anyhow = { workspace = true } 8 8 + serde = { workspace = true } 9 9 + serde_json = { workspace = true } 10 10 + tokio = { workspace = true } 11 11 + tracing = { workspace = true } 12 12 + zerod-events = { path = "../events" }

+268

crates/snapcast/src/client.rs

View file Reviewed

··· 1 1 + //! Long-lived JSON-RPC 2.0 client. 2 2 + //! 3 3 + //! Architecture: one supervisor task drives a single TCP connection. 4 4 + //! Caller-side `call_*` methods talk to the supervisor via an unbounded 5 5 + //! mpsc; the supervisor select!s between socket-reads and command-sends. 6 6 + //! Pending request ids live in a `HashMap<u64, oneshot::Sender>` owned by 7 7 + //! the supervisor — on disconnect the map is dropped, every in-flight 8 8 + //! caller observes `oneshot::Receiver::Canceled`, and the supervisor 9 9 + //! reconnects with backoff. 10 10 + 11 11 + use anyhow::{anyhow, bail, Context as _, Result}; 12 12 + use serde::de::DeserializeOwned; 13 13 + use serde_json::{json, Value}; 14 14 + use std::collections::HashMap; 15 15 + use std::sync::{Arc, Mutex}; 16 16 + use std::time::Duration; 17 17 + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; 18 18 + use tokio::net::TcpStream; 19 19 + use tokio::sync::{mpsc, oneshot}; 20 20 + 21 21 + use crate::types::{ServerStatusResult, SnapServer}; 22 22 + 23 23 + pub struct SnapcastClient { 24 24 + cmd_tx: mpsc::UnboundedSender<Command>, 25 25 + } 26 26 + 27 27 + struct Command { 28 28 + method: &'static str, 29 29 + params: Value, 30 30 + reply: oneshot::Sender<Result<Value>>, 31 31 + } 32 32 + 33 33 + impl SnapcastClient { 34 34 + /// Spawn the supervisor task. Drop the returned `SnapcastClient` to 35 35 + /// stop it (the unbounded sender closes; the supervisor returns Ok). 36 36 + pub fn spawn(host: String, port: u16, forward_notifications: bool) -> Self { 37 37 + let (cmd_tx, cmd_rx) = mpsc::unbounded_channel(); 38 38 + tokio::spawn(supervisor(host, port, forward_notifications, cmd_rx)); 39 39 + Self { cmd_tx } 40 40 + } 41 41 + 42 42 + pub async fn call_raw(&self, method: &'static str, params: Value) -> Result<Value> { 43 43 + let (tx, rx) = oneshot::channel(); 44 44 + self.cmd_tx 45 45 + .send(Command { 46 46 + method, 47 47 + params, 48 48 + reply: tx, 49 49 + }) 50 50 + .map_err(|_| anyhow!("snapcast: supervisor task is gone"))?; 51 51 + rx.await 52 52 + .map_err(|_| anyhow!("snapcast: connection dropped before reply"))? 53 53 + } 54 54 + 55 55 + pub async fn call<T: DeserializeOwned>( 56 56 + &self, 57 57 + method: &'static str, 58 58 + params: Value, 59 59 + ) -> Result<T> { 60 60 + let v = self.call_raw(method, params).await?; 61 61 + serde_json::from_value(v) 62 62 + .map_err(|e| anyhow!("snapcast: parse {method} result: {e}")) 63 63 + } 64 64 + 65 65 + pub async fn get_server_status(&self) -> Result<SnapServer> { 66 66 + let r: ServerStatusResult = self.call("Server.GetStatus", Value::Null).await?; 67 67 + Ok(r.server) 68 68 + } 69 69 + 70 70 + pub async fn set_client_volume(&self, id: &str, percent: u32, muted: bool) -> Result<()> { 71 71 + self.call_raw( 72 72 + "Client.SetVolume", 73 73 + json!({ "id": id, "volume": { "percent": percent.min(100), "muted": muted } }), 74 74 + ) 75 75 + .await?; 76 76 + Ok(()) 77 77 + } 78 78 + 79 79 + pub async fn set_client_latency(&self, id: &str, latency_ms: i32) -> Result<()> { 80 80 + self.call_raw( 81 81 + "Client.SetLatency", 82 82 + json!({ "id": id, "latency": latency_ms }), 83 83 + ) 84 84 + .await?; 85 85 + Ok(()) 86 86 + } 87 87 + 88 88 + pub async fn set_client_name(&self, id: &str, name: &str) -> Result<()> { 89 89 + self.call_raw("Client.SetName", json!({ "id": id, "name": name })) 90 90 + .await?; 91 91 + Ok(()) 92 92 + } 93 93 + 94 94 + pub async fn set_group_stream(&self, group_id: &str, stream_id: &str) -> Result<()> { 95 95 + self.call_raw( 96 96 + "Group.SetStream", 97 97 + json!({ "id": group_id, "stream_id": stream_id }), 98 98 + ) 99 99 + .await?; 100 100 + Ok(()) 101 101 + } 102 102 + 103 103 + pub async fn set_group_mute(&self, group_id: &str, muted: bool) -> Result<()> { 104 104 + self.call_raw("Group.SetMute", json!({ "id": group_id, "mute": muted })) 105 105 + .await?; 106 106 + Ok(()) 107 107 + } 108 108 + 109 109 + pub async fn set_group_clients(&self, group_id: &str, client_ids: Vec<String>) -> Result<()> { 110 110 + self.call_raw( 111 111 + "Group.SetClients", 112 112 + json!({ "id": group_id, "clients": client_ids }), 113 113 + ) 114 114 + .await?; 115 115 + Ok(()) 116 116 + } 117 117 + } 118 118 + 119 119 + async fn supervisor( 120 120 + host: String, 121 121 + port: u16, 122 122 + forward: bool, 123 123 + mut cmd_rx: mpsc::UnboundedReceiver<Command>, 124 124 + ) { 125 125 + let mut backoff = Duration::from_millis(500); 126 126 + loop { 127 127 + match TcpStream::connect((host.as_str(), port)).await { 128 128 + Ok(sock) => { 129 129 + tracing::info!("snapcast: connected to {host}:{port}"); 130 130 + backoff = Duration::from_millis(500); 131 131 + match run_connection(sock, forward, &mut cmd_rx).await { 132 132 + Ok(()) => return, 133 133 + Err(e) => tracing::warn!("snapcast: connection ended: {e:#}"), 134 134 + } 135 135 + } 136 136 + Err(e) => { 137 137 + tracing::warn!( 138 138 + "snapcast: connect {host}:{port} failed: {e}; retrying in {backoff:?}" 139 139 + ); 140 140 + } 141 141 + } 142 142 + // While we sleep before the next connect, refuse any commands 143 143 + // instead of letting them queue silently. Without this, RPCs on 144 144 + // a never-up snapserver hang indefinitely. 145 145 + let deadline = tokio::time::sleep(backoff); 146 146 + tokio::pin!(deadline); 147 147 + loop { 148 148 + tokio::select! { 149 149 + _ = &mut deadline => break, 150 150 + cmd = cmd_rx.recv() => { 151 151 + let Some(cmd) = cmd else { return; }; 152 152 + let _ = cmd.reply.send(Err(anyhow!("snapcast: not connected"))); 153 153 + } 154 154 + } 155 155 + } 156 156 + backoff = (backoff * 2).min(Duration::from_secs(30)); 157 157 + } 158 158 + } 159 159 + 160 160 + async fn run_connection( 161 161 + sock: TcpStream, 162 162 + forward: bool, 163 163 + cmd_rx: &mut mpsc::UnboundedReceiver<Command>, 164 164 + ) -> Result<()> { 165 165 + let (rd, mut wr) = sock.into_split(); 166 166 + let mut lines = BufReader::new(rd).lines(); 167 167 + let pending: Arc<Mutex<HashMap<u64, oneshot::Sender<Result<Value>>>>> = 168 168 + Arc::new(Mutex::new(HashMap::new())); 169 169 + let mut next_id: u64 = 1; 170 170 + 171 171 + loop { 172 172 + tokio::select! { 173 173 + line = lines.next_line() => match line { 174 174 + Ok(Some(s)) => { 175 175 + if let Err(e) = handle_inbound(&s, &pending, forward) { 176 176 + tracing::debug!("snapcast: bad inbound: {e}"); 177 177 + } 178 178 + } 179 179 + Ok(None) => return Err(anyhow!("connection closed")), 180 180 + Err(e) => return Err(anyhow!("read error: {e}")), 181 181 + }, 182 182 + cmd = cmd_rx.recv() => { 183 183 + let Some(cmd) = cmd else { 184 184 + // Owner dropped the client; clean shutdown. 185 185 + return Ok(()); 186 186 + }; 187 187 + let id = next_id; 188 188 + next_id = next_id.wrapping_add(1); 189 189 + let body = json!({ 190 190 + "jsonrpc": "2.0", 191 191 + "id": id, 192 192 + "method": cmd.method, 193 193 + "params": cmd.params, 194 194 + }); 195 195 + let mut bytes = serde_json::to_vec(&body)?; 196 196 + bytes.push(b'\r'); 197 197 + bytes.push(b'\n'); 198 198 + pending.lock().unwrap().insert(id, cmd.reply); 199 199 + if let Err(e) = wr.write_all(&bytes).await { 200 200 + return Err(anyhow!("write error: {e}")); 201 201 + } 202 202 + } 203 203 + } 204 204 + } 205 205 + } 206 206 + 207 207 + fn handle_inbound( 208 208 + line: &str, 209 209 + pending: &Mutex<HashMap<u64, oneshot::Sender<Result<Value>>>>, 210 210 + forward: bool, 211 211 + ) -> Result<()> { 212 212 + let line = line.trim(); 213 213 + if line.is_empty() { 214 214 + return Ok(()); 215 215 + } 216 216 + let v: Value = serde_json::from_str(line).context("parse inbound json")?; 217 217 + if let Some(id_v) = v.get("id") { 218 218 + // Response to a prior request. 219 219 + let Some(id) = id_v.as_u64() else { 220 220 + bail!("response with non-numeric id"); 221 221 + }; 222 222 + let reply = pending.lock().unwrap().remove(&id); 223 223 + if let Some(tx) = reply { 224 224 + if let Some(err) = v.get("error") { 225 225 + let _ = tx.send(Err(anyhow!("snapcast rpc error: {err}"))); 226 226 + } else if let Some(result) = v.get("result") { 227 227 + let _ = tx.send(Ok(result.clone())); 228 228 + } else { 229 229 + let _ = tx.send(Err(anyhow!("snapcast: response missing result and error"))); 230 230 + } 231 231 + } 232 232 + Ok(()) 233 233 + } else { 234 234 + // Notification (no id). 235 235 + if forward { 236 236 + let method = v.get("method").and_then(Value::as_str).unwrap_or(""); 237 237 + let params = v.get("params").cloned().unwrap_or(Value::Null); 238 238 + forward_notification(method, &params); 239 239 + } 240 240 + Ok(()) 241 241 + } 242 242 + } 243 243 + 244 244 + fn forward_notification(method: &str, params: &Value) { 245 245 + match method { 246 246 + "Client.OnVolumeChanged" => { 247 247 + let id = params.get("id").and_then(Value::as_str).unwrap_or(""); 248 248 + let vol = params.get("volume"); 249 249 + let percent = vol 250 250 + .and_then(|v| v.get("percent")) 251 251 + .and_then(Value::as_u64) 252 252 + .unwrap_or(0) as u32; 253 253 + let muted = vol 254 254 + .and_then(|v| v.get("muted")) 255 255 + .and_then(Value::as_bool) 256 256 + .unwrap_or(false); 257 257 + zerod_events::publish(zerod_events::Event::SnapcastClientChanged { 258 258 + client_id: id.to_string(), 259 259 + name: String::new(), 260 260 + volume_percent: percent, 261 261 + muted, 262 262 + }); 263 263 + } 264 264 + _ => { 265 265 + tracing::debug!("snapcast: notification {method}"); 266 266 + } 267 267 + } 268 268 + }

+14

crates/snapcast/src/lib.rs

View file Reviewed

··· 1 1 + //! Snapcast (snapserver) JSON-RPC 2.0 client over TCP. 2 2 + //! 3 3 + //! One long-lived connection per [`SnapcastClient`], with backoff reconnect. 4 4 + //! Server-push notifications are forwarded to the in-process event bus when 5 5 + //! `forward_notifications = true` so external `snapctl` changes don't go 6 6 + //! invisible to subscribers. 7 7 + 8 8 + mod client; 9 9 + mod types; 10 10 + 11 11 + pub use client::SnapcastClient; 12 12 + pub use types::{ 13 13 + SnapClient, SnapClientConfig, SnapGroup, SnapHost, SnapServer, SnapStream, SnapVolume, 14 14 + };

+92

crates/snapcast/src/types.rs

View file Reviewed

··· 1 1 + //! JSON shapes returned by `Server.GetStatus`. Fields use `#[serde(default)]` 2 2 + //! so a schema bump in snapserver doesn't break parsing — missing fields 3 3 + //! land at their default rather than failing the whole call. 4 4 + 5 5 + use serde::{Deserialize, Serialize}; 6 6 + 7 7 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 8 8 + #[serde(default)] 9 9 + pub(crate) struct ServerStatusResult { 10 10 + pub server: SnapServer, 11 11 + } 12 12 + 13 13 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 14 14 + #[serde(default)] 15 15 + pub struct SnapServer { 16 16 + pub groups: Vec<SnapGroup>, 17 17 + pub streams: Vec<SnapStream>, 18 18 + } 19 19 + 20 20 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 21 21 + #[serde(default)] 22 22 + pub struct SnapGroup { 23 23 + pub id: String, 24 24 + pub name: String, 25 25 + pub stream_id: String, 26 26 + pub muted: bool, 27 27 + pub clients: Vec<SnapClient>, 28 28 + } 29 29 + 30 30 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 31 31 + #[serde(default)] 32 32 + pub struct SnapClient { 33 33 + pub id: String, 34 34 + pub connected: bool, 35 35 + pub host: SnapHost, 36 36 + pub config: SnapClientConfig, 37 37 + pub snapclient: SnapClientMeta, 38 38 + } 39 39 + 40 40 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 41 41 + #[serde(default)] 42 42 + pub struct SnapHost { 43 43 + pub name: String, 44 44 + pub mac: String, 45 45 + pub ip: String, 46 46 + pub os: String, 47 47 + pub arch: String, 48 48 + } 49 49 + 50 50 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 51 51 + #[serde(default)] 52 52 + pub struct SnapClientConfig { 53 53 + pub name: String, 54 54 + pub volume: SnapVolume, 55 55 + pub latency: i32, 56 56 + pub instance: u32, 57 57 + } 58 58 + 59 59 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 60 60 + #[serde(default)] 61 61 + pub struct SnapVolume { 62 62 + pub percent: u32, 63 63 + pub muted: bool, 64 64 + } 65 65 + 66 66 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 67 67 + #[serde(default)] 68 68 + pub struct SnapClientMeta { 69 69 + pub version: String, 70 70 + pub protocol_version: u32, 71 71 + } 72 72 + 73 73 + #[derive(Debug, Clone, Default, Deserialize, Serialize)] 74 74 + #[serde(default)] 75 75 + pub struct SnapStream { 76 76 + pub id: String, 77 77 + pub status: String, 78 78 + } 79 79 + 80 80 + impl SnapClient { 81 81 + /// Best-effort display name: explicit config.name first, falling back to 82 82 + /// host name, then the client id. 83 83 + pub fn display_name(&self) -> &str { 84 84 + if !self.config.name.is_empty() { 85 85 + &self.config.name 86 86 + } else if !self.host.name.is_empty() { 87 87 + &self.host.name 88 88 + } else { 89 89 + &self.id 90 90 + } 91 91 + } 92 92 + }

+159

src/main.rs

View file Reviewed

··· 106 106 /// Subscribe to server-side events (server-streaming). 107 107 #[command(subcommand)] 108 108 Events(EventsCmd), 109 109 + /// Control a Snapcast server (snapserver JSON-RPC). 110 110 + #[command(subcommand)] 111 111 + Snapcast(SnapcastCmd), 109 112 /// Browse mDNS for zerod servers on the LAN. 110 113 Discover, 114 114 + } 115 115 + 116 116 + #[derive(Subcommand)] 117 117 + enum SnapcastCmd { 118 118 + /// Dump the full server status as JSON. 119 119 + Status, 120 120 + /// List clients across all groups. 121 121 + Clients, 122 122 + /// List configured streams. 123 123 + Streams, 124 124 + /// Set a client's volume (0..=100). Pass --muted to also mute. 125 125 + Volume { 126 126 + client_id: String, 127 127 + percent: u32, 128 128 + #[arg(long)] 129 129 + muted: bool, 130 130 + }, 131 131 + /// Set a client's audio latency offset in milliseconds. 132 132 + Latency { client_id: String, ms: i32 }, 133 133 + /// Rename a client. 134 134 + Name { client_id: String, name: String }, 135 135 + /// Move a group to a different stream. 136 136 + GroupStream { group_id: String, stream_id: String }, 137 137 + /// Mute / unmute a group. 138 138 + GroupMute { group_id: String, muted: bool }, 139 139 + /// Replace the client list for a group. 140 140 + GroupClients { 141 141 + group_id: String, 142 142 + client_ids: Vec<String>, 143 143 + }, 111 144 } 112 145 113 146 #[derive(Subcommand)] ··· 308 341 Some(Command::System(cmd)) => run_system(&endpoint.unwrap(), bearer_token, cmd).await, 309 342 Some(Command::Volume(cmd)) => run_volume(&endpoint.unwrap(), bearer_token, cmd).await, 310 343 Some(Command::Events(cmd)) => run_events(&endpoint.unwrap(), bearer_token, cmd).await, 344 344 + Some(Command::Snapcast(cmd)) => run_snapcast(&endpoint.unwrap(), bearer_token, cmd).await, 311 345 } 312 346 } 313 347 ··· 825 859 "timestamp_ms": ev.timestamp_ms, 826 860 "kind": kind, 827 861 "payload": payload, 862 862 + }) 863 863 + } 864 864 + 865 865 + // --- snapcast -------------------------------------------------------------- 866 866 + 867 867 + async fn run_snapcast(ep: &Endpoint, token: Option<String>, cmd: SnapcastCmd) -> Result<()> { 868 868 + let ch = channel(ep).await?; 869 869 + let mut client = pb::snapcast_service_client::SnapcastServiceClient::new(ch); 870 870 + match cmd { 871 871 + SnapcastCmd::Status => { 872 872 + let r = client 873 873 + .get_server_status(attach_token( 874 874 + Request::new(pb::GetServerStatusRequest {}), 875 875 + &token, 876 876 + )) 877 877 + .await? 878 878 + .into_inner(); 879 879 + print_json(&serde_json::to_value(snap_server_to_json(r.server.as_ref()))?)?; 880 880 + } 881 881 + SnapcastCmd::Clients => { 882 882 + let r = client 883 883 + .list_clients(attach_token(Request::new(pb::ListClientsRequest {}), &token)) 884 884 + .await? 885 885 + .into_inner(); 886 886 + for c in r.clients { 887 887 + println!( 888 888 + "{:36} {:24} connected={} vol={}% muted={} latency={}ms host={}", 889 889 + c.id, c.name, c.connected, c.volume_percent, c.muted, c.latency_ms, c.host, 890 890 + ); 891 891 + } 892 892 + } 893 893 + SnapcastCmd::Streams => { 894 894 + let r = client 895 895 + .list_snap_streams(attach_token( 896 896 + Request::new(pb::ListSnapStreamsRequest {}), 897 897 + &token, 898 898 + )) 899 899 + .await? 900 900 + .into_inner(); 901 901 + for s in r.streams { 902 902 + println!("{:24} {}", s.id, s.status); 903 903 + } 904 904 + } 905 905 + SnapcastCmd::Volume { client_id, percent, muted } => { 906 906 + client 907 907 + .set_client_volume(attach_token( 908 908 + Request::new(pb::SetClientVolumeRequest { 909 909 + client_id, 910 910 + volume_percent: percent, 911 911 + muted, 912 912 + }), 913 913 + &token, 914 914 + )) 915 915 + .await?; 916 916 + println!("ok"); 917 917 + } 918 918 + SnapcastCmd::Latency { client_id, ms } => { 919 919 + client 920 920 + .set_client_latency(attach_token( 921 921 + Request::new(pb::SetClientLatencyRequest { 922 922 + client_id, 923 923 + latency_ms: ms, 924 924 + }), 925 925 + &token, 926 926 + )) 927 927 + .await?; 928 928 + println!("ok"); 929 929 + } 930 930 + SnapcastCmd::Name { client_id, name } => { 931 931 + client 932 932 + .set_client_name(attach_token( 933 933 + Request::new(pb::SetClientNameRequest { client_id, name }), 934 934 + &token, 935 935 + )) 936 936 + .await?; 937 937 + println!("ok"); 938 938 + } 939 939 + SnapcastCmd::GroupStream { group_id, stream_id } => { 940 940 + client 941 941 + .set_group_stream(attach_token( 942 942 + Request::new(pb::SetGroupStreamRequest { group_id, stream_id }), 943 943 + &token, 944 944 + )) 945 945 + .await?; 946 946 + println!("ok"); 947 947 + } 948 948 + SnapcastCmd::GroupMute { group_id, muted } => { 949 949 + client 950 950 + .set_group_mute(attach_token( 951 951 + Request::new(pb::SetGroupMuteRequest { group_id, muted }), 952 952 + &token, 953 953 + )) 954 954 + .await?; 955 955 + println!("ok"); 956 956 + } 957 957 + SnapcastCmd::GroupClients { group_id, client_ids } => { 958 958 + client 959 959 + .set_group_clients(attach_token( 960 960 + Request::new(pb::SetGroupClientsRequest { group_id, client_ids }), 961 961 + &token, 962 962 + )) 963 963 + .await?; 964 964 + println!("ok"); 965 965 + } 966 966 + } 967 967 + Ok(()) 968 968 + } 969 969 + 970 970 + fn snap_server_to_json(s: Option<&pb::SnapServer>) -> serde_json::Value { 971 971 + use serde_json::json; 972 972 + let Some(s) = s else { 973 973 + return json!({"groups": [], "streams": []}); 974 974 + }; 975 975 + json!({ 976 976 + "groups": s.groups.iter().map(|g| json!({ 977 977 + "id": g.id, 978 978 + "name": g.name, 979 979 + "stream_id": g.stream_id, 980 980 + "muted": g.muted, 981 981 + "client_ids": g.client_ids, 982 982 + })).collect::<Vec<_>>(), 983 983 + "streams": s.streams.iter().map(|st| json!({ 984 984 + "id": st.id, 985 985 + "status": st.status, 986 986 + })).collect::<Vec<_>>(), 828 987 }) 829 988 } 830 989

+12

zerod.toml.example

View file Reviewed

··· 51 51 key = "squeezelite" 52 52 path = "/etc/default/squeezelite" 53 53 unit = "squeezelite.service" 54 54 + 55 55 + [snapcast] 56 56 + # When enabled, zerod connects to a local snapserver's JSON-RPC port on 57 57 + # startup and exposes SnapcastService (clients, groups, streams). 58 58 + # Disabled → the service is still reachable but every RPC returns 59 59 + # FAILED_PRECONDITION (reflection-based UIs still see it exists). 60 60 + enabled = false 61 61 + host = "127.0.0.1" 62 62 + port = 1705 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