Refactor audio input feed to use actor (#937)

* audio actor?

* blocking mic feed actor

* manage connect and error events properly

* restructure to be async

* cleanuop

* formatting

* don't hold app state while setting mic input

Author: Brendonovich

Cargo.lock

@@ -65,7 +65,7 @@ impl RecordStart {
             cap_recording::RecordingBaseInputs {
                 capture_target: target_info,
                 capture_system_audio: self.system_audio,
-                mic_feed: &None,
+                mic_feed: None,
             },
             camera.map(|c| Arc::new(Mutex::new(c))),
             false,

apps/cli/src/record.rs

@@ -101,6 +101,7 @@ md5 = "0.7.0"
 tokio-util = "0.7.15"
 wgpu.workspace = true
 bytemuck = "1.23.1"
+kameo = "0.17.2"
 
 [target.'cfg(target_os = "macos")'.dependencies]
 core-graphics = "0.24.0"

apps/desktop/src-tauri/Cargo.toml

@@ -1,4 +1,4 @@
-use cap_recording::feeds::{AudioInputSamples, AudioInputSamplesReceiver};
+use cap_recording::feeds::microphone::MicrophoneSamples;
 use cpal::{SampleFormat, StreamInstant};
 use keyed_priority_queue::KeyedPriorityQueue;
 use serde::{Deserialize, Serialize};
@@ -15,7 +15,10 @@ const MIN_DB: f64 = -96.0;
 #[derive(Deserialize, specta::Type, Serialize, tauri_specta::Event, Debug, Clone)]
 pub struct AudioInputLevelChange(f64);
 
-pub fn spawn_event_emitter(app_handle: AppHandle, audio_input_rx: AudioInputSamplesReceiver) {
+pub fn spawn_event_emitter(
+    app_handle: AppHandle,
+    audio_input_rx: flume::Receiver<MicrophoneSamples>,
+) {
     let mut time_window = VolumeMeter::new(0.2);
     tokio::spawn(async move {
         while let Ok(samples) = audio_input_rx.recv_async().await {
@@ -106,7 +109,7 @@ fn db_fs(data: impl Iterator<Item = f64>) -> f64 {
     (20.0 * (max as f64 / MAX_AMPLITUDE_F32).log10()).clamp(MIN_DB, 0.0)
 }
 
-fn samples_to_f64(samples: &AudioInputSamples) -> impl Iterator<Item = f64> + use<'_> {
+fn samples_to_f64(samples: &MicrophoneSamples) -> impl Iterator<Item = f64> + use<'_> {
     samples
         .data
         .chunks(samples.format.sample_size())

apps/desktop/src-tauri/src/audio_meter.rs

@@ -27,72 +27,64 @@ use audio::AppSounds;
 use auth::{AuthStore, AuthenticationInvalid, Plan};
 use camera::{CameraPreview, CameraWindowState};
 use cap_displays::{DisplayId, WindowId, bounds::LogicalBounds};
-use cap_editor::EditorInstance;
-use cap_editor::EditorState;
-use cap_project::RecordingMetaInner;
-use cap_project::XY;
+use cap_editor::{EditorInstance, EditorState};
 use cap_project::{
-    ProjectConfiguration, RecordingMeta, SharingMeta, StudioRecordingMeta, ZoomSegment,
+    ProjectConfiguration, RecordingMeta, RecordingMetaInner, SharingMeta, StudioRecordingMeta, XY,
+    ZoomSegment,
 };
-use cap_recording::feeds::DeviceOrModelID;
 use cap_recording::{
-    feeds::CameraFeed,
-    feeds::RawCameraFrame,
-    feeds::{AudioInputFeed, AudioInputSamplesSender},
+    feeds::{
+        self, CameraFeed, DeviceOrModelID, RawCameraFrame,
+        microphone::{self, MicrophoneFeed},
+    },
     sources::ScreenCaptureTarget,
 };
-use cap_rendering::ProjectRecordingsMeta;
-use cap_rendering::RenderedFrame;
+use cap_rendering::{ProjectRecordingsMeta, RenderedFrame};
 use clipboard_rs::common::RustImage;
 use clipboard_rs::{Clipboard, ClipboardContext};
-use editor_window::EditorInstances;
-use editor_window::WindowEditorInstance;
+use editor_window::{EditorInstances, WindowEditorInstance};
 use general_settings::GeneralSettingsStore;
+use kameo::{Actor, actor::ActorRef};
 use mp4::Mp4Reader;
 use notifications::NotificationType;
 use png::{ColorType, Encoder};
 use recording::InProgressRecording;
 use relative_path::RelativePathBuf;
-
-use scap::capturer::Capturer;
-use scap::frame::Frame;
-use scap::frame::VideoFrame;
+use scap::{
+    capturer::Capturer,
+    frame::{Frame, VideoFrame},
+};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use specta::Type;
-use std::collections::BTreeMap;
-use std::path::Path;
-use std::time::Duration;
 use std::{
+    collections::BTreeMap,
     fs::File,
     future::Future,
     io::{BufReader, BufWriter},
     marker::PhantomData,
-    path::PathBuf,
+    path::{Path, PathBuf},
     process::Command,
     str::FromStr,
     sync::Arc,
+    time::Duration,
 };
-use tauri::Window;
-use tauri::{AppHandle, Manager, State, WindowEvent};
+use tauri::{AppHandle, Manager, State, Window, WindowEvent};
 use tauri_plugin_deep_link::DeepLinkExt;
 use tauri_plugin_dialog::DialogExt;
 use tauri_plugin_global_shortcut::GlobalShortcutExt;
 use tauri_plugin_notification::{NotificationExt, PermissionState};
 use tauri_plugin_opener::OpenerExt;
 use tauri_plugin_shell::ShellExt;
 use tauri_specta::Event;
-use tokio::sync::mpsc;
-use tokio::sync::{Mutex, RwLock};
-use tokio::time::timeout;
-use tracing::debug;
-use tracing::error;
-use tracing::trace;
+use tokio::{
+    sync::{Mutex, RwLock, mpsc},
+    time::timeout,
+};
+use tracing::{error, trace};
 use upload::{S3UploadMeta, create_or_get_video, upload_image, upload_video};
 use web_api::ManagerExt as WebManagerExt;
-use windows::EditorWindowIds;
-use windows::set_window_transparent;
-use windows::{CapWindowId, ShowCapWindow};
+use windows::{CapWindowId, EditorWindowIds, ShowCapWindow, set_window_transparent};
 
 use crate::upload::build_video_meta;
 
@@ -116,15 +108,13 @@ pub struct App {
     #[serde(skip)]
     camera_feed_initialization: Option<mpsc::Sender<()>>,
     #[serde(skip)]
-    mic_feed: Option<AudioInputFeed>,
-    #[serde(skip)]
-    mic_samples_tx: AudioInputSamplesSender,
-    #[serde(skip)]
     handle: AppHandle,
     #[serde(skip)]
     recording_state: RecordingState,
     #[serde(skip)]
     recording_logging_handle: LoggingHandle,
+    #[serde(skip)]
+    mic_feed: ActorRef<feeds::microphone::MicrophoneFeed>,
     server_url: String,
 }
 
@@ -227,27 +217,26 @@ impl App {
 #[tauri::command]
 #[specta::specta]
 async fn set_mic_input(state: MutableState<'_, App>, label: Option<String>) -> Result<(), String> {
-    let mut app = state.write().await;
+    let mic_feed = state.read().await.mic_feed.clone();
 
-    match (label, &mut app.mic_feed) {
-        (Some(label), None) => {
-            AudioInputFeed::init(&label)
-                .await
-                .map_err(|e| e.to_string())
-                .map(async |feed| {
-                    feed.add_sender(app.mic_samples_tx.clone()).await.unwrap();
-                    app.mic_feed = Some(feed);
-                })
-                .transpose_async()
+    match label {
+        None => {
+            mic_feed
+                .ask(microphone::RemoveInput)
                 .await
+                .map_err(|e| e.to_string())?;
         }
-        (Some(label), Some(feed)) => feed.switch_input(&label).await.map_err(|e| e.to_string()),
-        (None, _) => {
-            debug!("removing mic in set_start_recording_options");
-            app.mic_feed.take();
-            Ok(())
+        Some(label) => {
+            mic_feed
+                .ask(feeds::microphone::SetInput { label })
+                .await
+                .map_err(|e| e.to_string())?
+                .await
+                .map_err(|e| e.to_string())?;
         }
     }
+
+    Ok(())
 }
 
 #[tauri::command]
@@ -1119,7 +1108,7 @@ async fn list_audio_devices() -> Result<Vec<String>, ()> {
         return Ok(vec![]);
     }
 
-    Ok(AudioInputFeed::list_devices().keys().cloned().collect())
+    Ok(MicrophoneFeed::list().keys().cloned().collect())
 }
 
 #[derive(Serialize, Type, tauri_specta::Event, Debug, Clone)]
@@ -2024,7 +2013,28 @@ pub async fn run(recording_logging_handle: LoggingHandle) {
 
     let (camera_tx, camera_ws_port, _shutdown) = camera_legacy::create_camera_preview_ws().await;
 
-    let (audio_input_tx, audio_input_rx) = AudioInputFeed::create_channel();
+    let (mic_samples_tx, mic_samples_rx) = flume::bounded(8);
+
+    let mic_feed = {
+        let (error_tx, error_rx) = flume::bounded(1);
+
+        let mic_feed = MicrophoneFeed::spawn(MicrophoneFeed::new(error_tx));
+
+        // TODO: make this part of a global actor one day
+        tokio::spawn(async move {
+            let Ok(err) = error_rx.recv_async().await else {
+                return;
+            };
+
+            error!("Mic feed actor error: {err}");
+        });
+
+        let _ = mic_feed
+            .ask(feeds::microphone::AddSender(mic_samples_tx))
+            .await;
+
+        mic_feed
+    };
 
     tauri::async_runtime::set(tokio::runtime::Handle::current());
 
@@ -2121,10 +2131,9 @@ pub async fn run(recording_logging_handle: LoggingHandle) {
                     handle: app.clone(),
                     camera_feed: None,
                     camera_feed_initialization: None,
-                    mic_samples_tx: audio_input_tx,
-                    mic_feed: None,
                     recording_state: RecordingState::None,
                     recording_logging_handle,
+                    mic_feed,
                     server_url: GeneralSettingsStore::get(&app)
                         .ok()
                         .flatten()
@@ -2173,7 +2182,7 @@ pub async fn run(recording_logging_handle: LoggingHandle) {
                 }
             });
 
-            audio_meter::spawn_event_emitter(app.clone(), audio_input_rx);
+            audio_meter::spawn_event_emitter(app.clone(), mic_samples_rx);
 
             tray::create_tray(&app).unwrap();
 
@@ -2213,7 +2222,8 @@ pub async fn run(recording_logging_handle: LoggingHandle) {
                                     let app_state = &mut *state.write().await;
 
                                     if !app_state.is_recording_active_or_pending() {
-                                        app_state.mic_feed.take();
+                                        let _ =
+                                            app_state.mic_feed.ask(microphone::RemoveInput).await;
                                         app_state.camera_feed.take();
 
                                         if let Some(camera) = CapWindowId::Camera.get(&app) {
@@ -2422,28 +2432,6 @@ trait EventExt: tauri_specta::Event {
 
 impl<T: tauri_specta::Event> EventExt for T {}
 
-trait TransposeAsync {
-    type Output;
-
-    fn transpose_async(self) -> impl Future<Output = Self::Output>
-    where
-        Self: Sized;
-}
-
-impl<F: Future<Output = T>, T, E> TransposeAsync for Result<F, E> {
-    type Output = Result<T, E>;
-
-    async fn transpose_async(self) -> Self::Output
-    where
-        Self: Sized,
-    {
-        match self {
-            Ok(f) => Ok(f.await),
-            Err(e) => Err(e),
-        }
-    }
-}
-
 fn open_project_from_path(path: &Path, app: AppHandle) -> Result<(), String> {
     let meta = RecordingMeta::load_for_project(path).map_err(|v| v.to_string())?;