audiobox/assets/
loader.rs

1use super::{AssetKey, AudioCache};
2use crate::{Audio, DecodeOptions, Error, ErrorCode, Result, decode_with_options};
3use std::fs::File;
4use std::io::Read;
5use std::path::{Path, PathBuf};
6use std::sync::Arc;
7use std::sync::mpsc::{self, Receiver, SyncSender, TryRecvError, TrySendError};
8use std::thread::{self, JoinHandle};
9
10/// Bounds shared by every request to a [`Loader`].
11#[derive(Clone, Debug)]
12pub struct LoaderConfig {
13    /// Retained cache-data budget in bytes. Default 64 MiB; zero disables caching.
14    pub cache_bytes: usize,
15    /// Requests waiting behind the single active decode. Default 16; must be > 0.
16    pub queue_capacity: usize,
17    /// Encoded bytes per request, checked before reading an entire file. Default 64 MiB.
18    /// Decoder buffers/output are separately bounded by `decode.limits`.
19    pub max_input_bytes: usize,
20    /// Fixed for this loader so cached data cannot bypass per-request decode limits.
21    pub decode: DecodeOptions,
22}
23
24impl Default for LoaderConfig {
25    fn default() -> Self {
26        Self {
27            cache_bytes: 64 * 1024 * 1024,
28            queue_capacity: 16,
29            max_input_bytes: 64 * 1024 * 1024,
30            decode: DecodeOptions::default(),
31        }
32    }
33}
34
35/// A one-shot result of a queued file read/decode. Dropping it does not cancel work.
36/// A completed result pins its cached sound until taken or dropped.
37pub struct LoadHandle {
38    receiver: Receiver<Result<Arc<Audio>>>,
39    taken: bool,
40}
41
42impl LoadHandle {
43    /// Take the result once, without waiting. `None` means pending or already taken.
44    /// Worker disconnection is delivered once as `ErrorCode::Io`.
45    /// This uses std channels and is not an audio-callback API.
46    pub fn try_take(&mut self) -> Option<Result<Arc<Audio>>> {
47        if self.taken {
48            return None;
49        }
50        let result = match self.receiver.try_recv() {
51            Ok(result) => result,
52            Err(TryRecvError::Empty) => return None,
53            Err(TryRecvError::Disconnected) => Err(disconnected()),
54        };
55        self.taken = true;
56        Some(result)
57    }
58
59    /// Wait for this request on a loading/control thread. After `try_take` consumed
60    /// the result this returns `InvalidArgument`.
61    pub fn wait(self) -> Result<Arc<Audio>> {
62        if self.taken {
63            return Err(Error::invalid("Load result already taken"));
64        }
65        self.receiver.recv().map_err(|_| disconnected())?
66    }
67}
68
69enum Input {
70    Bytes(Vec<u8>),
71    File(PathBuf),
72}
73
74struct Request {
75    key: AssetKey,
76    input: Input,
77    result: SyncSender<Result<Arc<Audio>>>,
78}
79
80/// One background worker with a bounded request queue and a private [`AudioCache`].
81///
82/// Submission never waits for queue space, but may allocate/free memory and uses
83/// std channels. All methods, polling and destruction belong off the audio thread.
84/// Dropping the loader closes its queue and **waits** for accepted requests to finish;
85/// their handles remain usable. A dropped handle does not cancel its decode/cache
86/// admission. No async runtime or mixer feature is required.
87pub struct Loader {
88    sender: Option<SyncSender<Request>>,
89    worker: Option<JoinHandle<()>>,
90    max_input_bytes: usize,
91}
92
93impl Loader {
94    /// Spawn the worker. Thread creation failures return `Io`.
95    pub fn new(config: LoaderConfig) -> Result<Self> {
96        if config.queue_capacity == 0 {
97            return Err(Error::invalid("Loader queue capacity must be positive"));
98        }
99        let (sender, receiver) = mpsc::sync_channel(config.queue_capacity);
100        let max_input_bytes = config.max_input_bytes;
101        let worker = thread::Builder::new()
102            .name("audiobox-loader".into())
103            .spawn(move || run(receiver, config))?;
104        Ok(Self {
105            sender: Some(sender),
106            worker: Some(worker),
107            max_input_bytes,
108        })
109    }
110
111    /// Queue owned encoded bytes under a name. Reusing a cached name returns its
112    /// original sound. Input size is checked even on cache hits; bytes are not copied.
113    pub fn load_bytes(&self, name: impl Into<String>, bytes: Vec<u8>) -> Result<LoadHandle> {
114        if bytes.len() > self.max_input_bytes {
115            return Err(Error::limit("Encoded input exceeds the loader limit"));
116        }
117        self.submit(AssetKey::Name(name.into()), Input::Bytes(bytes))
118    }
119
120    /// Queue a file by its absolute path. Resolving a relative path uses the current
121    /// directory now; file I/O and size/decode errors arrive through the handle.
122    pub fn load_file(&self, path: impl AsRef<Path>) -> Result<LoadHandle> {
123        let path = path.as_ref();
124        let path = if path.is_absolute() {
125            path.to_owned()
126        } else {
127            std::path::absolute(path)?
128        };
129        self.submit(AssetKey::Path(path.clone()), Input::File(path))
130    }
131
132    fn submit(&self, key: AssetKey, input: Input) -> Result<LoadHandle> {
133        let (result, receiver) = mpsc::sync_channel(1);
134        let request = Request { key, input, result };
135        match self.sender.as_ref().unwrap().try_send(request) {
136            Ok(()) => Ok(LoadHandle {
137                receiver,
138                taken: false,
139            }),
140            Err(TrySendError::Full(_)) => Err(Error::limit("Loader request queue is full")),
141            Err(TrySendError::Disconnected(_)) => Err(disconnected()),
142        }
143    }
144}
145
146impl Drop for Loader {
147    fn drop(&mut self) {
148        self.sender.take();
149        if let Some(worker) = self.worker.take() {
150            let _ = worker.join();
151        }
152    }
153}
154
155fn disconnected() -> Error {
156    Error::new(ErrorCode::Io, "Audio loader worker disconnected")
157}
158
159fn run(receiver: Receiver<Request>, config: LoaderConfig) {
160    let mut cache = AudioCache::new(config.cache_bytes);
161    for request in receiver {
162        let result = if let Some(audio) = cache.get(&request.key) {
163            Ok(audio)
164        } else {
165            load(request.input, &config).map(|audio| {
166                let audio = Arc::new(audio);
167                // Retention is optional: full/pinned caches must not fail a load.
168                let _ = cache.insert(request.key, Arc::clone(&audio));
169                audio
170            })
171        };
172        // One result slot, initially empty: sending never waits for the caller.
173        let _ = request.result.send(result);
174    }
175}
176
177fn load(input: Input, config: &LoaderConfig) -> Result<Audio> {
178    let bytes = match input {
179        Input::Bytes(bytes) => bytes,
180        Input::File(path) => {
181            let file = File::open(path)?;
182            if file.metadata()?.len() > config.max_input_bytes as u64 {
183                return Err(Error::limit("Encoded file exceeds the loader limit"));
184            }
185            let mut bytes = Vec::new();
186            // The extra byte catches a file that grows after the metadata check.
187            file.take((config.max_input_bytes as u64).saturating_add(1))
188                .read_to_end(&mut bytes)?;
189            if bytes.len() > config.max_input_bytes {
190                return Err(Error::limit("Encoded file exceeds the loader limit"));
191            }
192            bytes
193        }
194    };
195    decode_with_options(&bytes, &config.decode)
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201
202    #[test]
203    fn full_queue_and_disconnected_worker_report_errors() {
204        // No worker consumes this queue, making saturation independent of timing.
205        let (sender, receiver) = mpsc::sync_channel(1);
206        let loader = Loader {
207            sender: Some(sender),
208            worker: None,
209            max_input_bytes: 16,
210        };
211        let mut first = loader.load_bytes("first", vec![]).unwrap();
212        assert!(first.try_take().is_none());
213        assert_eq!(
214            loader.load_bytes("full", vec![]).err().unwrap().code,
215            ErrorCode::LimitExceeded
216        );
217        drop(receiver);
218        assert_eq!(first.try_take().unwrap().unwrap_err().code, ErrorCode::Io);
219        assert!(first.try_take().is_none());
220        assert_eq!(first.wait().unwrap_err().code, ErrorCode::InvalidArgument);
221        assert_eq!(
222            loader.load_bytes("closed", vec![]).err().unwrap().code,
223            ErrorCode::Io
224        );
225    }
226}
audiobox/assets/loader.rs

audiobox/assets/
loader.rs
