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#[derive(Debug, Clone, Copy, PartialEq)]
62pub struct ActivateInfo {
63 pub sample_rate: NonZeroU32,
65 pub max_block_frames: NonZeroU32,
67 pub num_stream_in_channels: u32,
69 pub num_stream_out_channels: u32,
71 pub input_to_output_latency_seconds: f64,
73}
74
75#[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 pub num_graph_inputs: ChannelCount,
82 pub num_graph_outputs: ChannelCount,
84
85 pub flags: FirewheelFlags,
89
90 pub initial_node_capacity: u32,
94 pub initial_edge_capacity: u32,
98 pub declick_seconds: f32,
103 pub initial_event_group_capacity: u32,
107 pub channel_capacity: u32,
111 pub event_queue_capacity: usize,
118 pub immediate_event_capacity: usize,
124 pub scheduled_event_capacity: usize,
134 pub buffer_out_of_space_mode: BufferOutOfSpaceMode,
138
139 pub logger_config: RealtimeLoggerConfig,
141
142 pub proc_store_capacity: usize,
146
147 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#[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 pub hard_clip_outputs: bool,
198
199 pub detect_clipping_on_output: bool,
208
209 pub validate_output_is_finite: bool,
215
216 pub force_clear_buffers: bool,
222
223 pub profile_engine_bookkeeping: bool,
229
230 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
282pub 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 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 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 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 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 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 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 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 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 #[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 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 pub fn heartbeat(&self) -> i64 {
604 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 #[cfg(feature = "scheduled_events")]
630 pub fn audio_clock(&self) -> AudioClock {
631 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 #[cfg(feature = "scheduled_events")]
671 pub fn audio_clock_corrected(&self) -> AudioClock {
672 let mut clock_borrowed = self.shared_clock_output.borrow_mut();
678 let clock = clock_borrowed.read();
679
680 if !self.is_active() {
681 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 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 #[cfg(feature = "scheduled_events")]
734 pub fn audio_clock_instant(&self) -> Option<Instant> {
735 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 #[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 #[cfg(feature = "musical_transport")]
770 pub fn transport_state(&self) -> &TransportState {
771 &self.transport_state
772 }
773
774 #[cfg(feature = "musical_transport")]
776 pub fn transport(&self) -> &TransportState {
777 &self.transport_state
778 }
779
780 pub fn flags(&self) -> &FirewheelFlags {
782 &self.config.flags
783 }
784
785 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 pub fn clipping_occurred(&self) -> bool {
806 self.shared_flags
807 .clipping_occurred
808 .swap(false, Ordering::Relaxed)
809 }
810
811 pub fn profiling_data(&mut self) -> &ProfilingData {
813 self.profiler_rx.fetch_info()
814 }
815
816 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 pub fn graph_in_node_id(&self) -> NodeID {
933 self.graph.graph_in_node()
934 }
935
936 pub fn graph_out_node_id(&self) -> NodeID {
938 self.graph.graph_out_node()
939 }
940
941 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 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 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 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 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 pub fn contains_node(&self, id: NodeID) -> bool {
1002 self.graph.contains_node(id)
1003 }
1004
1005 pub fn node_info(&self, id: NodeID) -> Option<&NodeEntry> {
1009 self.graph.node_info(id)
1010 }
1011
1012 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 pub fn node_state<T: 'static>(&self, id: NodeID) -> Option<&T> {
1023 self.graph.node_state(id)
1024 }
1025
1026 pub fn node_state_dyn(&self, id: NodeID) -> Option<&dyn Any> {
1030 self.graph.node_state_dyn(id)
1031 }
1032
1033 pub fn node_state_mut<T: 'static>(&mut self, id: NodeID) -> Option<&mut T> {
1037 self.graph.node_state_mut(id)
1038 }
1039
1040 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 pub fn nodes(&self) -> impl Iterator<Item = &NodeEntry> {
1049 self.graph.nodes()
1050 }
1051
1052 pub fn edges(&self) -> impl Iterator<Item = &Edge> {
1054 self.graph.edges()
1055 }
1056
1057 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 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 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 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 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 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 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 pub fn edge(&self, edge_id: EdgeID) -> Option<&Edge> {
1245 self.graph.edge(edge_id)
1246 }
1247
1248 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 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 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 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 #[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 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 #[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 #[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 #[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 #[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 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
1409pub 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 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#[cfg(feature = "scheduled_events")]
1463#[derive(Default, Debug, Clone, Copy, PartialEq)]
1464pub enum ClearScheduledEventsType {
1465 #[default]
1467 All,
1468 NonMusicalOnly,
1470 MusicalOnly,
1472}