// Copyright 2026 International Digital Economy Academy
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
///|
let threshold : Int = 512
///|
fn threshold_for_channels(channels : ChannelCount) -> Int {
let ch = if channels > 0 { channels } else { 1 }
(threshold + ch - 1) / ch * ch
}
///|
pub struct QueueSignal {
done : Ref[Bool]
}
///|
pub fn QueueSignal::is_done(self : QueueSignal) -> Bool {
self.done.val
}
///|
pub fn QueueSignal::mark_done(self : QueueSignal) -> Unit {
self.done.val = true
}
///|
struct QueuedSource {
source : DynSource
signal : QueueSignal?
}
///|
pub struct SourcesQueueInput {
next_sounds : Ref[Array[QueuedSource]]
keep_alive_if_empty : Ref[Bool]
}
///|
pub struct SourcesQueueOutput {
current : Ref[DynSource]
current_is_fallback : Ref[Bool]
prefetched : Ref[Sample?]
has_prefetched : Ref[Bool]
signal_after_end : Ref[QueueSignal?]
samples_consumed_in_span : Ref[Int]
padding_samples_remaining : Ref[Int]
input : SourcesQueueInput
}
///|
pub fn queue(
keep_alive_if_empty : Bool,
) -> (SourcesQueueInput, SourcesQueueOutput) {
let input = {
next_sounds: @ref.new([]),
keep_alive_if_empty: @ref.new(keep_alive_if_empty),
}
let output = {
current: @ref.new(make_empty_dyn_source(1, hz_44100)),
current_is_fallback: @ref.new(true),
prefetched: @ref.new(None),
has_prefetched: @ref.new(false),
signal_after_end: @ref.new(None),
samples_consumed_in_span: @ref.new(0),
padding_samples_remaining: @ref.new(0),
input,
}
(input, output)
}
///|
pub fn[S : Source] SourcesQueueInput::append(
self : SourcesQueueInput,
source : S,
) -> Unit {
self.next_sounds.val.push({ source: to_dyn(source), signal: None })
}
///|
pub fn[S : Source] SourcesQueueInput::append_with_signal(
self : SourcesQueueInput,
source : S,
) -> QueueSignal {
let signal = { done: @ref.new(false) }
self.next_sounds.val.push({ source: to_dyn(source), signal: Some(signal) })
signal
}
///|
pub fn SourcesQueueInput::set_keep_alive_if_empty(
self : SourcesQueueInput,
keep_alive_if_empty : Bool,
) -> Unit {
self.keep_alive_if_empty.val = keep_alive_if_empty
}
///|
pub fn SourcesQueueInput::clear(self : SourcesQueueInput) -> Int {
let len = self.next_sounds.val.length()
for entry in self.next_sounds.val {
match entry.signal {
Some(signal) => signal.mark_done()
None => ()
}
}
self.next_sounds.val.clear()
len
}
///|
fn SourcesQueueOutput::go_next(self : SourcesQueueOutput) -> Bool {
match self.signal_after_end.val {
Some(signal) => signal.mark_done()
None => ()
}
self.signal_after_end.val = None
if !self.input.next_sounds.val.is_empty() {
let next = self.input.next_sounds.val.remove(0)
self.current.val = next.source
self.signal_after_end.val = next.signal
self.current_is_fallback.val = false
self.prefetched.val = None
self.has_prefetched.val = false
self.samples_consumed_in_span.val = 0
true
} else if self.input.keep_alive_if_empty.val {
let channels = self.current.val.channels()
let sample_rate = self.current.val.sample_rate()
self.current.val = make_zero_dyn_source(
channels,
sample_rate,
threshold_for_channels(channels),
)
self.current_is_fallback.val = true
self.prefetched.val = None
self.has_prefetched.val = false
self.samples_consumed_in_span.val = 0
true
} else {
false
}
}
///|
fn SourcesQueueOutput::ensure_prefetched(self : SourcesQueueOutput) -> Bool {
if self.has_prefetched.val {
return true
}
while true {
if self.padding_samples_remaining.val > 0 {
self.prefetched.val = Some(0.0)
self.has_prefetched.val = true
self.padding_samples_remaining.val -= 1
return true
}
if self.current_is_fallback.val && !self.input.next_sounds.val.is_empty() {
let next = self.input.next_sounds.val.remove(0)
self.current.val = next.source
self.signal_after_end.val = next.signal
self.current_is_fallback.val = false
self.samples_consumed_in_span.val = 0
}
match self.current.val.next() {
Some(v) => {
self.prefetched.val = Some(v)
self.has_prefetched.val = true
self.samples_consumed_in_span.val += 1
return true
}
None => {
let channels = if self.current.val.channels() > 0 {
self.current.val.channels()
} else {
1
}
let incomplete_frame_samples = self.samples_consumed_in_span.val %
channels
self.samples_consumed_in_span.val = 0
if incomplete_frame_samples > 0 {
self.padding_samples_remaining.val = channels -
incomplete_frame_samples
continue
}
if !self.go_next() {
return false
}
}
}
}
false
}
///|
fn SourcesQueueOutput::next_internal(self : SourcesQueueOutput) -> Sample? {
if self.current_is_fallback.val && !self.input.next_sounds.val.is_empty() {
let next = self.input.next_sounds.val.remove(0)
self.current.val = next.source
self.signal_after_end.val = next.signal
self.current_is_fallback.val = false
self.prefetched.val = None
self.has_prefetched.val = false
self.samples_consumed_in_span.val = 0
self.padding_samples_remaining.val = 0
}
if !self.ensure_prefetched() {
return None
}
let value = match self.prefetched.val {
Some(v) => v
None => return None
}
self.prefetched.val = None
self.has_prefetched.val = false
// Eagerly prefetch next sample so metadata/state are updated at source boundaries.
ignore(self.ensure_prefetched())
Some(value)
}
///|
pub fn SourcesQueueOutput::next(self : SourcesQueueOutput) -> Sample? {
self.next_internal()
}
///|
pub fn SourcesQueueOutput::skip_one(self : SourcesQueueOutput) -> Unit {
self.prefetched.val = None
self.has_prefetched.val = false
self.samples_consumed_in_span.val = 0
self.padding_samples_remaining.val = 0
if !self.go_next() {
self.current.val = make_empty_dyn_source(1, hz_44100)
self.current_is_fallback.val = true
}
}
///|
pub fn SourcesQueueOutput::channels(self : SourcesQueueOutput) -> ChannelCount {
if !self.input.next_sounds.val.is_empty() &&
(
self.current_is_fallback.val ||
self.current.val.current_span_len() == Some(0)
) {
return self.input.next_sounds.val[0].source.channels()
}
self.current.val.channels()
}
///|
pub fn SourcesQueueOutput::sample_rate(self : SourcesQueueOutput) -> SampleRate {
if !self.input.next_sounds.val.is_empty() &&
(
self.current_is_fallback.val ||
self.current.val.current_span_len() == Some(0)
) {
return self.input.next_sounds.val[0].source.sample_rate()
}
self.current.val.sample_rate()
}
///|
pub impl Source for SourcesQueueOutput with next(self : SourcesQueueOutput) {
self.next()
}
///|
pub impl Source for SourcesQueueOutput with channels(self : SourcesQueueOutput) {
self.channels()
}
///|
pub impl Source for SourcesQueueOutput with sample_rate(
self : SourcesQueueOutput,
) {
self.sample_rate()
}
///|
pub impl Source for SourcesQueueOutput with current_span_len(
self : SourcesQueueOutput,
) {
if !self.input.next_sounds.val.is_empty() &&
(
self.current_is_fallback.val ||
self.current.val.current_span_len() == Some(0)
) {
match self.input.next_sounds.val[0].source.current_span_len() {
Some(v) => if v != 0 { return Some(v) }
None => ()
}
return Some(
threshold_for_channels(self.input.next_sounds.val[0].source.channels()),
)
}
match self.current.val.current_span_len() {
Some(v) =>
if v != 0 {
return Some(v)
} else if self.input.keep_alive_if_empty.val &&
self.input.next_sounds.val.is_empty() {
return Some(threshold_for_channels(self.current.val.channels()))
}
None => ()
}
Some(threshold_for_channels(self.current.val.channels()))
}
///|
pub impl Source for SourcesQueueOutput with total_duration(
_self : SourcesQueueOutput,
) {
source_default_total_duration()
}
///|
pub impl Source for SourcesQueueOutput with try_seek(
self : SourcesQueueOutput,
pos : @moon_cpal.Duration,
) -> Unit raise SeekError {
self.prefetched.val = None
self.has_prefetched.val = false
self.samples_consumed_in_span.val = 0
self.padding_samples_remaining.val = 0
self.current.val.try_seek(pos)
}