1use std::{collections::HashMap, sync::Arc};
2
3use tracing::info;
4use wayle_core::Property;
5use wayle_traits::{ModelMonitoring, ServiceMonitoring};
6
7use crate::{
8 core::{
9 device::{input::InputDevice, output::OutputDevice},
10 stream::AudioStream,
11 },
12 error::Error,
13 events::AudioEvent,
14 service::AudioService,
15 types::{
16 device::{Device, DeviceKey},
17 stream::{StreamKey, StreamType},
18 },
19};
20
21impl ServiceMonitoring for AudioService {
22 type Error = Error;
23
24 #[allow(clippy::too_many_lines)]
25 async fn start_monitoring(&self) -> Result<(), Self::Error> {
26 let mut event_rx = self.event_tx.subscribe();
27 let mut output_devs: HashMap<DeviceKey, Arc<OutputDevice>> = HashMap::new();
28 let mut input_devs: HashMap<DeviceKey, Arc<InputDevice>> = HashMap::new();
29 let mut streams: HashMap<StreamKey, Arc<AudioStream>> = HashMap::new();
30
31 let command_tx = self.command_tx.clone();
32 let event_tx = self.event_tx.clone();
33 let output_devices = self.output_devices.clone();
34 let input_devices = self.input_devices.clone();
35 let playback_streams = self.playback_streams.clone();
36 let recording_streams = self.recording_streams.clone();
37 let default_input = self.default_input.clone();
38 let default_output = self.default_output.clone();
39 let cancellation_token = self.cancellation_token.clone();
40
41 tokio::spawn(async move {
42 loop {
43 tokio::select! {
44 _ = cancellation_token.cancelled() => {
45 info!("AudioMonitoring cancelled, stopping");
46 return;
47 }
48 Ok(event) = event_rx.recv() => {
49 match event {
50 AudioEvent::DeviceAdded(device) => {
51 match device {
52 Device::Sink(sink) => {
53 let key = sink.key();
54 let output = Arc::new(OutputDevice::from_sink(
55 &sink,
56 command_tx.clone(),
57 Some(event_tx.clone()),
58 Some(cancellation_token.child_token()),
59 ));
60 output.clone().start_monitoring().await.ok();
61 output_devs.insert(key, output);
62 output_devices.set(output_devs.values().cloned().collect());
63 }
64 Device::Source(source) => {
65 let key = source.key();
66 let input = Arc::new(InputDevice::from_source(
67 &source,
68 command_tx.clone(),
69 Some(event_tx.clone()),
70 Some(cancellation_token.child_token()),
71 ));
72 input.clone().start_monitoring().await.ok();
73 input_devs.insert(key, input);
74 input_devices.set(input_devs.values().cloned().collect());
75 }
76 }
77 }
78
79 AudioEvent::DeviceChanged(device) => {
80 match device {
81 Device::Sink(sink) => {
82 let key = sink.key();
83 if let Some(existing) = output_devs.get(&key) {
84 existing.update_from_sink(&sink);
85 } else {
86 let output = Arc::new(OutputDevice::from_sink(
87 &sink,
88 command_tx.clone(),
89 Some(event_tx.clone()),
90 Some(cancellation_token.child_token()),
91 ));
92 output.clone().start_monitoring().await.ok();
93 output_devs.insert(key, output);
94 output_devices.set(output_devs.values().cloned().collect());
95 }
96 }
97 Device::Source(source) => {
98 let key = source.key();
99 if let Some(existing) = input_devs.get(&key) {
100 existing.update_from_source(&source);
101 } else {
102 let input = Arc::new(InputDevice::from_source(
103 &source,
104 command_tx.clone(),
105 Some(event_tx.clone()),
106 Some(cancellation_token.child_token()),
107 ));
108 input.clone().start_monitoring().await.ok();
109 input_devs.insert(key, input);
110 input_devices.set(input_devs.values().cloned().collect());
111 }
112 }
113 }
114 }
115
116 AudioEvent::DeviceRemoved(key) => {
117 if let Some(device) = output_devs.remove(&key) {
118 if let Some(ref cancel_token) = device.cancellation_token {
119 cancel_token.cancel();
120 };
121
122 output_devices.set(output_devs.values().cloned().collect());
123 }
124 if input_devs.remove(&key).is_some() {
125 input_devices.set(input_devs.values().cloned().collect());
126 }
127 }
128
129 AudioEvent::StreamAdded(info) => {
130 let stream = Arc::new(AudioStream::from_info(
131 info.clone(),
132 command_tx.clone(),
133 Some(event_tx.clone()),
134 Some(cancellation_token.child_token()),
135 ));
136 stream.clone().start_monitoring().await.ok();
137 streams.insert(info.key(), stream);
138 update_stream_properties(&streams, &playback_streams, &recording_streams);
139 }
140
141 AudioEvent::StreamChanged(info) => {
142 let key = info.key();
143 if let Some(existing) = streams.get(&key) {
144 existing.update_from_info(&info);
145 } else {
146 let stream = Arc::new(AudioStream::from_info(
147 info.clone(),
148 command_tx.clone(),
149 Some(event_tx.clone()),
150 Some(cancellation_token.child_token()),
151 ));
152 stream.clone().start_monitoring().await.ok();
153 streams.insert(key, stream);
154 update_stream_properties(&streams, &playback_streams, &recording_streams);
155 }
156 }
157
158 AudioEvent::StreamRemoved(key) => {
159 if let Some(cancel_token) = streams
160 .remove(&key)
161 .and_then(|stream| stream.cancellation_token.clone())
162 {
163 cancel_token.cancel();
164 }
165 update_stream_properties(&streams, &playback_streams, &recording_streams);
166 }
167
168 AudioEvent::DefaultInputChanged(maybe_device) => {
169 let device = maybe_device.and_then(|dev| {
170 match dev {
171 Device::Source(source) => {
172 let key = source.key();
173 input_devs.get(&key).cloned()
174 }
175 _ => None,
176 }
177 });
178 default_input.set(device);
179 }
180
181 AudioEvent::DefaultOutputChanged(maybe_device) => {
182 let device = maybe_device.and_then(|dev| {
183 match dev {
184 Device::Sink(sink) => {
185 let key = sink.key();
186 output_devs.get(&key).cloned()
187 }
188 _ => None,
189 }
190 });
191 default_output.set(device);
192 }
193 }
194 }
195 }
196 }
197 });
198
199 Ok(())
200 }
201}
202
203fn update_stream_properties(
204 streams: &HashMap<StreamKey, Arc<AudioStream>>,
205 playback_streams: &Property<Vec<Arc<AudioStream>>>,
206 recording_streams: &Property<Vec<Arc<AudioStream>>>,
207) {
208 let playback: Vec<Arc<AudioStream>> = streams
209 .values()
210 .filter(|s| s.key.stream_type == StreamType::Playback)
211 .cloned()
212 .collect();
213
214 let recording: Vec<Arc<AudioStream>> = streams
215 .values()
216 .filter(|s| s.key.stream_type == StreamType::Record)
217 .cloned()
218 .collect();
219
220 playback_streams.set(playback);
221 recording_streams.set(recording);
222}
223
224#[cfg(test)]
225mod tests {
226 use std::collections::HashMap;
227
228 use tokio::sync::mpsc;
229 use wayle_core::Property;
230
231 use super::*;
232 use crate::{
233 backend::types::CommandSender,
234 types::{
235 format::{ChannelMap, SampleFormat, SampleSpec},
236 stream::{MediaInfo, StreamInfo, StreamKey, StreamState},
237 },
238 volume::types::Volume,
239 };
240
241 fn create_test_stream(index: u32, stream_type: StreamType) -> Arc<AudioStream> {
242 let (command_tx, _): (CommandSender, _) = mpsc::unbounded_channel();
243
244 let stream_info = StreamInfo {
245 index,
246 stream_type,
247 name: format!("test-stream-{}", index),
248 application_name: None,
249 binary: None,
250 pid: None,
251 owner_module: None,
252 client: None,
253 device_index: 0,
254 volume: Volume::mono(1.0),
255 muted: false,
256 corked: false,
257 has_volume: true,
258 volume_writable: true,
259 state: StreamState::Running,
260 sample_spec: SampleSpec {
261 format: SampleFormat::S16LE,
262 rate: 44100,
263 channels: 2,
264 },
265 channel_map: ChannelMap {
266 channels: 2,
267 positions: vec![],
268 },
269 properties: HashMap::new(),
270 media: MediaInfo {
271 title: None,
272 artist: None,
273 album: None,
274 icon_name: None,
275 },
276 buffer_latency: 0,
277 device_latency: 0,
278 resample_method: None,
279 driver: String::from("test"),
280 format: None,
281 };
282
283 Arc::new(AudioStream::from_info(stream_info, command_tx, None, None))
284 }
285
286 #[test]
287 fn update_stream_properties_filters_playback_streams_correctly() {
288 let mut streams = HashMap::new();
289 streams.insert(
290 StreamKey::new(1, StreamType::Playback),
291 create_test_stream(1, StreamType::Playback),
292 );
293 streams.insert(
294 StreamKey::new(2, StreamType::Playback),
295 create_test_stream(2, StreamType::Playback),
296 );
297 streams.insert(
298 StreamKey::new(3, StreamType::Record),
299 create_test_stream(3, StreamType::Record),
300 );
301
302 let playback = Property::new(Vec::new());
303 let recording = Property::new(Vec::new());
304
305 update_stream_properties(&streams, &playback, &recording);
306
307 assert_eq!(playback.get().len(), 2);
308 assert_eq!(recording.get().len(), 1);
309 }
310
311 #[test]
312 fn update_stream_properties_filters_recording_streams_correctly() {
313 let mut streams = HashMap::new();
314 streams.insert(
315 StreamKey::new(1, StreamType::Record),
316 create_test_stream(1, StreamType::Record),
317 );
318 streams.insert(
319 StreamKey::new(2, StreamType::Record),
320 create_test_stream(2, StreamType::Record),
321 );
322 streams.insert(
323 StreamKey::new(3, StreamType::Playback),
324 create_test_stream(3, StreamType::Playback),
325 );
326
327 let playback = Property::new(Vec::new());
328 let recording = Property::new(Vec::new());
329
330 update_stream_properties(&streams, &playback, &recording);
331
332 assert_eq!(playback.get().len(), 1);
333 assert_eq!(recording.get().len(), 2);
334 }
335
336 #[test]
337 fn update_stream_properties_handles_empty_streams() {
338 let streams = HashMap::new();
339
340 let playback = Property::new(Vec::new());
341 let recording = Property::new(Vec::new());
342
343 update_stream_properties(&streams, &playback, &recording);
344
345 assert_eq!(playback.get().len(), 0);
346 assert_eq!(recording.get().len(), 0);
347 }
348
349 #[test]
350 fn update_stream_properties_handles_mixed_stream_types() {
351 let mut streams = HashMap::new();
352 streams.insert(
353 StreamKey::new(1, StreamType::Playback),
354 create_test_stream(1, StreamType::Playback),
355 );
356 streams.insert(
357 StreamKey::new(2, StreamType::Record),
358 create_test_stream(2, StreamType::Record),
359 );
360 streams.insert(
361 StreamKey::new(3, StreamType::Playback),
362 create_test_stream(3, StreamType::Playback),
363 );
364 streams.insert(
365 StreamKey::new(4, StreamType::Record),
366 create_test_stream(4, StreamType::Record),
367 );
368
369 let playback = Property::new(Vec::new());
370 let recording = Property::new(Vec::new());
371
372 update_stream_properties(&streams, &playback, &recording);
373
374 assert_eq!(playback.get().len(), 2);
375 assert_eq!(recording.get().len(), 2);
376 }
377}