Merge pull request #480 from orottier/feature/offline-event-loop

Also run the EventLoop for OfflineAudioContext

Author: orottier

src/context/concrete_base.rs

@@ -13,7 +13,7 @@ use crate::spatial::AudioListenerParams;
 
 use crate::AudioListener;
 
-use crossbeam_channel::{Receiver, SendError, Sender};
+use crossbeam_channel::{SendError, Sender};
 use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
 use std::sync::{Arc, Mutex, RwLock, RwLockWriteGuard};
 
@@ -108,7 +108,7 @@ struct ConcreteBaseAudioContextInner {
     /// Stores the event handlers
     event_loop: EventLoop,
     /// Sender for events that will be handled by the EventLoop
-    event_send: Option<Sender<EventDispatch>>,
+    event_send: Sender<EventDispatch>,
 }
 
 impl BaseAudioContext for ConcreteBaseAudioContext {
@@ -126,16 +126,11 @@ impl ConcreteBaseAudioContext {
         state: Arc<AtomicU8>,
         frames_played: Arc<AtomicU64>,
         render_channel: Sender<ControlMessage>,
-        event_channel: Option<(Sender<EventDispatch>, Receiver<EventDispatch>)>,
+        event_send: Sender<EventDispatch>,
+        event_loop: EventLoop,
         offline: bool,
         node_id_consumer: llq::Consumer<AudioNodeId>,
     ) -> Self {
-        let event_loop = EventLoop::new();
-        let (event_send, event_recv) = match event_channel {
-            None => (None, None),
-            Some((send, recv)) => (Some(send), Some(recv)),
-        };
-
         let audio_node_id_provider = AudioNodeIdProvider::new(node_id_consumer);
 
         let base_inner = ConcreteBaseAudioContextInner {
@@ -150,7 +145,7 @@ impl ConcreteBaseAudioContext {
             listener_params: None,
             offline,
             state,
-            event_loop: event_loop.clone(),
+            event_loop,
             event_send,
         };
         let base = Self {
@@ -219,13 +214,6 @@ impl ConcreteBaseAudioContext {
             crate::node::load_hrtf_processor(sample_rate as u32);
         }
 
-        // Boot the event loop thread that handles the events spawned by the render thread
-        // (we don't do this for offline rendering because it makes little sense, the graph cannot
-        // be mutated once rendering has started anyway)
-        if let Some(event_channel) = event_recv {
-            event_loop.run(event_channel);
-        }
-
         base
     }
 
@@ -288,10 +276,7 @@ impl ConcreteBaseAudioContext {
     }
 
     pub(crate) fn send_event(&self, msg: EventDispatch) -> Result<(), SendError<EventDispatch>> {
-        match self.inner.event_send.as_ref() {
-            Some(s) => s.send(msg),
-            None => Err(SendError(msg)),
-        }
+        self.inner.event_send.send(msg)
     }
 
     pub(crate) fn lock_control_msg_sender(&self) -> RwLockWriteGuard<'_, Sender<ControlMessage>> {

src/context/offline.rs

@@ -5,10 +5,13 @@ use std::sync::{Arc, Mutex};
 
 use crate::buffer::AudioBuffer;
 use crate::context::{AudioContextState, BaseAudioContext, ConcreteBaseAudioContext};
+use crate::events::{
+    Event, EventDispatch, EventHandler, EventPayload, EventType, OfflineAudioCompletionEvent,
+};
 use crate::render::RenderThread;
 use crate::{assert_valid_sample_rate, RENDER_QUANTUM_SIZE};
-use crate::{Event, OfflineAudioCompletionEvent};
 
+use crate::events::EventLoop;
 use futures_channel::{mpsc, oneshot};
 use futures_util::SinkExt as _;
 
@@ -48,28 +51,14 @@ struct OfflineAudioContextRenderer {
     suspend_callbacks: Vec<(usize, Box<OfflineAudioContextCallback>)>,
     /// channel to listen for `resume` calls on a suspended context
     resume_receiver: mpsc::Receiver<()>,
-    /// event handler for statechange event
-    onstatechange_handler: Option<Box<dyn FnMut(Event) + Send + 'static>>,
-    /// event handler for complete event
-    oncomplete_handler: Option<Box<dyn FnOnce(OfflineAudioCompletionEvent) + Send + 'static>>,
+    /// event loop to run after each render quantum
+    event_loop: EventLoop,
 }
 
 impl BaseAudioContext for OfflineAudioContext {
     fn base(&self) -> &ConcreteBaseAudioContext {
         &self.base
     }
-
-    fn set_onstatechange<F: FnMut(Event) + Send + 'static>(&self, callback: F) {
-        if let Some(renderer) = self.renderer.lock().unwrap().as_mut() {
-            renderer.onstatechange_handler = Some(Box::new(callback));
-        }
-    }
-
-    fn clear_onstatechange(&self) {
-        if let Some(renderer) = self.renderer.lock().unwrap().as_mut() {
-            renderer.onstatechange_handler = None;
-        }
-    }
 }
 
 impl OfflineAudioContext {
@@ -100,13 +89,19 @@ impl OfflineAudioContext {
         let state = Arc::new(AtomicU8::new(AudioContextState::Suspended as u8));
         let state_clone = Arc::clone(&state);
 
+        // Communication channel for events from the render thread to the control thread.
+        // Use an unbounded channel because we do not require real-time safety.
+        let (event_send, event_recv) = crossbeam_channel::unbounded();
+        let event_loop = EventLoop::new(event_recv);
+
         // setup the render 'thread', which will run inside the control thread
         let renderer = RenderThread::new(
             sample_rate,
             number_of_channels,
             receiver,
             state_clone,
             frames_played_clone,
+            event_send.clone(),
         );
 
         // first, setup the base audio context
@@ -116,7 +111,8 @@ impl OfflineAudioContext {
             state,
             frames_played,
             sender,
-            None,
+            event_send,
+            event_loop.clone(),
             true,
             node_id_consumer,
         );
@@ -128,8 +124,7 @@ impl OfflineAudioContext {
             suspend_promises: Vec::new(),
             suspend_callbacks: Vec::new(),
             resume_receiver,
-            onstatechange_handler: None,
-            oncomplete_handler: None,
+            event_loop,
         };
 
         Self {
@@ -159,23 +154,25 @@ impl OfflineAudioContext {
             .unwrap()
             .take()
             .expect("InvalidStateError - Cannot call `startRendering` twice");
+
         let OfflineAudioContextRenderer {
             renderer,
             suspend_callbacks,
-            oncomplete_handler,
-            mut onstatechange_handler,
+            event_loop,
             ..
         } = renderer;
 
         self.base.set_state(AudioContextState::Running);
-        Self::emit_statechange(&mut onstatechange_handler);
 
-        let result = renderer.render_audiobuffer_sync(self.length, suspend_callbacks, self);
+        let result = renderer.render_audiobuffer_sync(self, suspend_callbacks, &event_loop);
 
         self.base.set_state(AudioContextState::Closed);
-        Self::emit_statechange(&mut onstatechange_handler);
+        let _ = self
+            .base
+            .send_event(EventDispatch::complete(result.clone()));
 
-        Self::emit_complete(oncomplete_handler, &result);
+        // spin the event loop once more to handle the statechange/complete events
+        event_loop.handle_pending_events();
 
         result
     }
@@ -204,48 +201,27 @@ impl OfflineAudioContext {
             renderer,
             suspend_promises,
             resume_receiver,
-            oncomplete_handler,
-            mut onstatechange_handler,
+            event_loop,
             ..
         } = renderer;
 
         self.base.set_state(AudioContextState::Running);
-        Self::emit_statechange(&mut onstatechange_handler);
 
         let result = renderer
-            .render_audiobuffer(self.length, suspend_promises, resume_receiver)
+            .render_audiobuffer(self.length, suspend_promises, resume_receiver, &event_loop)
             .await;
 
         self.base.set_state(AudioContextState::Closed);
-        Self::emit_statechange(&mut onstatechange_handler);
+        let _ = self
+            .base
+            .send_event(EventDispatch::complete(result.clone()));
 
-        Self::emit_complete(oncomplete_handler, &result);
+        // spin the event loop once more to handle the statechange/complete events
+        event_loop.handle_pending_events();
 
         result
     }
 
-    fn emit_complete(
-        oncomplete_handler: Option<Box<dyn FnOnce(OfflineAudioCompletionEvent) + Send>>,
-        result: &AudioBuffer,
-    ) {
-        if let Some(callback) = oncomplete_handler {
-            let event = OfflineAudioCompletionEvent {
-                rendered_buffer: result.clone(),
-                event: Event { type_: "complete" },
-            };
-            (callback)(event);
-        }
-    }
-
-    fn emit_statechange(onstatechange_handler: &mut Option<Box<dyn FnMut(Event) + Send>>) {
-        if let Some(callback) = onstatechange_handler.as_mut() {
-            let event = Event {
-                type_: "statechange",
-            };
-            (callback)(event);
-        }
-    }
-
     /// get the length of rendering audio buffer
     // false positive: OfflineAudioContext is not const
     #[allow(clippy::missing_const_for_fn, clippy::unused_self)]
@@ -422,17 +398,24 @@ impl OfflineAudioContext {
         &self,
         callback: F,
     ) {
-        if let Some(renderer) = self.renderer.lock().unwrap().as_mut() {
-            renderer.oncomplete_handler = Some(Box::new(callback));
-        }
+        let callback = move |v| match v {
+            EventPayload::Complete(v) => {
+                let event = OfflineAudioCompletionEvent {
+                    rendered_buffer: v,
+                    event: Event { type_: "complete" },
+                };
+                callback(event)
+            }
+            _ => unreachable!(),
+        };
+
+        self.base()
+            .set_event_handler(EventType::Complete, EventHandler::Once(Box::new(callback)));
     }
 
     /// Unset the callback to run when the rendering has completed
-    #[allow(clippy::missing_panics_doc)]
     pub fn clear_oncomplete(&self) {
-        if let Some(renderer) = self.renderer.lock().unwrap().as_mut() {
-            renderer.oncomplete_handler = None;
-        }
+        self.base().clear_event_handler(EventType::Complete);
     }
 }
 
@@ -587,6 +570,23 @@ mod tests {
         assert!(changed.load(Ordering::Relaxed));
     }
 
+    #[test]
+    fn test_onstatechange_async() {
+        use futures::executor;
+
+        let context = OfflineAudioContext::new(2, 555, 44_100.);
+
+        let changed = Arc::new(AtomicBool::new(false));
+        let changed_clone = Arc::clone(&changed);
+        context.set_onstatechange(move |_event| {
+            changed_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = executor::block_on(context.start_rendering());
+
+        assert!(changed.load(Ordering::Relaxed));
+    }
+
     #[test]
     fn test_oncomplete() {
         let mut context = OfflineAudioContext::new(2, 555, 44_100.);
@@ -602,4 +602,22 @@ mod tests {
 
         assert!(complete.load(Ordering::Relaxed));
     }
+
+    #[test]
+    fn test_oncomplete_async() {
+        use futures::executor;
+
+        let context = OfflineAudioContext::new(2, 555, 44_100.);
+
+        let complete = Arc::new(AtomicBool::new(false));
+        let complete_clone = Arc::clone(&complete);
+        context.set_oncomplete(move |event| {
+            assert_eq!(event.rendered_buffer.length(), 555);
+            complete_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = executor::block_on(context.start_rendering());
+
+        assert!(complete.load(Ordering::Relaxed));
+    }
 }

src/context/online.rs

@@ -3,7 +3,7 @@ use std::error::Error;
 use std::sync::Mutex;
 
 use crate::context::{AudioContextState, BaseAudioContext, ConcreteBaseAudioContext};
-use crate::events::{EventDispatch, EventHandler, EventPayload, EventType};
+use crate::events::{EventDispatch, EventHandler, EventLoop, EventPayload, EventType};
 use crate::io::{self, AudioBackendManager, ControlThreadInit, NoneBackend, RenderThreadInit};
 use crate::media_devices::{enumerate_devices_sync, MediaDeviceInfoKind};
 use crate::media_streams::{MediaStream, MediaStreamTrack};
@@ -181,11 +181,13 @@ impl AudioContext {
     #[allow(clippy::needless_pass_by_value)]
     #[must_use]
     pub fn new(mut options: AudioContextOptions) -> Self {
+        // Log, but ignore invalid sinks
         if !is_valid_sink_id(&options.sink_id) {
             log::error!("NotFoundError: invalid sinkId {:?}", options.sink_id);
             options.sink_id = String::from("");
         }
 
+        // Set up the audio output thread
         let (control_thread_init, render_thread_init) = io::thread_init();
         let backend = io::build_output(options, render_thread_init.clone());
 
@@ -198,26 +200,37 @@ impl AudioContext {
             event_recv,
         } = control_thread_init;
 
+        // Construct the audio Graph and hand it to the render thread
         let (node_id_producer, node_id_consumer) = llq::Queue::new().split();
         let graph = Graph::new(node_id_producer);
         let message = ControlMessage::Startup { graph };
         ctrl_msg_send.send(message).unwrap();
 
+        // Set up the event loop thread that handles the events spawned by the render thread
+        let event_loop = EventLoop::new(event_recv);
+
+        // Put everything together in the BaseAudioContext (shared with offline context)
         let base = ConcreteBaseAudioContext::new(
             backend.sample_rate(),
             backend.number_of_channels(),
             state,
             frames_played,
             ctrl_msg_send,
-            Some((event_send, event_recv)),
+            event_send,
+            event_loop.clone(),
             false,
             node_id_consumer,
         );
 
-        // setup AudioRenderCapacity for this context
+        // Setup AudioRenderCapacity for this context
         let base_clone = base.clone();
         let render_capacity = AudioRenderCapacity::new(base_clone, load_value_recv);
 
+        // As the final step, spawn a thread for the event loop. If we do this earlier we may miss
+        // event handling of the initial events that are emitted right after render thread
+        // construction.
+        event_loop.run_in_thread();
+
         Self {
             base,
             backend_manager: Mutex::new(backend),