cranpose_services/
image_loading.rs

1//! Images loaded from the network, a file or bytes the app holds, decoded off
2//! the UI thread at the size they are drawn, as Compose apps load them with
3//! Coil's `AsyncImage`.
4//!
5//! [`rememberAsyncImage`] gives a composable the [`ImageLoadState`] of an
6//! [`ImageRequest`]: `Loading` until the image is decoded, then `Ready` or
7//! `Failed`. All images share one [`ImageLoader`]: a memory cache bounded by
8//! the bytes of its decoded pixels, a disk cache of downloaded files with a
9//! size limit, and one download for every request of one image. A request
10//! whose composables all left the composition stops its download.
11//!
12//! The network goes through `local_http_client()`, so an app enables the
13//! `http-native` (or `web-http`) feature as it does for its own requests.
14
15use std::{
16    future::Future,
17    path::PathBuf,
18    pin::Pin,
19    sync::{Arc, Mutex, PoisonError},
20};
21
22use cranpose_core::{
23    LaunchedEffectAsync, 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/// The image to load, and the largest size in pixels it is drawn at.
34#[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    /// `http://`, `https://` or `file://`, or a file path.
44    Url(Arc<str>),
45    /// Encoded bytes the app read itself, from a picked file or a bundled
46    /// asset, named by `key` for the memory cache.
47    Bytes { key: Arc<str>, bytes: Arc<[u8]> },
48}
49
50impl ImageSource {
51    /// What names the image: its URL, or the key of its bytes. Comparing
52    /// and hashing the name keeps a lookup from reading the bytes.
53    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    /// An image at `url`: `http://`, `https://`, `file://` or a file path.
77    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    /// An image in encoded `bytes` (PNG, JPEG, WebP or GIF) the app already
86    /// holds. `key` names it in the memory cache; one key is one image.
87    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    /// Decodes the image no larger than `width` × `height` pixels, keeping
99    /// its aspect ratio: a thumbnail of a large photo holds a thumbnail's
100    /// memory. Pass the size it is drawn at, in device pixels.
101    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    /// [`max_size`](Self::max_size), unless the request already has one: a
108    /// composable that knows the size it draws at fills it in, and an app
109    /// that asked for a size keeps it.
110    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/// Where a request is.
120#[derive(Clone, Debug, PartialEq)]
121pub enum ImageLoadState {
122    /// The image is still being read and decoded.
123    Loading,
124    /// The image has loaded and is ready to draw.
125    Ready(ImageBitmap),
126    /// Loading failed for the stated reason.
127    Failed(ImageLoadError),
128}
129
130/// Why an image did not load.
131#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
132pub enum ImageLoadError {
133    /// The image bytes could not be read from their source. The text describes the cause.
134    #[error("could not read the image: {0}")]
135    Read(String),
136    /// The bytes were read but could not be decoded as an image. The text describes the cause.
137    #[error("could not decode the image: {0}")]
138    Decode(String),
139}
140
141/// How much the shared [`ImageLoader`] keeps.
142#[derive(Clone, Debug, PartialEq, Eq)]
143pub struct ImageCacheLimits {
144    /// The bytes of decoded pixels the memory cache holds.
145    pub memory_bytes: usize,
146    /// The bytes of downloaded files the disk cache holds.
147    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/// What a loader shares with its loads: the browser's futures run on its one
167/// thread, so there they share through `Rc`.
168#[cfg(not(target_arch = "wasm32"))]
169type Handle<T> = Arc<T>;
170#[cfg(target_arch = "wasm32")]
171type Handle<T> = std::rc::Rc<T>;
172
173/// Loads, decodes and caches images. A clone is the same loader;
174/// [`image_loader`] is the one every [`rememberAsyncImage`] uses.
175#[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    /// A loader that keeps downloaded files in `disk_dir`, or none on disk
188    /// when it is `None`.
189    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    /// The decoded image when the memory cache holds it.
200    pub fn cached(&self, request: &ImageRequest) -> Option<ImageBitmap> {
201        lock(&self.inner.memory).get(request)
202    }
203
204    /// Loads `request`, joining a load of the same request already under way.
205    /// The load stops when every future it returned is dropped.
206    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    /// The cached image, or the load every caller of `request` shares.
221    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/// The loader every [`rememberAsyncImage`] shares, with the default limits
291/// and its disk cache in the application's cache folder.
292#[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/// The loader every [`rememberAsyncImage`] shares, with the default limits.
299/// The browser has no disk cache for it; its own HTTP cache stands in.
300#[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/// Loads `request` through [`image_loader`] while the caller stays in the
323/// composition. The caller reads the returned state, so it recomposes when
324/// the image is ready or failed. An image the memory cache holds is `Ready`
325/// in the first composition.
326///
327/// ```ignore
328/// match rememberAsyncImage(ImageRequest::new(url).max_size(240, 240)).value() {
329///     ImageLoadState::Ready(bitmap) => Image(BitmapPainter(bitmap), ..),
330///     ImageLoadState::Loading => Placeholder(),
331///     ImageLoadState::Failed(_) => BrokenImage(),
332/// }
333/// ```
334#[composable]
335pub fn rememberAsyncImage(request: ImageRequest) -> State<ImageLoadState> {
336    let loader = image_loader();
337    let cached = loader.cached(&request);
338    // Keyed: another request starts from its own cached image or `Loading`,
339    // never from the image of the request before it.
340    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    LaunchedEffectAsync(request.clone(), move |_| {
345        Box::pin(async move {
346            if let Some(bitmap) = loader.cached(&request) {
347                state.set(ImageLoadState::Ready(bitmap));
348                return;
349            }
350            state.set(ImageLoadState::Loading);
351            state.set(match loader.load(client, &request).await {
352                Ok(bitmap) => ImageLoadState::Ready(bitmap),
353                Err(error) => ImageLoadState::Failed(error),
354            });
355        })
356    });
357    state.as_state()
358}
359
360fn is_remote(url: &str) -> bool {
361    url.starts_with("http://") || url.starts_with("https://")
362}
363
364fn local_path(url: &str) -> PathBuf {
365    PathBuf::from(url.strip_prefix("file://").unwrap_or(url))
366}
367
368/// Where the encoded image comes from, read on the decode worker.
369enum Encoded {
370    Bytes(Arc<[u8]>),
371    File(PathBuf),
372    /// The disk cache's copy, if it holds one.
373    Cached(Arc<DiskCache>, Arc<str>),
374    /// A download, which the worker also writes to the disk cache.
375    Download {
376        disk: Arc<DiskCache>,
377        url: Arc<str>,
378        bytes: Arc<[u8]>,
379    },
380}
381
382impl Encoded {
383    fn read(self) -> Result<Arc<[u8]>, ImageLoadError> {
384        match self {
385            Self::Bytes(bytes) => Ok(bytes),
386            Self::File(path) => std::fs::read(&path)
387                .map(Into::into)
388                .map_err(|error| ImageLoadError::Read(format!("{}: {error}", path.display()))),
389            Self::Cached(disk, url) => disk
390                .read(&url)
391                .map(Into::into)
392                .ok_or_else(|| ImageLoadError::Read(format!("{url} is not on disk"))),
393            Self::Download { disk, url, bytes } => {
394                disk.write(&url, &bytes);
395                Ok(bytes)
396            }
397        }
398    }
399}
400
401fn decode(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
402    let bytes = encoded.read()?;
403    let image = image::load_from_memory(&bytes)
404        .map_err(|error| ImageLoadError::Decode(error.to_string()))?;
405    let image = if image.width() > max_width || image.height() > max_height {
406        image.resize(max_width, max_height, image::imageops::FilterType::Triangle)
407    } else {
408        image
409    };
410    let rgba = image.into_rgba8();
411    let (width, height) = rgba.dimensions();
412    ImageBitmap::from_rgba8(width, height, rgba.into_raw())
413        .map_err(|error| ImageLoadError::Decode(error.to_string()))
414}
415
416/// Decodes on a worker thread, so a large image never holds a frame. The
417/// browser has no threads to hand the work to.
418#[cfg(not(target_arch = "wasm32"))]
419async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
420    let signal = Signal::new();
421    let done = signal.clone();
422    // Held while a caller waits: a job whose callers all left is skipped.
423    let waiting = Arc::new(());
424    let waiter = Arc::downgrade(&waiting);
425    let job: DecodeJob = Box::new(move || {
426        if waiter.strong_count() == 0 {
427            done.close();
428        } else {
429            done.set(decode(encoded, max_width, max_height));
430        }
431    });
432    if decode_queue().send(job).is_err() {
433        return Err(ImageLoadError::Decode("the decode threads stopped".into()));
434    }
435    let result = signal.wait().await;
436    drop(waiting);
437    result.unwrap_or_else(|| {
438        Err(ImageLoadError::Decode(
439            "the decode thread dropped the job".into(),
440        ))
441    })
442}
443
444#[cfg(target_arch = "wasm32")]
445async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
446    decode(encoded, max_width, max_height)
447}
448
449#[cfg(not(target_arch = "wasm32"))]
450type DecodeJob = Box<dyn FnOnce() + Send>;
451
452/// The queue of the decode threads: as many as the machine has cores, up to
453/// four, so a screen of thumbnails decodes side by side.
454#[cfg(not(target_arch = "wasm32"))]
455fn decode_queue() -> &'static std::sync::mpsc::Sender<DecodeJob> {
456    static QUEUE: std::sync::OnceLock<std::sync::mpsc::Sender<DecodeJob>> =
457        std::sync::OnceLock::new();
458    QUEUE.get_or_init(|| {
459        let (sender, receiver) = std::sync::mpsc::channel::<DecodeJob>();
460        let receiver = Arc::new(Mutex::new(receiver));
461        let threads = std::thread::available_parallelism().map_or(1, |cores| cores.get().min(4));
462        for index in 0..threads {
463            let receiver = Arc::clone(&receiver);
464            let started = std::thread::Builder::new()
465                .name(format!("cranpose-image-{index}"))
466                .spawn(move || {
467                    while let Ok(job) = lock(&receiver).recv() {
468                        job();
469                    }
470                });
471            if let Err(error) = started {
472                log::warn!("image loading: could not start a decode thread: {error}");
473            }
474        }
475        sender
476    })
477}
478
479/// Decoded images by request, the least recently used leaving first once
480/// their pixels pass the limit.
481struct MemoryCache {
482    entries: HashMap<ImageRequest, (ImageBitmap, u64)>,
483    bytes: usize,
484    limit: usize,
485    clock: u64,
486}
487
488impl MemoryCache {
489    fn new(limit: usize) -> Self {
490        Self {
491            entries: HashMap::default(),
492            bytes: 0,
493            limit,
494            clock: 0,
495        }
496    }
497
498    fn get(&mut self, request: &ImageRequest) -> Option<ImageBitmap> {
499        self.clock += 1;
500        let clock = self.clock;
501        self.entries.get_mut(request).map(|(bitmap, used)| {
502            *used = clock;
503            bitmap.clone()
504        })
505    }
506
507    fn insert(&mut self, request: ImageRequest, bitmap: ImageBitmap) {
508        let size = pixel_bytes(&bitmap);
509        if size > self.limit {
510            return;
511        }
512        self.clock += 1;
513        if let Some((old, _)) = self.entries.insert(request, (bitmap, self.clock)) {
514            self.bytes -= pixel_bytes(&old);
515        }
516        self.bytes += size;
517        while self.bytes > self.limit {
518            let Some(oldest) = self
519                .entries
520                .iter()
521                .min_by_key(|(_, (_, used))| *used)
522                .map(|(request, _)| request.clone())
523            else {
524                break;
525            };
526            if let Some((evicted, _)) = self.entries.remove(&oldest) {
527                self.bytes -= pixel_bytes(&evicted);
528            }
529        }
530    }
531}
532
533fn pixel_bytes(bitmap: &ImageBitmap) -> usize {
534    bitmap.width() as usize * bitmap.height() as usize * 4
535}
536
537/// Downloaded files, one per URL, named by the URL's hash and starting with
538/// the URL itself, so two URLs with one hash never read as each other. Past
539/// the limit the least recently read files go.
540struct DiskCache {
541    dir: PathBuf,
542    limit: u64,
543    /// The bytes the folder holds: scanned at the first write, then added up.
544    size: Mutex<Option<u64>>,
545}
546
547impl DiskCache {
548    fn new(dir: PathBuf, limit: u64) -> Self {
549        Self {
550            dir,
551            limit,
552            size: Mutex::new(None),
553        }
554    }
555
556    fn path(&self, url: &str) -> PathBuf {
557        self.dir.join(format!("{:016x}", fnv1a(url.as_bytes())))
558    }
559
560    fn read(&self, url: &str) -> Option<Vec<u8>> {
561        let path = self.path(url);
562        let mut contents = std::fs::read(&path).ok()?;
563        let header = contents.iter().position(|&byte| byte == b'\n')?;
564        if &contents[..header] != url.as_bytes() {
565            return None;
566        }
567        // Reading counts as use for the eviction order.
568        if let Ok(file) = std::fs::File::options().append(true).open(&path) {
569            let _ = file.set_modified(std::time::SystemTime::now());
570        }
571        contents.drain(..=header);
572        Some(contents)
573    }
574
575    fn write(&self, url: &str, bytes: &[u8]) {
576        if std::fs::create_dir_all(&self.dir).is_err() {
577            return;
578        }
579        let path = self.path(url);
580        let partial = path.with_extension("partial");
581        let mut contents = Vec::with_capacity(url.len() + 1 + bytes.len());
582        contents.extend_from_slice(url.as_bytes());
583        contents.push(b'\n');
584        contents.extend_from_slice(bytes);
585        let written = contents.len() as u64;
586        if std::fs::write(&partial, contents).is_err() || std::fs::rename(&partial, &path).is_err()
587        {
588            return;
589        }
590        let mut size = lock(&self.size);
591        let total = match *size {
592            Some(total) => total + written,
593            None => self.files().iter().map(|(_, len, _)| len).sum(),
594        };
595        *size = Some(if total > self.limit {
596            self.trim()
597        } else {
598            total
599        });
600    }
601
602    fn files(&self) -> Vec<(std::time::SystemTime, u64, PathBuf)> {
603        std::fs::read_dir(&self.dir)
604            .into_iter()
605            .flatten()
606            .filter_map(Result::ok)
607            .filter_map(|entry| {
608                let metadata = entry.metadata().ok()?;
609                Some((metadata.modified().ok()?, metadata.len(), entry.path()))
610            })
611            .collect()
612    }
613
614    /// Removes the least recently used files until the rest fit the limit,
615    /// and returns the bytes left.
616    fn trim(&self) -> u64 {
617        let mut files = self.files();
618        let mut total: u64 = files.iter().map(|(_, len, _)| len).sum();
619        files.sort_by_key(|(modified, _, _)| *modified);
620        for (_, len, path) in files {
621            if total <= self.limit {
622                break;
623            }
624            if std::fs::remove_file(&path).is_ok() {
625                total -= len;
626            }
627        }
628        total
629    }
630}
631
632/// FNV-1a: a hash that stays the same across builds, which a file name must.
633fn fnv1a(bytes: &[u8]) -> u64 {
634    bytes.iter().fold(0xcbf2_9ce4_8422_2325, |hash, &byte| {
635        (hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
636    })
637}
638
639fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
640    mutex.lock().unwrap_or_else(PoisonError::into_inner)
641}
cranpose_services/image_loading.rs

cranpose_services/
image_loading.rs
