Skip to main content

firewheel_graph/
context.rs

1use bevy_platform::sync::{
2    Arc,
3    atomic::{AtomicBool, Ordering},
4};
5use core::cell::RefCell;
6use core::error::Error;
7use core::num::NonZeroU32;
8use core::time::Duration;
9use core::{any::Any, f64};
10use firewheel_core::node::{NodeError, ProcStore};
11use firewheel_core::{
12    StreamInfo,
13    channel_config::{ChannelConfig, ChannelCount},
14    diff::EventQueue,
15    dsp::declick::DeclickValues,
16    event::{NodeEvent, NodeEventType},
17    node::{AudioNode, DynAudioNode, NodeID},
18};
19use firewheel_core::{
20    dsp::volume::Volume,
21    log::{RealtimeLogger, RealtimeLoggerConfig, RealtimeLoggerMainThread},
22};
23use ringbuf::traits::{Consumer, Producer, Split};
24use smallvec::SmallVec;
25
26#[cfg(not(feature = "std"))]
27use num_traits::Float;
28
29#[cfg(feature = "scheduled_events")]
30use bevy_platform::time::Instant;
31#[cfg(feature = "scheduled_events")]
32use firewheel_core::clock::{AudioClock, DurationSeconds};
33
34#[cfg(all(not(feature = "std"), feature = "musical_transport"))]
35use bevy_platform::prelude::Box;
36#[cfg(not(feature = "std"))]
37use bevy_platform::prelude::Vec;
38
39use crate::{
40    error::{
41        ActivateError, AddEdgeError, CompileGraphError, DeactivateError, RemoveNodeError,
42        UpdateError,
43    },
44    graph::{AudioGraph, Edge, EdgeID, NodeEntry, PortIdx},
45    processor::{
46        BufferOutOfSpaceMode, ContextToProcessorMsg, FirewheelProcessor, FirewheelProcessorConfig,
47        FirewheelProcessorInner, ProcessorToContextMsg, ProfilingData, SharedClock, SharedFlags,
48        profiling::{ProfilerRx, ProfilerTx},
49    },
50};
51
52#[cfg(feature = "scheduled_events")]
53use crate::processor::ClearScheduledEventsEvent;
54#[cfg(feature = "scheduled_events")]
55use firewheel_core::clock::EventInstant;
56
57#[cfg(feature = "musical_transport")]
58use firewheel_core::clock::TransportState;
59
60/// Information about the running audio stream.
61#[derive(Debug, Clone, Copy, PartialEq)]
62pub struct ActivateInfo {
63    /// The sample rate of the audio stream.
64    pub sample_rate: NonZeroU32,
65    /// The maximum number of frames that can appear in a single process cyle.
66    pub max_block_frames: NonZeroU32,
67    /// The number of input audio channels in the stream.
68    pub num_stream_in_channels: u32,
69    /// The number of output audio channels in the stream.
70    pub num_stream_out_channels: u32,
71    /// The latency of the input to output stream in seconds.
72    pub input_to_output_latency_seconds: f64,
73}
74
75/// The configuration of a Firewheel context.
76#[derive(Debug, Clone, Copy, PartialEq)]
77#[cfg_attr(feature = "bevy_reflect", derive(bevy_reflect::Reflect))]
78#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
79pub struct FirewheelConfig {
80    /// The number of input channels in the audio graph.
81    pub num_graph_inputs: ChannelCount,
82    /// The number of output channels in the audio graph.
83    pub num_graph_outputs: ChannelCount,
84
85    /// Extra configuration flags.
86    ///
87    /// By default, no flags are set.
88    pub flags: FirewheelFlags,
89
90    /// An initial capacity to allocate for the nodes in the audio graph.
91    ///
92    /// By default this is set to `128`.
93    pub initial_node_capacity: u32,
94    /// An initial capacity to allocate for the edges in the audio graph.
95    ///
96    /// By default this is set to `256`.
97    pub initial_edge_capacity: u32,
98    /// The amount of time in seconds to fade in/out when pausing/resuming
99    /// to avoid clicks and pops.
100    ///
101    /// By default this is set to `10.0 / 1_000.0`.
102    pub declick_seconds: f32,
103    /// The initial capacity for a group of events.
104    ///
105    /// By default this is set to `128`.
106    pub initial_event_group_capacity: u32,
107    /// The capacity of the engine's internal message channel.
108    ///
109    /// By default this is set to `64`.
110    pub channel_capacity: u32,
111    /// The maximum number of events that can be sent in a single call
112    /// to [`AudioNodeProcessor::process`].
113    ///
114    /// By default this is set to `128`.
115    ///
116    /// [`AudioNodeProcessor::process`]: firewheel_core::node::AudioNodeProcessor::process
117    pub event_queue_capacity: usize,
118    /// The maximum number of immediate events (events that do *NOT* have a
119    /// scheduled time component) that can be stored at once in the audio
120    /// thread.
121    ///
122    /// By default this is set to `512`.
123    pub immediate_event_capacity: usize,
124    /// The maximum number of scheduled events (events that have a scheduled
125    /// time component) that can be stored at once in the audio thread.
126    ///
127    /// This can be set to `0` to save some memory if you do not plan on using
128    /// scheduled events.
129    ///
130    /// This has no effect if the `scheduled_events` feature is disabled.
131    ///
132    /// By default this is set to `512`.
133    pub scheduled_event_capacity: usize,
134    /// How to handle event buffers on the audio thread running out of space.
135    ///
136    /// By default this is set to [`BufferOutOfSpaceMode::AllocateOnAudioThread`].
137    pub buffer_out_of_space_mode: BufferOutOfSpaceMode,
138
139    /// The configuration of the realtime safe logger.
140    pub logger_config: RealtimeLoggerConfig,
141
142    /// The initial number of slots to allocate for the [`ProcStore`].
143    ///
144    /// By default this is set to `8`.
145    pub proc_store_capacity: usize,
146
147    /// If `Some`, then inputs to the audio graph will be clamped to silence if the
148    /// max peak amplitude is less than the given volume. This can help improve the
149    /// performance of processing chains which use the graph inputs.
150    ///
151    /// If this is `None`, then no clamping will occur.
152    ///
153    /// Note, while this is functionally a noise gate, it is not a good noise gate,
154    /// and values above -70dB may cause audible clicking. If you need to increase
155    /// the threshold, it is recommended to instead use a dedicated noise gate node.
156    ///
157    /// By default this is set to `Some(Volume::Decibels(-70.0)`.
158    pub clamp_graph_inputs_below: Option<Volume>,
159}
160
161impl Default for FirewheelConfig {
162    fn default() -> Self {
163        Self {
164            num_graph_inputs: ChannelCount::ZERO,
165            num_graph_outputs: ChannelCount::STEREO,
166            flags: FirewheelFlags::default(),
167            initial_node_capacity: 128,
168            initial_edge_capacity: 256,
169            declick_seconds: DeclickValues::DEFAULT_FADE_SECONDS,
170            initial_event_group_capacity: 128,
171            channel_capacity: 64,
172            event_queue_capacity: 128,
173            immediate_event_capacity: 512,
174            scheduled_event_capacity: 512,
175            buffer_out_of_space_mode: BufferOutOfSpaceMode::AllocateOnAudioThread,
176            logger_config: RealtimeLoggerConfig::default(),
177            proc_store_capacity: 8,
178            clamp_graph_inputs_below: Some(Volume::Decibels(-70.0)),
179        }
180    }
181}
182
183/// Configuration flags for a [`FirewheelContext`]
184///
185/// Unlike [`FirewheelConfig`], these flags can be changed after the context has
186/// been created.
187#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
188#[cfg_attr(feature = "bevy_reflect", derive(bevy_reflect::Reflect))]
189#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
190pub struct FirewheelFlags {
191    /// Hard clip all samples in the final output buffer to the range `[-1.0, 1.0]`.
192    ///
193    /// This usually isn't necessary since the OS itself generally hard clips the
194    /// output, but it is here if you need it.
195    ///
196    /// By default this is set to `false`.
197    pub hard_clip_outputs: bool,
198
199    /// Detect when a sample in the final output buffer falls outside the range
200    /// `[-1.0, 1.0]`. If a sample falls outside this range, then
201    /// [`FirewheelContext::clipping_occurred`] will return `true`.
202    ///
203    /// This check takes place before hard clipping if the
204    /// [`FirewheelFlags::hard_clip_outputs`] is set to `true`.
205    ///
206    /// By default this is set to `false`.
207    pub detect_clipping_on_output: bool,
208
209    /// Validate that all samples in the final output buffer are a valid finite
210    /// number. If a non finite number is detected, then the sample will will be
211    /// set to `0.0` and an error is logged.
212    ///
213    /// By default this is set to `false`.
214    pub validate_output_is_finite: bool,
215
216    /// Force all of a node's output buffers to be cleared before processing.
217    /// This shouldn't be necessary, but it can be used to debug nodes that
218    /// are misusing the silence flag feature.
219    ///
220    /// By default this is set to `false`.
221    pub force_clear_buffers: bool,
222
223    /// Enables performance profiling for engine bookkeeping operations on the
224    /// audio thread such as message handling, event sorting, event searching,
225    /// and final output processing.
226    ///
227    /// By default this is set to `false`.
228    pub profile_engine_bookkeeping: bool,
229
230    /// Enable per-node performance profiling.
231    ///
232    /// This has no effect when the `node_profiling` feature is disabled.
233    ///
234    /// By default this is set to `false`.
235    pub profile_nodes: bool,
236}
237
238bitflags::bitflags! {
239    #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
240    pub struct FirewheelBitFlags: u32 {
241        const HARD_CLIP_OUTPUTS = 1 << 0;
242        const DETECT_CLIPPING_ON_OUTPUT = 1 << 1;
243        const VALIDATE_OUTPUT_IS_FINITE = 1 << 2;
244        const FORCE_CLEAR_BUFFERS = 1 << 3;
245        const PROFILE_ENGINE_BOOKKEEPING = 1 << 4;
246        const PROFILE_NODES = 1 << 5;
247    }
248}
249
250impl From<FirewheelFlags> for FirewheelBitFlags {
251    fn from(value: FirewheelFlags) -> Self {
252        let mut b = Self::empty();
253        b.set(Self::HARD_CLIP_OUTPUTS, value.hard_clip_outputs);
254        b.set(
255            Self::DETECT_CLIPPING_ON_OUTPUT,
256            value.detect_clipping_on_output,
257        );
258        b.set(
259            Self::VALIDATE_OUTPUT_IS_FINITE,
260            value.validate_output_is_finite,
261        );
262        b.set(Self::FORCE_CLEAR_BUFFERS, value.force_clear_buffers);
263        b.set(
264            Self::PROFILE_ENGINE_BOOKKEEPING,
265            value.profile_engine_bookkeeping,
266        );
267        b.set(Self::PROFILE_NODES, value.profile_nodes);
268        b
269    }
270}
271
272pub(crate) struct ProcessorChannel {
273    pub(crate) shared_flags: Arc<SharedFlags>,
274    pub(crate) from_context_rx: ringbuf::HeapCons<ContextToProcessorMsg>,
275    pub(crate) to_context_tx: ringbuf::HeapProd<ProcessorToContextMsg>,
276    pub(crate) logger: RealtimeLogger,
277    pub(crate) store: ProcStore,
278    pub(crate) profiler_tx: ProfilerTx,
279    pub(crate) shared_clock_input: triple_buffer::Input<SharedClock>,
280}
281
282/// A Firewheel context
283pub struct FirewheelContext {
284    graph: AudioGraph,
285
286    to_processor_tx: ringbuf::HeapProd<ContextToProcessorMsg>,
287    from_processor_rx: ringbuf::HeapCons<ProcessorToContextMsg>,
288    processor_drop_flag: Option<Arc<AtomicBool>>,
289    profiler_rx: ProfilerRx,
290    logger_rx: RealtimeLoggerMainThread,
291
292    pending_processor_channel: Option<ProcessorChannel>,
293    processor_drop_rx: Option<ringbuf::HeapCons<FirewheelProcessorInner>>,
294
295    shared_clock_output: RefCell<triple_buffer::Output<SharedClock>>,
296
297    sample_rate: NonZeroU32,
298    sample_rate_recip: f64,
299    stream_info: Option<StreamInfo>,
300    shared_flags: Arc<SharedFlags>,
301
302    #[cfg(feature = "musical_transport")]
303    transport_state: Box<TransportState>,
304    #[cfg(feature = "musical_transport")]
305    transport_state_alloc_reuse: Option<Box<TransportState>>,
306
307    // Re-use the allocations for groups of events.
308    event_group_pool: Vec<Vec<NodeEvent>>,
309    event_group: Vec<NodeEvent>,
310    initial_event_group_capacity: usize,
311
312    #[cfg(feature = "scheduled_events")]
313    queued_clear_scheduled_events: Vec<ClearScheduledEventsEvent>,
314
315    config: FirewheelConfig,
316}
317
318impl FirewheelContext {
319    /// Create a new Firewheel context.
320    pub fn new(config: FirewheelConfig) -> Self {
321        let (to_processor_tx, from_context_rx) =
322            ringbuf::HeapRb::<ContextToProcessorMsg>::new(config.channel_capacity as usize).split();
323        let (to_context_tx, from_processor_rx) =
324            ringbuf::HeapRb::<ProcessorToContextMsg>::new(config.channel_capacity as usize * 2)
325                .split();
326
327        let initial_event_group_capacity = config.initial_event_group_capacity as usize;
328        let mut event_group_pool = Vec::with_capacity(16);
329        for _ in 0..3 {
330            event_group_pool.push(Vec::with_capacity(initial_event_group_capacity));
331        }
332
333        let graph = AudioGraph::new(&config);
334
335        let (shared_clock_input, shared_clock_output) =
336            triple_buffer::triple_buffer(&SharedClock::default());
337
338        let (logger, logger_rx) = firewheel_core::log::realtime_logger(config.logger_config);
339        let (profiler_tx, profiler_rx) = crate::processor::profiling::profiler_channel(
340            config.initial_node_capacity as usize,
341            #[cfg(feature = "node_profiling")]
342            graph.graph_out_node(),
343        );
344        let shared_flags = Arc::new(SharedFlags::default());
345
346        let store = ProcStore::with_capacity(config.proc_store_capacity);
347
348        Self {
349            graph,
350            to_processor_tx,
351            from_processor_rx,
352            processor_drop_flag: None,
353            profiler_rx,
354            logger_rx,
355            pending_processor_channel: Some(ProcessorChannel {
356                shared_flags: Arc::clone(&shared_flags),
357                from_context_rx,
358                to_context_tx,
359                logger,
360                store,
361                profiler_tx,
362                shared_clock_input,
363            }),
364            processor_drop_rx: None,
365            shared_clock_output: RefCell::new(shared_clock_output),
366            sample_rate: NonZeroU32::new(44100).unwrap(),
367            sample_rate_recip: 44100.0f64.recip(),
368            stream_info: None,
369            shared_flags,
370            #[cfg(feature = "musical_transport")]
371            transport_state: Box::new(TransportState::default()),
372            #[cfg(feature = "musical_transport")]
373            transport_state_alloc_reuse: None,
374            event_group_pool,
375            event_group: Vec::with_capacity(initial_event_group_capacity),
376            initial_event_group_capacity,
377            #[cfg(feature = "scheduled_events")]
378            queued_clear_scheduled_events: Vec::new(),
379            config,
380        }
381    }
382
383    /// Try to modify the graph. If the given closure returns an error (or
384    /// if a cycle is detected), then any changes made to the graph inside
385    /// the closure will be reverted.
386    ///
387    /// Any custom error type can be used, though
388    /// [`ModifyGraphError`](crate::error::ModifyGraphError) is provided
389    /// for convenience.
390    pub fn try_modify_graph<E: Error>(
391        &mut self,
392        f: impl FnOnce(&mut Self) -> Result<(), E>,
393    ) -> Result<(), E> {
394        self.graph.begin_modify_guard();
395
396        let res = (f)(self);
397
398        self.graph.end_modify_guard(res.is_err());
399
400        res
401    }
402
403    /// Get an immutable reference to the processor store.
404    ///
405    /// If an audio stream is currently running, this will return `None`.
406    pub fn proc_store(&self) -> Option<&ProcStore> {
407        if let Some(proc_channel) = &self.pending_processor_channel {
408            Some(&proc_channel.store)
409        } else if let Some(processor) = self.processor_drop_rx.as_ref() {
410            if let Some(processor) = processor.last() {
411                if processor.poisoned {
412                    panic!("The audio thread has panicked!");
413                }
414
415                Some(&processor.extra.store)
416            } else {
417                None
418            }
419        } else {
420            None
421        }
422    }
423
424    /// Get a mutable reference to the processor store.
425    ///
426    /// If an audio stream is currently running, this will return `None`.
427    pub fn proc_store_mut(&mut self) -> Option<&mut ProcStore> {
428        if let Some(proc_channel) = &mut self.pending_processor_channel {
429            Some(&mut proc_channel.store)
430        } else if let Some(processor) = self.processor_drop_rx.as_mut() {
431            if let Some(processor) = processor.last_mut() {
432                if processor.poisoned {
433                    panic!("The audio thread has panicked!");
434                }
435
436                Some(&mut processor.extra.store)
437            } else {
438                None
439            }
440        } else {
441            None
442        }
443    }
444
445    /// Returns `true` if the context is currently active (the [`FirewheelProcessor`]
446    /// counterpart is still alive).
447    pub fn is_active(&self) -> bool {
448        if let Some(rx) = &self.processor_drop_rx {
449            rx.try_peek().is_none()
450        } else {
451            false
452        }
453    }
454
455    /// Activate the context with the given audio stream.
456    ///
457    /// Use [`FirewheelContext::is_active`] to check if the context is ready to
458    /// be activated.
459    ///
460    /// Note, in rare cases where the audio thread crashes without cleanly dropping
461    /// its contents, this may never succeed. Consider adding a timeout to avoid
462    /// deadlocking.
463    pub fn activate(&mut self, info: ActivateInfo) -> Result<FirewheelProcessor, ActivateError> {
464        let ActivateInfo {
465            sample_rate,
466            max_block_frames,
467            num_stream_in_channels,
468            num_stream_out_channels,
469            input_to_output_latency_seconds,
470        } = info;
471
472        if self.is_active() {
473            return Err(ActivateError::AlreadyActive);
474        }
475
476        let maybe_proc_channel = self.pending_processor_channel.take();
477
478        let prev_sample_rate = if maybe_proc_channel.is_some() {
479            sample_rate
480        } else {
481            self.sample_rate
482        };
483
484        let stream_info = StreamInfo {
485            sample_rate,
486            sample_rate_recip: (sample_rate.get() as f64).recip(),
487            prev_sample_rate,
488            max_block_frames,
489            num_stream_in_channels,
490            num_stream_out_channels,
491            input_to_output_latency_seconds,
492            declick_frames: NonZeroU32::new(
493                (self.config.declick_seconds * sample_rate.get() as f32).round() as u32,
494            )
495            .unwrap_or(NonZeroU32::MIN),
496        };
497
498        self.sample_rate = stream_info.sample_rate;
499        self.sample_rate_recip = stream_info.sample_rate_recip;
500
501        let schedule = self.graph.compile(&stream_info)?;
502
503        let (drop_tx, drop_rx) = ringbuf::HeapRb::<FirewheelProcessorInner>::new(1).split();
504
505        let processor = if let Some(proc_channel) = maybe_proc_channel {
506            FirewheelProcessorInner::new(
507                FirewheelProcessorConfig {
508                    flags: self.config.flags.into(),
509                    immediate_event_buffer_capacity: self.config.immediate_event_capacity,
510                    buffer_out_of_space_mode: self.config.buffer_out_of_space_mode,
511                    clamp_graph_inputs_below_amp: self
512                        .config
513                        .clamp_graph_inputs_below
514                        .map(|v| v.amp()),
515                    node_event_buffer_capacity: self.config.event_queue_capacity,
516                    #[cfg(feature = "scheduled_events")]
517                    scheduled_event_buffer_capacity: self.config.scheduled_event_capacity,
518                },
519                proc_channel,
520                &stream_info,
521            )
522        } else {
523            let mut processor = self.processor_drop_rx.as_mut().unwrap().try_pop().unwrap();
524
525            if processor.poisoned {
526                panic!("The audio thread has panicked!");
527            }
528
529            processor.new_stream(&stream_info);
530
531            processor
532        };
533
534        if self
535            .send_message_to_processor(ContextToProcessorMsg::NewSchedule(schedule))
536            .is_err()
537        {
538            panic!("Firewheel message channel is full!");
539        }
540
541        self.processor_drop_rx = Some(drop_rx);
542        self.stream_info = Some(stream_info);
543
544        let drop_flag = Arc::new(AtomicBool::new(false));
545        self.processor_drop_flag = Some(drop_flag.clone());
546
547        Ok(FirewheelProcessor::new(processor, drop_tx, drop_flag))
548    }
549
550    /// Request the context to be deactivated if it is active.
551    ///
552    /// This does not block the current thread. It might take a while for the
553    /// context to deactivate. Use [`FirewheelContext::is_active`] to check when
554    /// the context has been deactivated.
555    ///
556    /// Note, another way to deactivate the context is to drop the [`FirewheelProcessor`]
557    /// counterpart in the audio thread.
558    pub fn request_deactivate(&mut self) {
559        if let Some(flag) = &self.processor_drop_flag {
560            flag.store(true, Ordering::Relaxed);
561        }
562    }
563
564    /// If the context is active, request the context to be deactivated and wait
565    /// for the context to be deactivated before returning.
566    ///
567    /// If the `timeout` duration has been reached and the context is still not
568    /// deactivated, then an error is returned.
569    #[cfg(not(target_family = "wasm"))]
570    pub fn deactivate_blocking(&mut self, timeout: Duration) -> Result<(), DeactivateError> {
571        self.request_deactivate();
572
573        let now = bevy_platform::time::Instant::now();
574
575        while self.is_active() {
576            if now.elapsed() > timeout {
577                return Err(DeactivateError::TimedOut);
578            }
579
580            bevy_platform::thread::sleep(core::time::Duration::from_millis(1));
581        }
582
583        Ok(())
584    }
585
586    /// Information about the running audio stream.
587    ///
588    /// Returns `None` if the context is not currently active.
589    pub fn stream_info(&self) -> Option<&StreamInfo> {
590        if self.is_active() {
591            self.stream_info.as_ref()
592        } else {
593            None
594        }
595    }
596
597    /// Get the "heartbeat" of the processor. This is equal to the total number of samples
598    /// that the processor has processed.
599    ///
600    /// This can be used to detect when the processor stalls. For example, if this counter
601    /// doesn't change for half a second or so, then the processor is stalled, and events
602    /// should not be sent until it is resumed to avoid filling up the message channel.
603    pub fn heartbeat(&self) -> i64 {
604        // Reading the latest value of the clock doesn't meaningfully mutate
605        // state, so treat it as an immutable operation with interior mutability.
606        //
607        // PANIC SAFETY: This struct is the only place this is ever borrowed, so this
608        // will never panic.
609        let mut clock_borrowed = self.shared_clock_output.borrow_mut();
610        let clock = clock_borrowed.read();
611
612        clock.clock_samples.0
613    }
614
615    /// Get the current time of the audio clock, without accounting for the delay
616    /// between when the clock was last updated and now.
617    ///
618    /// For most use cases you probably want to use [`FirewheelContext::audio_clock_corrected`]
619    /// instead, but this method is provided if needed.
620    ///
621    /// Note, due to the nature of audio processing, this clock is is *NOT* synced with
622    /// the system's time (`Instant::now`). (Instead it is based on the amount of data
623    /// that has been processed.) For applications where the timing of audio events is
624    /// critical (i.e. a rhythm game), sync the game to this audio clock instead of the
625    /// OS's clock (`Instant::now()`).
626    ///
627    /// Note, calling this method is not super cheap, so avoid calling it many
628    /// times within the same game loop iteration if possible.
629    #[cfg(feature = "scheduled_events")]
630    pub fn audio_clock(&self) -> AudioClock {
631        // Reading the latest value of the clock doesn't meaningfully mutate
632        // state, so treat it as an immutable operation with interior mutability.
633        //
634        // PANIC SAFETY: This struct is the only place this is ever borrowed, so this
635        // will never panic.
636        let mut clock_borrowed = self.shared_clock_output.borrow_mut();
637        let clock = clock_borrowed.read();
638
639        AudioClock {
640            samples: clock.clock_samples,
641            seconds: clock
642                .clock_samples
643                .to_seconds(self.sample_rate, self.sample_rate_recip),
644            #[cfg(feature = "musical_transport")]
645            musical: clock.current_playhead,
646            #[cfg(feature = "musical_transport")]
647            transport_is_playing: clock.transport_is_playing,
648            update_instant: self.is_active().then_some(clock.update_instant),
649        }
650    }
651
652    /// Get the current time of the audio clock.
653    ///
654    /// Unlike, [`FirewheelContext::audio_clock`], this method accounts for the delay
655    /// between when the audio clock was last updated and now, leading to a more
656    /// accurate result for games and other applications.
657    ///
658    /// If the delay could not be determined (i.e. an audio stream is not currently
659    /// running), then this will assume there was no delay between when the audio
660    /// clock was last updated and now.
661    ///
662    /// Note, due to the nature of audio processing, this clock is is *NOT* synced with
663    /// the system's time (`Instant::now`). (Instead it is based on the amount of data
664    /// that has been processed.) For applications where the timing of audio events is
665    /// critical (i.e. a rhythm game), sync the game to this audio clock instead of the
666    /// OS's clock (`Instant::now()`).
667    ///
668    /// Note, calling this method is not super cheap, so avoid calling it many
669    /// times within the same game loop iteration if possible.
670    #[cfg(feature = "scheduled_events")]
671    pub fn audio_clock_corrected(&self) -> AudioClock {
672        // Reading the latest value of the clock doesn't meaningfully mutate
673        // state, so treat it as an immutable operation with interior mutability.
674        //
675        // PANIC SAFETY: This struct is the only place this is ever borrowed, so this
676        // will never panic.
677        let mut clock_borrowed = self.shared_clock_output.borrow_mut();
678        let clock = clock_borrowed.read();
679
680        if !self.is_active() {
681            // The audio thread is not currently running, so just return the
682            // latest value of the clock.
683            return AudioClock {
684                samples: clock.clock_samples,
685                seconds: clock
686                    .clock_samples
687                    .to_seconds(self.sample_rate, self.sample_rate_recip),
688                #[cfg(feature = "musical_transport")]
689                musical: clock.current_playhead,
690                #[cfg(feature = "musical_transport")]
691                transport_is_playing: clock.transport_is_playing,
692                update_instant: None,
693            };
694        }
695
696        let update_instant = clock.update_instant;
697        let delay = update_instant.elapsed();
698
699        // Account for the delay between when the clock was last updated and now.
700        let delta_seconds = DurationSeconds(delay.as_secs_f64());
701
702        let samples = clock.clock_samples + delta_seconds.to_samples(self.sample_rate);
703
704        #[cfg(feature = "musical_transport")]
705        let musical = clock.current_playhead.map(|musical_time| {
706            if clock.transport_is_playing
707                && let Some(transport) = &self.transport_state.transport
708            {
709                transport.delta_seconds_from(musical_time, delta_seconds, clock.speed_multiplier)
710            } else {
711                musical_time
712            }
713        });
714
715        AudioClock {
716            samples,
717            seconds: samples.to_seconds(self.sample_rate, self.sample_rate_recip),
718            #[cfg(feature = "musical_transport")]
719            musical,
720            #[cfg(feature = "musical_transport")]
721            transport_is_playing: clock.transport_is_playing,
722            update_instant: Some(update_instant),
723        }
724    }
725
726    /// Get the instant the audio clock was last updated.
727    ///
728    /// If the audio thread is not currently running, or if the instant could not
729    /// be determined for any other reason, then this will return `None`.
730    ///
731    /// Note, calling this method is not super cheap, so avoid calling it many
732    /// times within the same game loop iteration if possible.
733    #[cfg(feature = "scheduled_events")]
734    pub fn audio_clock_instant(&self) -> Option<Instant> {
735        // Reading the latest value of the clock doesn't meaningfully mutate
736        // state, so treat it as an immutable operation with interior mutability.
737        //
738        // PANIC SAFETY: This struct is the only place this is ever borrowed, so this
739        // will never panic.
740        let mut clock_borrowed = self.shared_clock_output.borrow_mut();
741        let clock = clock_borrowed.read();
742
743        self.is_active().then_some(clock.update_instant)
744    }
745
746    /// Sync the state of the musical transport.
747    ///
748    /// If the message channel is full, then this will return an error.
749    #[cfg(feature = "musical_transport")]
750    pub fn sync_transport(&mut self, transport: &TransportState) -> Result<(), UpdateError> {
751        if &*self.transport_state != transport {
752            let transport_msg = if let Some(mut t) = self.transport_state_alloc_reuse.take() {
753                *t = transport.clone();
754                t
755            } else {
756                Box::new(transport.clone())
757            };
758
759            self.send_message_to_processor(ContextToProcessorMsg::SetTransportState(transport_msg))
760                .map_err(|(_, e)| e)?;
761
762            *self.transport_state = transport.clone();
763        }
764
765        Ok(())
766    }
767
768    /// Get the current transport state.
769    #[cfg(feature = "musical_transport")]
770    pub fn transport_state(&self) -> &TransportState {
771        &self.transport_state
772    }
773
774    /// Get the current transport state.
775    #[cfg(feature = "musical_transport")]
776    pub fn transport(&self) -> &TransportState {
777        &self.transport_state
778    }
779
780    /// The current configuration flags being used by this context.
781    pub fn flags(&self) -> &FirewheelFlags {
782        &self.config.flags
783    }
784
785    /// Set the configuration flags.
786    ///
787    /// This can be set while the context is active or inactive.
788    ///
789    /// If the message channel is full, then this will return an error.
790    pub fn set_flags(&mut self, flags: FirewheelFlags) -> Result<(), UpdateError> {
791        if self.config.flags == flags {
792            return Ok(());
793        }
794        self.config.flags = flags;
795
796        self.send_message_to_processor(ContextToProcessorMsg::SetFlags(flags.into()))
797            .map_err(|(_, e)| e)
798    }
799
800    /// Returns `true` if both the `FirewheelFlags::VALIDATE_OUTPUT_DOES_NOT_CLIP`
801    /// flag is set and a sample in the final output buffer fell outside the range
802    /// `[-1.0, 1.0]`.
803    ///
804    /// Calling this method resets the internal flag.
805    pub fn clipping_occurred(&self) -> bool {
806        self.shared_flags
807            .clipping_occurred
808            .swap(false, Ordering::Relaxed)
809    }
810
811    /// Retrieve the latest performance profiling data.
812    pub fn profiling_data(&mut self) -> &ProfilingData {
813        self.profiler_rx.fetch_info()
814    }
815
816    /// Update the firewheel context.
817    ///
818    /// This must be called regularly (i.e. once every frame).
819    pub fn update(&mut self) -> Result<(), UpdateError> {
820        self.logger_rx.flush(
821            |msg| {
822                #[cfg(feature = "tracing")]
823                tracing::error!("{}", msg);
824
825                #[cfg(all(feature = "log", not(feature = "tracing")))]
826                log::error!("{}", msg);
827
828                let _ = msg;
829            },
830            |msg| {
831                #[cfg(feature = "tracing")]
832                tracing::debug!("{}", msg);
833
834                #[cfg(all(feature = "log", not(feature = "tracing")))]
835                log::debug!("{}", msg);
836
837                let _ = msg;
838            },
839        );
840
841        firewheel_core::collector::GlobalRtGc::collect();
842
843        for msg in self.from_processor_rx.pop_iter() {
844            match msg {
845                ProcessorToContextMsg::DropEventGroup(mut event_group) => {
846                    event_group.clear();
847                    self.event_group_pool.push(event_group);
848                }
849                ProcessorToContextMsg::DropSchedule(schedule_data) => {
850                    self.graph.drop_old_schedule_data(schedule_data);
851                }
852                #[cfg(feature = "musical_transport")]
853                ProcessorToContextMsg::DropTransportState(transport_state) => {
854                    if self.transport_state_alloc_reuse.is_none() {
855                        self.transport_state_alloc_reuse = Some(transport_state);
856                    }
857                }
858                #[cfg(feature = "scheduled_events")]
859                ProcessorToContextMsg::DropClearScheduledEvents(msgs) => {
860                    let _ = msgs;
861                }
862            }
863        }
864
865        self.graph
866            .update(self.stream_info.as_ref(), &mut self.event_group);
867
868        if self.is_active() {
869            if self.graph.needs_compile() {
870                let schedule_data = self.graph.compile(self.stream_info.as_ref().unwrap())?;
871
872                if let Err((msg, e)) = self
873                    .send_message_to_processor(ContextToProcessorMsg::NewSchedule(schedule_data))
874                {
875                    let ContextToProcessorMsg::NewSchedule(schedule) = msg else {
876                        unreachable!();
877                    };
878
879                    self.graph.on_schedule_send_failed(schedule);
880
881                    return Err(e);
882                }
883            }
884
885            #[cfg(feature = "scheduled_events")]
886            if !self.queued_clear_scheduled_events.is_empty() {
887                let msgs: SmallVec<[ClearScheduledEventsEvent; 1]> =
888                    self.queued_clear_scheduled_events.drain(..).collect();
889
890                if let Err((msg, e)) = self
891                    .send_message_to_processor(ContextToProcessorMsg::ClearScheduledEvents(msgs))
892                {
893                    let ContextToProcessorMsg::ClearScheduledEvents(mut msgs) = msg else {
894                        unreachable!();
895                    };
896
897                    self.queued_clear_scheduled_events = msgs.drain(..).collect();
898
899                    return Err(e);
900                }
901            }
902
903            if !self.event_group.is_empty() {
904                let mut next_event_group = self
905                    .event_group_pool
906                    .pop()
907                    .unwrap_or_else(|| Vec::with_capacity(self.initial_event_group_capacity));
908                core::mem::swap(&mut next_event_group, &mut self.event_group);
909
910                if let Err((msg, e)) = self
911                    .send_message_to_processor(ContextToProcessorMsg::EventGroup(next_event_group))
912                {
913                    let ContextToProcessorMsg::EventGroup(mut event_group) = msg else {
914                        unreachable!();
915                    };
916
917                    core::mem::swap(&mut event_group, &mut self.event_group);
918                    self.event_group_pool.push(event_group);
919
920                    return Err(e);
921                }
922            }
923        } else {
924            self.stream_info = None;
925            self.graph.deactivate();
926        }
927
928        Ok(())
929    }
930
931    /// The ID of the graph input node
932    pub fn graph_in_node_id(&self) -> NodeID {
933        self.graph.graph_in_node()
934    }
935
936    /// The ID of the graph output node
937    pub fn graph_out_node_id(&self) -> NodeID {
938        self.graph.graph_out_node()
939    }
940
941    /// Add a node to the audio graph.
942    pub fn add_node<T: AudioNode + 'static>(
943        &mut self,
944        node: T,
945        config: Option<T::Configuration>,
946    ) -> Result<NodeID, NodeError> {
947        self.graph.add_node(node, config)
948    }
949
950    /// Add a node to the audio graph which implements the type-erased [`DynAudioNode`] trait.
951    pub fn add_dyn_node<T: DynAudioNode + 'static>(
952        &mut self,
953        node: T,
954    ) -> Result<NodeID, NodeError> {
955        self.graph.add_dyn_node(node)
956    }
957
958    /// Add a node to the audio graph with the given bypass state.
959    pub fn add_node_bypassed<T: AudioNode + 'static>(
960        &mut self,
961        node: T,
962        config: Option<T::Configuration>,
963        bypassed: bool,
964    ) -> Result<NodeID, NodeError> {
965        let node_id = self.add_node(node, config)?;
966        if bypassed {
967            self.queue_event_for(node_id, NodeEventType::SetBypassed(true));
968        }
969        Ok(node_id)
970    }
971
972    /// Add a node with the given bypass state to the audio graph which implements
973    /// the type-erased [`DynAudioNode`] trait.
974    pub fn add_dyn_node_bypassed<T: DynAudioNode + 'static>(
975        &mut self,
976        node: T,
977        bypassed: bool,
978    ) -> Result<NodeID, NodeError> {
979        let node_id = self.graph.add_dyn_node(node)?;
980        if bypassed {
981            self.queue_event_for(node_id, NodeEventType::SetBypassed(true));
982        }
983        Ok(node_id)
984    }
985
986    /// Remove the given node from the audio graph.
987    ///
988    /// This will automatically remove all edges from the graph that
989    /// were connected to this node.
990    ///
991    /// On success, this returns a list of all edges that were removed
992    /// from the graph as a result of removing this node.
993    ///
994    /// This will return an error if the ID is of the graph input or graph
995    /// output node.
996    pub fn remove_node(&mut self, node_id: NodeID) -> Result<SmallVec<[Edge; 4]>, RemoveNodeError> {
997        self.graph.remove_node(node_id, false)
998    }
999
1000    /// Returns `true` if the node exists in the graph.
1001    pub fn contains_node(&self, id: NodeID) -> bool {
1002        self.graph.contains_node(id)
1003    }
1004
1005    /// Get information about a node in the graph.
1006    ///
1007    /// If the node does not exist in the graph, then `None` will be returned.
1008    pub fn node_info(&self, id: NodeID) -> Option<&NodeEntry> {
1009        self.graph.node_info(id)
1010    }
1011
1012    /// Get the [`ChannelConfig`] of a node in the graph.
1013    ///
1014    /// If the node does not exist in the graph, then `None` will be returned.
1015    pub fn node_channel_config(&self, id: NodeID) -> Option<ChannelConfig> {
1016        self.graph.node_info(id).map(|n| n.info.channel_config)
1017    }
1018
1019    /// Get an immutable reference to the custom state of a node.
1020    ///
1021    /// If the node does not exist in the graph, then `None` will be returned.
1022    pub fn node_state<T: 'static>(&self, id: NodeID) -> Option<&T> {
1023        self.graph.node_state(id)
1024    }
1025
1026    /// Get a type-erased, immutable reference to the custom state of a node.
1027    ///
1028    /// If the node does not exist in the graph, then `None` will be returned.
1029    pub fn node_state_dyn(&self, id: NodeID) -> Option<&dyn Any> {
1030        self.graph.node_state_dyn(id)
1031    }
1032
1033    /// Get a mutable reference to the custom state of a node.
1034    ///
1035    /// If the node does not exist in the graph, then `None` will be returned.
1036    pub fn node_state_mut<T: 'static>(&mut self, id: NodeID) -> Option<&mut T> {
1037        self.graph.node_state_mut(id)
1038    }
1039
1040    /// Get a type-erased, mutable reference to the custom state of a node.
1041    ///
1042    /// If the node does not exist in the graph, then `None` will be returned.
1043    pub fn node_state_dyn_mut(&mut self, id: NodeID) -> Option<&mut dyn Any> {
1044        self.graph.node_state_dyn_mut(id)
1045    }
1046
1047    /// Get a list of all the existing nodes in the graph.
1048    pub fn nodes(&self) -> impl Iterator<Item = &NodeEntry> {
1049        self.graph.nodes()
1050    }
1051
1052    /// Get a list of all the existing edges in the graph.
1053    pub fn edges(&self) -> impl Iterator<Item = &Edge> {
1054        self.graph.edges()
1055    }
1056
1057    /// Set the number of input and output channels to and from the audio graph.
1058    ///
1059    /// Returns the list of edges that were removed.
1060    pub fn set_graph_channel_config(
1061        &mut self,
1062        channel_config: ChannelConfig,
1063    ) -> SmallVec<[Edge; 4]> {
1064        self.graph.set_graph_channel_config(channel_config, false)
1065    }
1066
1067    /// Add connections (edges) between two nodes to the graph.
1068    ///
1069    /// * `src_node` - The ID of the source node.
1070    /// * `dst_node` - The ID of the destination node.
1071    /// * `ports_src_dst` - The port indices for each connection to make,
1072    ///   where the first value in a tuple is the output port on `src_node`,
1073    ///   and the second value in that tuple is the input port on `dst_node`.
1074    /// * `check_for_cycles` - If `true`, then this will run a check to
1075    ///   see if adding these edges will create a cycle in the graph, and
1076    ///   return an error if it does. Note, checking for cycles can be quite
1077    ///   expensive, so avoid enabling this when calling this method many times
1078    ///   in a row.
1079    ///
1080    /// If successful, then this returns a list of edge IDs in order.
1081    ///
1082    /// If this returns an error, then the audio graph has not been
1083    /// modified.
1084    pub fn connect(
1085        &mut self,
1086        src_node: NodeID,
1087        dst_node: NodeID,
1088        ports_src_dst: &[(PortIdx, PortIdx)],
1089        check_for_cycles: bool,
1090    ) -> Result<SmallVec<[EdgeID; 4]>, AddEdgeError> {
1091        self.graph
1092            .connect(src_node, dst_node, ports_src_dst, check_for_cycles, false)
1093    }
1094
1095    /// Connect two nodes in the graph, connecting output port 0 to input port
1096    /// 0, output port 1 to input port 1, etc.
1097    ///
1098    /// If the number of output ports on `src_node` does not equal the number
1099    /// of input ports on `dst_node`, then only the first valid ports will be
1100    /// connected.
1101    ///
1102    /// If successful, then this returns a list of edge IDs in order.
1103    ///
1104    /// If this returns an error, then the audio graph has not been
1105    /// modified.
1106    pub fn auto_connect(
1107        &mut self,
1108        src_node: NodeID,
1109        dst_node: NodeID,
1110        check_for_cycles: bool,
1111    ) -> Result<SmallVec<[EdgeID; 4]>, AddEdgeError> {
1112        let num_src_out_ports = self
1113            .node_info(src_node)
1114            .ok_or(AddEdgeError::SrcNodeNotFound(src_node))?
1115            .info
1116            .channel_config
1117            .num_outputs
1118            .get();
1119        let num_dst_in_ports = self
1120            .node_info(dst_node)
1121            .ok_or(AddEdgeError::DstNodeNotFound(dst_node))?
1122            .info
1123            .channel_config
1124            .num_inputs
1125            .get();
1126        let num_connect_ports = num_src_out_ports.min(num_dst_in_ports);
1127
1128        let ports_src_dst: SmallVec<[(u32, u32); 4]> =
1129            (0..num_connect_ports).map(|i| (i, i)).collect();
1130
1131        self.graph
1132            .connect(src_node, dst_node, &ports_src_dst, check_for_cycles, false)
1133    }
1134
1135    /// Connect the first two output ports of a node to the first two input
1136    /// ports of a second node.
1137    ///
1138    /// * `src_node` - The ID of the source node.
1139    /// * `dst_node` - The ID of the destination node.
1140    /// * `check_for_cycles` - If `true`, then this will run a check to
1141    ///   see if adding these edges will create a cycle in the graph, and
1142    ///   return an error if it does. Note, checking for cycles can be quite
1143    ///   expensive, so avoid enabling this when calling this method many times
1144    ///   in a row.
1145    ///
1146    /// ## Behavior
1147    ///
1148    /// * If `num_out_ports_on_src_node >= 2 && num_in_ports_on_dst_node >= 2`,
1149    ///   then src port 0 will be connected to dst port 0, and src port 1 will be
1150    ///   connected to dst port 1.
1151    /// * If `num_out_ports_on_src_node == 1 && num_in_ports_on_dst_node >= 2`,
1152    ///   then src port 0 will be connected to both dst port 0 and dst port 1.
1153    /// * In all other cases, an error will be returned. (Note that converting
1154    ///   a stereo signal into a mono signal should be done with the
1155    ///   `StereoToMonoNode`.)
1156    ///
1157    /// If successful, then this returns a list of edge IDs in order.
1158    ///
1159    /// If this returns an error, then the audio graph has not been
1160    /// modified.
1161    pub fn connect_stereo(
1162        &mut self,
1163        src_node: NodeID,
1164        dst_node: NodeID,
1165        check_for_cycles: bool,
1166    ) -> Result<SmallVec<[EdgeID; 4]>, AddEdgeError> {
1167        let num_src_out_ports = self
1168            .node_info(src_node)
1169            .ok_or(AddEdgeError::SrcNodeNotFound(src_node))?
1170            .info
1171            .channel_config
1172            .num_outputs;
1173        let num_dst_in_ports = self
1174            .node_info(dst_node)
1175            .ok_or(AddEdgeError::DstNodeNotFound(dst_node))?
1176            .info
1177            .channel_config
1178            .num_inputs;
1179
1180        let ports_src_dst = if num_src_out_ports.get() >= 2 && num_dst_in_ports.get() >= 2 {
1181            &[(0, 0), (1, 1)]
1182        } else if num_src_out_ports.get() == 1 && num_dst_in_ports.get() >= 2 {
1183            &[(0, 0), (0, 1)]
1184        } else {
1185            return Err(if num_dst_in_ports.get() < 2 {
1186                AddEdgeError::InPortOutOfRange {
1187                    node: dst_node,
1188                    port_idx: 1,
1189                    num_in_ports: num_dst_in_ports,
1190                }
1191            } else {
1192                AddEdgeError::InPortOutOfRange {
1193                    node: src_node,
1194                    port_idx: 0,
1195                    num_in_ports: num_src_out_ports,
1196                }
1197            });
1198        };
1199
1200        self.graph
1201            .connect(src_node, dst_node, ports_src_dst, check_for_cycles, false)
1202    }
1203
1204    /// Remove connections (edges) between two nodes from the graph.
1205    ///
1206    /// * `src_node` - The ID of the source node.
1207    /// * `dst_node` - The ID of the destination node.
1208    /// * `ports_src_dst` - The port indices for each connection to make,
1209    ///   where the first value in a tuple is the output port on `src_node`,
1210    ///   and the second value in that tuple is the input port on `dst_node`.
1211    ///
1212    /// Returns the list of edges that were successfully removed.
1213    pub fn disconnect(
1214        &mut self,
1215        src_node: NodeID,
1216        dst_node: NodeID,
1217        ports_src_dst: &[(PortIdx, PortIdx)],
1218    ) -> SmallVec<[Edge; 4]> {
1219        self.graph.disconnect(src_node, dst_node, ports_src_dst)
1220    }
1221
1222    /// Remove all connections (edges) between two nodes in the graph.
1223    ///
1224    /// * `src_node` - The ID of the source node.
1225    /// * `dst_node` - The ID of the destination node.
1226    ///
1227    /// Returns the list of edges that were successfully removed.
1228    pub fn disconnect_all_between(
1229        &mut self,
1230        src_node: NodeID,
1231        dst_node: NodeID,
1232    ) -> SmallVec<[Edge; 4]> {
1233        self.graph.disconnect_all_between(src_node, dst_node)
1234    }
1235
1236    /// Remove a connection (edge) via the edge's unique ID.
1237    ///
1238    /// If the edge did not exist in this graph, then `None` will be returned.
1239    pub fn disconnect_by_edge_id(&mut self, edge_id: EdgeID) -> Option<Edge> {
1240        self.graph.disconnect_by_edge_id(edge_id, false)
1241    }
1242
1243    /// Get information about the given [Edge]
1244    pub fn edge(&self, edge_id: EdgeID) -> Option<&Edge> {
1245        self.graph.edge(edge_id)
1246    }
1247
1248    /// Runs a check to see if a cycle exists in the audio graph. If a cycle
1249    /// exists, an error is returned.
1250    ///
1251    /// Note, this method is expensive.
1252    pub fn cycle_detected(&mut self) -> Result<(), CompileGraphError> {
1253        if self.graph.cycle_detected() {
1254            Err(CompileGraphError::CycleDetected)
1255        } else {
1256            Ok(())
1257        }
1258    }
1259
1260    /// Queue an event to be sent to an audio node's processor.
1261    ///
1262    /// Note, this event will not be sent until the event queue is flushed
1263    /// in [`FirewheelContext::update`].
1264    pub fn queue_event(&mut self, event: NodeEvent) {
1265        if self.contains_node(event.node_id) {
1266            self.event_group.push(event);
1267        }
1268    }
1269
1270    /// Queue an event to be sent to an audio node's processor.
1271    ///
1272    /// Note, this event will not be sent until the event queue is flushed
1273    /// in [`FirewheelContext::update`].
1274    pub fn queue_event_for(&mut self, node_id: NodeID, event: NodeEventType) {
1275        self.queue_event(NodeEvent {
1276            node_id,
1277            #[cfg(feature = "scheduled_events")]
1278            time: None,
1279            event,
1280        });
1281    }
1282
1283    /// Queue a [`NodeEventType::SetBypassed`] event for the given node.
1284    pub fn queue_bypassed_for(&mut self, node_id: NodeID, bypassed: bool) {
1285        self.queue_event(NodeEvent {
1286            node_id,
1287            #[cfg(feature = "scheduled_events")]
1288            time: None,
1289            event: NodeEventType::SetBypassed(bypassed),
1290        });
1291    }
1292
1293    /// Queue an event at a certain time, to be sent to an audio node's processor.
1294    ///
1295    /// If `time` is `None`, then the event will occur as soon as the node's
1296    /// processor receives the event.
1297    ///
1298    /// Note, this event will not be sent until the event queue is flushed
1299    /// in [`FirewheelContext::update`].
1300    #[cfg(feature = "scheduled_events")]
1301    pub fn schedule_event_for(
1302        &mut self,
1303        node_id: NodeID,
1304        event: NodeEventType,
1305        time: Option<EventInstant>,
1306    ) {
1307        self.queue_event(NodeEvent {
1308            node_id,
1309            time,
1310            event,
1311        });
1312    }
1313
1314    /// Construct a [`ContextQueue`] for diffing.
1315    ///
1316    /// Returns `None` if the node does not exist in the graph.
1317    pub fn event_queue(&mut self, id: NodeID) -> ContextQueue<'_> {
1318        ContextQueue {
1319            context: self,
1320            id,
1321            #[cfg(feature = "scheduled_events")]
1322            time: None,
1323        }
1324    }
1325
1326    /// Construct a [`ContextQueue`] for diffing, along with the event instant that
1327    /// any pushed events will be scheduled for.
1328    ///
1329    /// Returns `None` if the node does not exist in the graph.
1330    #[cfg(feature = "scheduled_events")]
1331    pub fn event_queue_scheduled(
1332        &mut self,
1333        id: NodeID,
1334        time: Option<EventInstant>,
1335    ) -> ContextQueue<'_> {
1336        ContextQueue {
1337            context: self,
1338            id,
1339            time,
1340        }
1341    }
1342
1343    /// Cancel scheduled events for all nodes.
1344    ///
1345    /// This will clear all events that have been scheduled since the last call to
1346    /// [`FirewheelContext::update`]. Any events scheduled between then and the next call
1347    /// to [`FirewheelContext::update`] will not be canceled.
1348    ///
1349    /// This only takes effect once [`FirewheelContext::update`] is called.
1350    #[cfg(feature = "scheduled_events")]
1351    pub fn cancel_all_scheduled_events(&mut self, event_type: ClearScheduledEventsType) {
1352        self.queued_clear_scheduled_events
1353            .push(ClearScheduledEventsEvent {
1354                node_id: None,
1355                event_type,
1356            });
1357    }
1358
1359    /// Cancel scheduled events for a specific node.
1360    ///
1361    /// This will clear all events that have been scheduled since the last call to
1362    /// [`FirewheelContext::update`]. Any events scheduled between then and the next call
1363    /// to [`FirewheelContext::update`] will not be canceled.
1364    ///
1365    /// This only takes effect once [`FirewheelContext::update`] is called.
1366    #[cfg(feature = "scheduled_events")]
1367    pub fn cancel_scheduled_events_for(
1368        &mut self,
1369        node_id: NodeID,
1370        event_type: ClearScheduledEventsType,
1371    ) {
1372        self.queued_clear_scheduled_events
1373            .push(ClearScheduledEventsEvent {
1374                node_id: Some(node_id),
1375                event_type,
1376            });
1377    }
1378
1379    fn send_message_to_processor(
1380        &mut self,
1381        msg: ContextToProcessorMsg,
1382    ) -> Result<(), (ContextToProcessorMsg, UpdateError)> {
1383        self.to_processor_tx
1384            .try_push(msg)
1385            .map_err(|msg| (msg, UpdateError::MsgChannelFull))
1386    }
1387}
1388
1389impl Drop for FirewheelContext {
1390    fn drop(&mut self) {
1391        // Wait for the processor to be drop to avoid deallocating it on
1392        // the audio thread.
1393        #[cfg(not(target_family = "wasm"))]
1394        let _ = self.deactivate_blocking(core::time::Duration::from_secs(3));
1395
1396        #[cfg(target_family = "wasm")]
1397        self.request_deactivate();
1398
1399        // Make sure all node processors are dropped before node states in
1400        // order to be compatible with CLAP plugin hosting.
1401        if let Some(p) = &mut self.processor_drop_rx {
1402            p.clear();
1403        }
1404        self.from_processor_rx.clear();
1405        firewheel_core::collector::GlobalRtGc::collect();
1406    }
1407}
1408
1409/// An event queue acquired from [`FirewheelContext::event_queue`].
1410///
1411/// This can help reduce event queue allocations
1412/// when you have direct access to the context.
1413///
1414/// ```
1415/// # use firewheel_core::{diff::{Diff, PathBuilder}, node::NodeID};
1416/// # use firewheel_graph::{backend::AudioBackend, FirewheelContext, ContextQueue};
1417/// # fn context_queue<B: AudioBackend, D: Diff>(
1418/// #     context: &mut FirewheelContext,
1419/// #     node_id: NodeID,
1420/// #     params: &D,
1421/// #     baseline: &D,
1422/// # ) {
1423/// // Get a queue that will send events directly to the provided node.
1424/// let mut queue = context.event_queue(node_id);
1425/// // Perform diffing using this queue.
1426/// params.diff(baseline, PathBuilder::default(), &mut queue);
1427/// # }
1428/// ```
1429pub struct ContextQueue<'a> {
1430    pub context: &'a mut FirewheelContext,
1431    id: NodeID,
1432    #[cfg(feature = "scheduled_events")]
1433    time: Option<EventInstant>,
1434}
1435
1436impl ContextQueue<'_> {
1437    /// Send an event to set the bypass state of the node.
1438    pub fn push_bypassed(&mut self, bypassed: bool) {
1439        self.push(NodeEventType::SetBypassed(bypassed));
1440    }
1441}
1442
1443#[cfg(feature = "scheduled_events")]
1444impl ContextQueue<'_> {
1445    pub fn time(&self) -> Option<EventInstant> {
1446        self.time
1447    }
1448}
1449
1450impl EventQueue for ContextQueue<'_> {
1451    fn push(&mut self, data: NodeEventType) {
1452        self.context.queue_event(NodeEvent {
1453            event: data,
1454            #[cfg(feature = "scheduled_events")]
1455            time: self.time,
1456            node_id: self.id,
1457        });
1458    }
1459}
1460
1461/// The type of scheduled events to clear
1462#[cfg(feature = "scheduled_events")]
1463#[derive(Default, Debug, Clone, Copy, PartialEq)]
1464pub enum ClearScheduledEventsType {
1465    /// Clear both musical and non-musical scheduled events.
1466    #[default]
1467    All,
1468    /// Clear only non-musical scheduled events.
1469    NonMusicalOnly,
1470    /// Clear only musical scheduled events.
1471    MusicalOnly,
1472}