1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

use crate::audio::{AudioBus, AudioSpec};
use crate::cpal_utils;
use crate::signal_flow::node::{ControlMessage, Processor, ProcessorState};

use anyhow::Result;
use cpal::{
    self,
    traits::{DeviceTrait, HostTrait, StreamTrait},
};
use crossbeam_channel::{unbounded, Receiver, Sender, TryRecvError};

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread::{self, JoinHandle};
use std::time::Duration;

const RECORDER_POLL: Duration = Duration::from_millis(100);

#[derive(Debug)]
pub enum RecorderProcessorControlMessage {
    Shutdown,
}

impl ControlMessage for RecorderProcessorControlMessage {
    fn shutdown_msg() -> Self {
        RecorderProcessorControlMessage::Shutdown
    }
}

pub struct RecorderProcessor {
    spec: AudioSpec,
    finished: Arc<AtomicBool>,
    channel_senders: Vec<Sender<Vec<f32>>>,
}

impl RecorderProcessor {
    pub fn new(spec: AudioSpec) -> (RecorderProcessor, AudioBus) {
        let (bus, channel_senders) = AudioBus::from_spec(spec, None);
        (
            RecorderProcessor {
                spec,
                channel_senders,
                finished: Arc::new(AtomicBool::new(false)),
            },
            bus,
        )
    }

    fn run(mut self, ctrl_rx: Receiver<RecorderProcessorControlMessage>) -> Result<()> {
        let host = cpal::default_host();
        let input_device = host
            .default_input_device()
            .expect("failed to get default input device");
        info!(
            "Using default input device: \"{}\"",
            input_device.name().unwrap()
        );

        let supported_configs = input_device
            .supported_input_configs()
            .expect("failed to query input device configs");
        let stream_config = cpal_utils::find_input_stream_config(
            supported_configs,
            self.spec.channels,
            self.spec.sample_rate,
        )?;

        let channel_senders = self.channel_senders.clone();

        let input_stream = input_device
            .build_input_stream(
                &stream_config,
                move |data: &[f32], _: &cpal::InputCallbackInfo| {
                    // react to stream events and read or write stream data here.
                    send_samples_from_raw_input(data, self.spec.channels, &channel_senders)
                },
                move |err| {
                    panic!("audio input stream failed: {:?}", err);
                },
            )
            .expect("failed to build input stream");
        input_stream.play().expect("failed to start input stream");
        loop {
            if self.finished.load(Ordering::Relaxed) {
                break;
            }
            match self.handle_control_messages(&ctrl_rx)? {
                ProcessorState::Finished => {
                    break;
                }
                _ => {}
            }
            thread::sleep(RECORDER_POLL);
        }
        Ok(())
    }
}

fn send_samples_from_raw_input(
    buf: &[f32],
    n_channels: u16,
    channel_senders: &Vec<Sender<Vec<f32>>>,
) {
    // optimisation opportunity here by creating inner vecs with capacities
    let mut channels: Vec<Vec<f32>> = (0..n_channels).map(|_| vec![]).collect();
    for buffer_interleaved_samples in buf.chunks(n_channels as usize) {
        for i in 0..n_channels as usize {
            unsafe {
                channels
                    .get_unchecked_mut(i)
                    .push(*buffer_interleaved_samples.get_unchecked(i));
            }
        }
    }
    for (i, channel) in channels.into_iter().enumerate() {
        unsafe {
            channel_senders.get_unchecked(i).send(channel).unwrap();
        }
    }
}

impl Processor<RecorderProcessorControlMessage> for RecorderProcessor {
    fn handle_control_messages(
        &mut self,
        rx: &Receiver<RecorderProcessorControlMessage>,
    ) -> Result<ProcessorState> {
        match rx.try_recv() {
            Ok(msg) => match msg {
                RecorderProcessorControlMessage::Shutdown => {
                    self.finished.store(true, Ordering::Relaxed);
                    Ok(ProcessorState::Finished)
                }
            },
            Err(TryRecvError::Disconnected) => Ok(ProcessorState::Finished),
            Err(TryRecvError::Empty) => Ok(ProcessorState::Running),
        }
    }

    fn start(
        self,
        finished: Arc<AtomicBool>,
    ) -> (Sender<RecorderProcessorControlMessage>, JoinHandle<()>) {
        let (ctrl_tx, ctrl_rx) = unbounded();
        let handle = thread::spawn(move || {
            self.run(ctrl_rx).unwrap();
            finished.store(true, Ordering::Relaxed);
        });
        (ctrl_tx, handle)
    }
}