1use std::{
16 future::Future,
17 path::PathBuf,
18 pin::Pin,
19 sync::{Arc, Mutex, PoisonError},
20};
21
22use cranpose_core::{
23 LaunchedEffect, State, collections::map::HashMap, mutableStateOf, rememberKeyed,
24};
25use cranpose_macros::composable;
26use cranpose_ui_graphics::ImageBitmap;
27use futures_util::future::{FutureExt, Shared, WeakShared};
28
29#[cfg(not(target_arch = "wasm32"))]
30use crate::async_io::Signal;
31use crate::http::HttpClientRef;
32
33#[derive(Clone, Debug, PartialEq, Eq, Hash)]
35pub struct ImageRequest {
36 source: ImageSource,
37 max_width: u32,
38 max_height: u32,
39}
40
41#[derive(Clone, Debug)]
42enum ImageSource {
43 Url(Arc<str>),
45 Bytes { key: Arc<str>, bytes: Arc<[u8]> },
48}
49
50impl ImageSource {
51 fn name(&self) -> (bool, &str) {
54 match self {
55 Self::Url(url) => (true, url),
56 Self::Bytes { key, .. } => (false, key),
57 }
58 }
59}
60
61impl PartialEq for ImageSource {
62 fn eq(&self, other: &Self) -> bool {
63 self.name() == other.name()
64 }
65}
66
67impl Eq for ImageSource {}
68
69impl std::hash::Hash for ImageSource {
70 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
71 self.name().hash(state);
72 }
73}
74
75impl ImageRequest {
76 pub fn new(url: impl Into<Arc<str>>) -> Self {
78 Self {
79 source: ImageSource::Url(url.into()),
80 max_width: u32::MAX,
81 max_height: u32::MAX,
82 }
83 }
84
85 pub fn bytes(key: impl Into<Arc<str>>, bytes: impl Into<Arc<[u8]>>) -> Self {
88 Self {
89 source: ImageSource::Bytes {
90 key: key.into(),
91 bytes: bytes.into(),
92 },
93 max_width: u32::MAX,
94 max_height: u32::MAX,
95 }
96 }
97
98 pub fn max_size(mut self, width: u32, height: u32) -> Self {
102 self.max_width = width.max(1);
103 self.max_height = height.max(1);
104 self
105 }
106
107 pub fn or_max_size(self, width: u32, height: u32) -> Self {
111 if self.max_width == u32::MAX && self.max_height == u32::MAX {
112 self.max_size(width, height)
113 } else {
114 self
115 }
116 }
117}
118
119#[derive(Clone, Debug, PartialEq)]
121pub enum ImageLoadState {
122 Loading,
124 Ready(ImageBitmap),
126 Failed(ImageLoadError),
128}
129
130#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
132pub enum ImageLoadError {
133 #[error("could not read the image: {0}")]
135 Read(String),
136 #[error("could not decode the image: {0}")]
138 Decode(String),
139}
140
141#[derive(Clone, Debug, PartialEq, Eq)]
143pub struct ImageCacheLimits {
144 pub memory_bytes: usize,
146 pub disk_bytes: u64,
148}
149
150impl Default for ImageCacheLimits {
151 fn default() -> Self {
152 Self {
153 memory_bytes: 64 << 20,
154 disk_bytes: 128 << 20,
155 }
156 }
157}
158
159type LoadResult = Result<ImageBitmap, ImageLoadError>;
160
161#[cfg(not(target_arch = "wasm32"))]
162type LoadFuture = Pin<Box<dyn Future<Output = LoadResult> + Send>>;
163#[cfg(target_arch = "wasm32")]
164type LoadFuture = Pin<Box<dyn Future<Output = LoadResult>>>;
165
166#[cfg(not(target_arch = "wasm32"))]
169type Handle<T> = Arc<T>;
170#[cfg(target_arch = "wasm32")]
171type Handle<T> = std::rc::Rc<T>;
172
173#[derive(Clone)]
176pub struct ImageLoader {
177 inner: Handle<LoaderState>,
178}
179
180struct LoaderState {
181 memory: Mutex<MemoryCache>,
182 disk: Option<Arc<DiskCache>>,
183 in_flight: Mutex<HashMap<ImageRequest, WeakShared<LoadFuture>>>,
184}
185
186impl ImageLoader {
187 pub fn new(limits: ImageCacheLimits, disk_dir: Option<PathBuf>) -> Self {
190 Self {
191 inner: Handle::new(LoaderState {
192 memory: Mutex::new(MemoryCache::new(limits.memory_bytes)),
193 disk: disk_dir.map(|dir| Arc::new(DiskCache::new(dir, limits.disk_bytes))),
194 in_flight: Mutex::new(HashMap::default()),
195 }),
196 }
197 }
198
199 pub fn cached(&self, request: &ImageRequest) -> Option<ImageBitmap> {
201 lock(&self.inner.memory).get(request)
202 }
203
204 pub fn load(
207 &self,
208 client: HttpClientRef,
209 request: &ImageRequest,
210 ) -> impl Future<Output = LoadResult> + use<> {
211 let shared = self.shared_load(client, request);
212 async move {
213 match shared {
214 Ok(bitmap) => Ok(bitmap),
215 Err(load) => load.await,
216 }
217 }
218 }
219
220 fn shared_load(
222 &self,
223 client: HttpClientRef,
224 request: &ImageRequest,
225 ) -> Result<ImageBitmap, Shared<LoadFuture>> {
226 if let Some(bitmap) = self.cached(request) {
227 return Ok(bitmap);
228 }
229 let mut in_flight = lock(&self.inner.in_flight);
230 if let Some(load) = in_flight.get(request).and_then(WeakShared::upgrade) {
231 return Err(load);
232 }
233 let loader = self.clone();
234 let owned = request.clone();
235 let load: LoadFuture = Box::pin(async move {
236 let result = loader.fetch(client, &owned).await;
237 lock(&loader.inner.in_flight).remove(&owned);
238 if let Ok(bitmap) = &result {
239 lock(&loader.inner.memory).insert(owned, bitmap.clone());
240 }
241 result
242 });
243 let load = load.shared();
244 if let Some(weak) = load.downgrade() {
245 in_flight.insert(request.clone(), weak);
246 }
247 Err(load)
248 }
249
250 async fn fetch(&self, client: HttpClientRef, request: &ImageRequest) -> LoadResult {
251 let (width, height) = (request.max_width, request.max_height);
252 match &request.source {
253 ImageSource::Bytes { bytes, .. } => {
254 decode_off_thread(Encoded::Bytes(Arc::clone(bytes)), width, height).await
255 }
256 ImageSource::Url(url) if !is_remote(url) => {
257 decode_off_thread(Encoded::File(local_path(url)), width, height).await
258 }
259 ImageSource::Url(url) => {
260 if let Some(disk) = &self.inner.disk
261 && let Some(cached) = decode_off_thread(
262 Encoded::Cached(Arc::clone(disk), Arc::clone(url)),
263 width,
264 height,
265 )
266 .await
267 .ok()
268 {
269 return Ok(cached);
270 }
271 let bytes: Arc<[u8]> = client
272 .get_bytes(url)
273 .await
274 .map_err(|error| ImageLoadError::Read(error.to_string()))?
275 .into();
276 let encoded = match &self.inner.disk {
277 Some(disk) => Encoded::Download {
278 disk: Arc::clone(disk),
279 url: Arc::clone(url),
280 bytes,
281 },
282 None => Encoded::Bytes(bytes),
283 };
284 decode_off_thread(encoded, width, height).await
285 }
286 }
287 }
288}
289
290#[cfg(not(target_arch = "wasm32"))]
293pub fn image_loader() -> ImageLoader {
294 static LOADER: std::sync::OnceLock<ImageLoader> = std::sync::OnceLock::new();
295 LOADER.get_or_init(default_loader).clone()
296}
297
298#[cfg(target_arch = "wasm32")]
301pub fn image_loader() -> ImageLoader {
302 crate::thread_state::with(|state| {
303 state
304 .image_loader
305 .borrow_mut()
306 .get_or_insert_with(default_loader)
307 .clone()
308 })
309}
310
311fn default_loader() -> ImageLoader {
312 let disk_dir = if cfg!(target_arch = "wasm32") {
313 None
314 } else {
315 crate::host::application_directories()
316 .ok()
317 .map(|directories| directories.cache.join("images"))
318 };
319 ImageLoader::new(ImageCacheLimits::default(), disk_dir)
320}
321
322#[composable]
335pub fn rememberAsyncImage(request: ImageRequest) -> State<ImageLoadState> {
336 let loader = image_loader();
337 let cached = loader.cached(&request);
338 let state = rememberKeyed(request.clone(), move |_| {
341 mutableStateOf(cached.map_or(ImageLoadState::Loading, ImageLoadState::Ready))
342 });
343 let client = crate::http::local_http_client().current();
344 LaunchedEffect(request.clone(), async move |_| {
345 if let Some(bitmap) = loader.cached(&request) {
346 state.set(ImageLoadState::Ready(bitmap));
347 return;
348 }
349 state.set(ImageLoadState::Loading);
350 state.set(match loader.load(client, &request).await {
351 Ok(bitmap) => ImageLoadState::Ready(bitmap),
352 Err(error) => ImageLoadState::Failed(error),
353 });
354 });
355 state.as_state()
356}
357
358fn is_remote(url: &str) -> bool {
359 url.starts_with("http://") || url.starts_with("https://")
360}
361
362fn local_path(url: &str) -> PathBuf {
363 PathBuf::from(url.strip_prefix("file://").unwrap_or(url))
364}
365
366enum Encoded {
368 Bytes(Arc<[u8]>),
369 File(PathBuf),
370 Cached(Arc<DiskCache>, Arc<str>),
372 Download {
374 disk: Arc<DiskCache>,
375 url: Arc<str>,
376 bytes: Arc<[u8]>,
377 },
378}
379
380impl Encoded {
381 fn read(self) -> Result<Arc<[u8]>, ImageLoadError> {
382 match self {
383 Self::Bytes(bytes) => Ok(bytes),
384 Self::File(path) => std::fs::read(&path)
385 .map(Into::into)
386 .map_err(|error| ImageLoadError::Read(format!("{}: {error}", path.display()))),
387 Self::Cached(disk, url) => disk
388 .read(&url)
389 .map(Into::into)
390 .ok_or_else(|| ImageLoadError::Read(format!("{url} is not on disk"))),
391 Self::Download { disk, url, bytes } => {
392 disk.write(&url, &bytes);
393 Ok(bytes)
394 }
395 }
396 }
397}
398
399fn decode(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
400 let bytes = encoded.read()?;
401 let image = image::load_from_memory(&bytes)
402 .map_err(|error| ImageLoadError::Decode(error.to_string()))?;
403 let image = if image.width() > max_width || image.height() > max_height {
404 image.resize(max_width, max_height, image::imageops::FilterType::Triangle)
405 } else {
406 image
407 };
408 let rgba = image.into_rgba8();
409 let (width, height) = rgba.dimensions();
410 ImageBitmap::from_rgba8(width, height, rgba.into_raw())
411 .map_err(|error| ImageLoadError::Decode(error.to_string()))
412}
413
414#[cfg(not(target_arch = "wasm32"))]
417async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
418 let signal = Signal::new();
419 let done = signal.clone();
420 let waiting = Arc::new(());
422 let waiter = Arc::downgrade(&waiting);
423 let job: DecodeJob = Box::new(move || {
424 if waiter.strong_count() == 0 {
425 done.close();
426 } else {
427 done.set(decode(encoded, max_width, max_height));
428 }
429 });
430 if decode_queue().send(job).is_err() {
431 return Err(ImageLoadError::Decode("the decode threads stopped".into()));
432 }
433 let result = signal.wait().await;
434 drop(waiting);
435 result.unwrap_or_else(|| {
436 Err(ImageLoadError::Decode(
437 "the decode thread dropped the job".into(),
438 ))
439 })
440}
441
442#[cfg(target_arch = "wasm32")]
443async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
444 decode(encoded, max_width, max_height)
445}
446
447#[cfg(not(target_arch = "wasm32"))]
448type DecodeJob = Box<dyn FnOnce() + Send>;
449
450#[cfg(not(target_arch = "wasm32"))]
453fn decode_queue() -> &'static std::sync::mpsc::Sender<DecodeJob> {
454 static QUEUE: std::sync::OnceLock<std::sync::mpsc::Sender<DecodeJob>> =
455 std::sync::OnceLock::new();
456 QUEUE.get_or_init(|| {
457 let (sender, receiver) = std::sync::mpsc::channel::<DecodeJob>();
458 let receiver = Arc::new(Mutex::new(receiver));
459 let threads = std::thread::available_parallelism().map_or(1, |cores| cores.get().min(4));
460 for index in 0..threads {
461 let receiver = Arc::clone(&receiver);
462 let started = std::thread::Builder::new()
463 .name(format!("cranpose-image-{index}"))
464 .spawn(move || {
465 while let Ok(job) = lock(&receiver).recv() {
466 job();
467 }
468 });
469 if let Err(error) = started {
470 log::warn!("image loading: could not start a decode thread: {error}");
471 }
472 }
473 sender
474 })
475}
476
477struct MemoryCache {
480 entries: HashMap<ImageRequest, (ImageBitmap, u64)>,
481 bytes: usize,
482 limit: usize,
483 clock: u64,
484}
485
486impl MemoryCache {
487 fn new(limit: usize) -> Self {
488 Self {
489 entries: HashMap::default(),
490 bytes: 0,
491 limit,
492 clock: 0,
493 }
494 }
495
496 fn get(&mut self, request: &ImageRequest) -> Option<ImageBitmap> {
497 self.clock += 1;
498 let clock = self.clock;
499 self.entries.get_mut(request).map(|(bitmap, used)| {
500 *used = clock;
501 bitmap.clone()
502 })
503 }
504
505 fn insert(&mut self, request: ImageRequest, bitmap: ImageBitmap) {
506 let size = pixel_bytes(&bitmap);
507 if size > self.limit {
508 return;
509 }
510 self.clock += 1;
511 if let Some((old, _)) = self.entries.insert(request, (bitmap, self.clock)) {
512 self.bytes -= pixel_bytes(&old);
513 }
514 self.bytes += size;
515 while self.bytes > self.limit {
516 let Some(oldest) = self
517 .entries
518 .iter()
519 .min_by_key(|(_, (_, used))| *used)
520 .map(|(request, _)| request.clone())
521 else {
522 break;
523 };
524 if let Some((evicted, _)) = self.entries.remove(&oldest) {
525 self.bytes -= pixel_bytes(&evicted);
526 }
527 }
528 }
529}
530
531fn pixel_bytes(bitmap: &ImageBitmap) -> usize {
532 bitmap.width() as usize * bitmap.height() as usize * 4
533}
534
535struct DiskCache {
539 dir: PathBuf,
540 limit: u64,
541 size: Mutex<Option<u64>>,
543}
544
545impl DiskCache {
546 fn new(dir: PathBuf, limit: u64) -> Self {
547 Self {
548 dir,
549 limit,
550 size: Mutex::new(None),
551 }
552 }
553
554 fn path(&self, url: &str) -> PathBuf {
555 self.dir.join(format!("{:016x}", fnv1a(url.as_bytes())))
556 }
557
558 fn read(&self, url: &str) -> Option<Vec<u8>> {
559 let path = self.path(url);
560 let mut contents = std::fs::read(&path).ok()?;
561 let header = contents.iter().position(|&byte| byte == b'\n')?;
562 if &contents[..header] != url.as_bytes() {
563 return None;
564 }
565 if let Ok(file) = std::fs::File::options().append(true).open(&path) {
567 let _ = file.set_modified(std::time::SystemTime::now());
568 }
569 contents.drain(..=header);
570 Some(contents)
571 }
572
573 fn write(&self, url: &str, bytes: &[u8]) {
574 if std::fs::create_dir_all(&self.dir).is_err() {
575 return;
576 }
577 let path = self.path(url);
578 let partial = path.with_extension("partial");
579 let mut contents = Vec::with_capacity(url.len() + 1 + bytes.len());
580 contents.extend_from_slice(url.as_bytes());
581 contents.push(b'\n');
582 contents.extend_from_slice(bytes);
583 let written = contents.len() as u64;
584 if std::fs::write(&partial, contents).is_err() || std::fs::rename(&partial, &path).is_err()
585 {
586 return;
587 }
588 let mut size = lock(&self.size);
589 let total = match *size {
590 Some(total) => total + written,
591 None => self.files().iter().map(|(_, len, _)| len).sum(),
592 };
593 *size = Some(if total > self.limit {
594 self.trim()
595 } else {
596 total
597 });
598 }
599
600 fn files(&self) -> Vec<(std::time::SystemTime, u64, PathBuf)> {
601 std::fs::read_dir(&self.dir)
602 .into_iter()
603 .flatten()
604 .filter_map(Result::ok)
605 .filter_map(|entry| {
606 let metadata = entry.metadata().ok()?;
607 Some((metadata.modified().ok()?, metadata.len(), entry.path()))
608 })
609 .collect()
610 }
611
612 fn trim(&self) -> u64 {
615 let mut files = self.files();
616 let mut total: u64 = files.iter().map(|(_, len, _)| len).sum();
617 files.sort_by_key(|(modified, _, _)| *modified);
618 for (_, len, path) in files {
619 if total <= self.limit {
620 break;
621 }
622 if std::fs::remove_file(&path).is_ok() {
623 total -= len;
624 }
625 }
626 total
627 }
628}
629
630fn fnv1a(bytes: &[u8]) -> u64 {
632 bytes.iter().fold(0xcbf2_9ce4_8422_2325, |hash, &byte| {
633 (hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
634 })
635}
636
637fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
638 mutex.lock().unwrap_or_else(PoisonError::into_inner)
639}