Introduce AudioProcessor::before_drop pre-destructor hook

This allows the processor to send some final events, or move some items
to the garbage collector thread.

In this commit we use the before_drop hook to ensure that the ended
event is sent for all AudioScheduledSourceNodes when the stop time is
implicit. This fixes #489

Author: orottier

src/node/audio_buffer_source.rs

@@ -771,13 +771,23 @@ impl AudioProcessor for AudioBufferSourceRenderer {
 
         log::warn!("AudioBufferSourceRenderer: Dropping incoming message {msg:?}");
     }
+
+    fn before_drop(&mut self, scope: &AudioWorkletGlobalScope) {
+        if !self.ended_triggered && scope.current_time >= self.start_time {
+            scope.send_ended_event();
+            self.ended_triggered = true;
+        }
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use float_eq::assert_float_eq;
     use std::f32::consts::PI;
 
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
+
     use crate::context::{BaseAudioContext, OfflineAudioContext};
     use crate::RENDER_QUANTUM_SIZE;
 
@@ -1458,4 +1468,72 @@ mod tests {
         src.stop();
         src.stop();
     }
+
+    #[test]
+    fn test_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_buffer_source();
+        src.start_at(0.);
+        src.stop_at(0.5);
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn test_no_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let src = context.create_constant_source();
+
+        // do not start the node
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(!ended.load(Ordering::Relaxed)); // should not have triggered
+    }
+
+    #[test]
+    fn test_exact_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_buffer_source();
+        src.start_at(0.);
+        src.stop_at(1.); // end right at the end of the offline buffer
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn test_implicit_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_buffer_source();
+        src.start_at(0.);
+        // no explicit stop, so we stop at end of offline context
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
 }

src/node/constant_source.rs

@@ -255,6 +255,13 @@ impl AudioProcessor for ConstantSourceRenderer {
 
         log::warn!("ConstantSourceRenderer: Dropping incoming message {msg:?}");
     }
+
+    fn before_drop(&mut self, scope: &AudioWorkletGlobalScope) {
+        if !self.ended_triggered && scope.current_time >= self.start_time {
+            scope.send_ended_event();
+            self.ended_triggered = true;
+        }
+    }
 }
 
 #[cfg(test)]
@@ -264,6 +271,9 @@ mod tests {
 
     use float_eq::assert_float_eq;
 
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
+
     use super::*;
 
     #[test]
@@ -369,4 +379,72 @@ mod tests {
         src.stop();
         src.stop();
     }
+
+    #[test]
+    fn test_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_constant_source();
+        src.start_at(0.);
+        src.stop_at(0.5);
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn test_no_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let src = context.create_constant_source();
+
+        // do not start the node
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(!ended.load(Ordering::Relaxed)); // should not have triggered
+    }
+
+    #[test]
+    fn test_exact_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_constant_source();
+        src.start_at(0.);
+        src.stop_at(1.); // end right at the end of the offline buffer
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn test_implicit_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_constant_source();
+        src.start_at(0.);
+        // no explicit stop, so we stop at end of offline context
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
 }

src/node/oscillator.rs

@@ -457,6 +457,13 @@ impl AudioProcessor for OscillatorRenderer {
 
         log::warn!("OscillatorRenderer: Dropping incoming message {msg:?}");
     }
+
+    fn before_drop(&mut self, scope: &AudioWorkletGlobalScope) {
+        if !self.ended_triggered && scope.current_time >= self.start_time {
+            scope.send_ended_event();
+            self.ended_triggered = true;
+        }
+    }
 }
 
 impl OscillatorRenderer {
@@ -608,6 +615,9 @@ mod tests {
     use float_eq::assert_float_eq;
     use std::f64::consts::PI;
 
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
+
     use crate::context::{BaseAudioContext, OfflineAudioContext};
     use crate::node::{AudioNode, AudioScheduledSourceNode};
     use crate::periodic_wave::{PeriodicWave, PeriodicWaveOptions};
@@ -1247,4 +1257,72 @@ mod tests {
         osc.stop();
         osc.stop();
     }
+
+    #[test]
+    fn test_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_oscillator();
+        src.start_at(0.);
+        src.stop_at(0.5);
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn test_no_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let src = context.create_oscillator();
+
+        // do not start the node
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(!ended.load(Ordering::Relaxed)); // should not have triggered
+    }
+
+    #[test]
+    fn test_exact_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_oscillator();
+        src.start_at(0.);
+        src.stop_at(1.); // end right at the end of the offline buffer
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
+
+    #[test]
+    fn test_implicit_ended_event() {
+        let mut context = OfflineAudioContext::new(2, 44_100, 44_100.);
+        let mut src = context.create_oscillator();
+        src.start_at(0.);
+        // no explicit stop, so we stop at end of offline context
+
+        let ended = Arc::new(AtomicBool::new(false));
+        let ended_clone = Arc::clone(&ended);
+        src.set_onended(move |_event| {
+            ended_clone.store(true, Ordering::Relaxed);
+        });
+
+        let _ = context.start_rendering_sync();
+        assert!(ended.load(Ordering::Relaxed));
+    }
 }

src/render/graph.rs

@@ -503,6 +503,7 @@ impl Graph {
                 let mut node = self.nodes.remove(*index).into_inner();
                 self.reclaim_id_channel
                     .push(node.reclaim_id.take().unwrap());
+                node.processor.before_drop(scope);
                 drop(node);
 
                 // And remove it from the ordering after we have processed all nodes
@@ -536,6 +537,13 @@ impl Graph {
         // Return the output buffer of destination node
         &self.nodes.get_unchecked_mut(AudioNodeId(0)).outputs[0]
     }
+
+    pub fn before_drop(&mut self, scope: &AudioWorkletGlobalScope) {
+        self.nodes.iter_mut().for_each(|(id, node)| {
+            scope.node_id.set(id);
+            node.get_mut().processor.before_drop(scope);
+        });
+    }
 }
 
 #[cfg(test)]

src/render/node_collection.rs

@@ -57,6 +57,14 @@ impl NodeCollection {
         self.nodes.iter_mut().filter_map(Option::as_mut)
     }
 
+    #[inline(always)]
+    pub fn iter_mut(&mut self) -> impl Iterator<Item = (AudioNodeId, &mut RefCell<Node>)> {
+        self.nodes
+            .iter_mut()
+            .enumerate()
+            .filter_map(|(i, v)| v.as_mut().map(|m| (AudioNodeId(i as u64), m)))
+    }
+
     #[inline(always)]
     pub fn contains(&self, index: AudioNodeId) -> bool {
         self.nodes[index.0 as usize].is_some()

src/render/processor.rs

@@ -172,6 +172,8 @@ pub trait AudioProcessor: Send {
     fn has_side_effects(&self) -> bool {
         false
     }
+
+    fn before_drop(&mut self, _scope: &AudioWorkletGlobalScope) {}
 }
 
 impl std::fmt::Debug for dyn AudioProcessor {