codewandler_audio/
source.rs

1use crate::{AudioSource, SampleFormat};
2use crossbeam_channel::{Receiver, Sender, unbounded};
3use std::sync::{Arc, Mutex};
4use std::thread;
5
6pub trait IntoAudioSource {
7    type Format: SampleFormat;
8    fn into_audio_source(&mut self) -> anyhow::Result<Box<dyn AudioSource<Format = Self::Format>>>;
9}
10
11pub struct AudioSourceFanOut<F, K>
12where
13    F: SampleFormat,
14    K: IntoAudioSource,
15{
16    _k: K,
17    subscribers: Arc<Mutex<Vec<Sender<F>>>>,
18}
19
20impl<F, K> AudioSourceFanOut<F, K>
21where
22    F: SampleFormat + 'static,
23    K: IntoAudioSource<Format = F>,
24{
25    pub fn new(mut k: K) -> Self {
26        let mut source = k.into_audio_source().unwrap();
27        let subscribers = Arc::new(Mutex::new(Vec::<Sender<F>>::new()));
28        let subs_clone = Arc::clone(&subscribers);
29
30        // Fan-out thread
31        thread::spawn(move || {
32            while let Some(sample) = source.audio_read() {
33                let mut disconnected = vec![];
34                let mut subs = subs_clone.lock().unwrap();
35                for (i, tx) in subs.iter().enumerate() {
36                    if tx.send(sample).is_err() {
37                        disconnected.push(i);
38                    }
39                }
40                // Remove any dropped subscribers
41                for i in disconnected.into_iter().rev() {
42                    subs.remove(i);
43                }
44            }
45        });
46
47        Self { _k: k, subscribers }
48    }
49
50    pub fn subscribe(&self) -> Receiver<F> {
51        let (tx, rx) = unbounded();
52        self.subscribers.lock().unwrap().push(tx);
53        rx
54    }
55}
codewandler_audio/source.rs

codewandler_audio/
source.rs
