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    collections::HashMap,
17    future::Future,
18    path::PathBuf,
19    pin::Pin,
20    sync::{Arc, Mutex, PoisonError},
21};
22
23use cranpose_core::{LaunchedEffectAsync, State, mutableStateOf, rememberKeyed};
24use cranpose_macros::composable;
25use cranpose_ui_graphics::ImageBitmap;
26use futures_util::future::{FutureExt, Shared, WeakShared};
27
28#[cfg(not(target_arch = "wasm32"))]
29use crate::async_io::Signal;
30use crate::http::HttpClientRef;
31
32/// The image to load, and the largest size in pixels it is drawn at.
33#[derive(Clone, Debug, PartialEq, Eq, Hash)]
34pub struct ImageRequest {
35    source: ImageSource,
36    max_width: u32,
37    max_height: u32,
38}
39
40#[derive(Clone, Debug)]
41enum ImageSource {
42    /// `http://`, `https://` or `file://`, or a file path.
43    Url(Arc<str>),
44    /// Encoded bytes the app read itself, from a picked file or a bundled
45    /// asset, named by `key` for the memory cache.
46    Bytes { key: Arc<str>, bytes: Arc<[u8]> },
47}
48
49impl ImageSource {
50    /// What names the image: its URL, or the key of its bytes. Comparing
51    /// and hashing the name keeps a lookup from reading the bytes.
52    fn name(&self) -> (bool, &str) {
53        match self {
54            Self::Url(url) => (true, url),
55            Self::Bytes { key, .. } => (false, key),
56        }
57    }
58}
59
60impl PartialEq for ImageSource {
61    fn eq(&self, other: &Self) -> bool {
62        self.name() == other.name()
63    }
64}
65
66impl Eq for ImageSource {}
67
68impl std::hash::Hash for ImageSource {
69    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
70        self.name().hash(state);
71    }
72}
73
74impl ImageRequest {
75    /// An image at `url`: `http://`, `https://`, `file://` or a file path.
76    pub fn new(url: impl Into<Arc<str>>) -> Self {
77        Self {
78            source: ImageSource::Url(url.into()),
79            max_width: u32::MAX,
80            max_height: u32::MAX,
81        }
82    }
83
84    /// An image in encoded `bytes` (PNG, JPEG, WebP or GIF) the app already
85    /// holds. `key` names it in the memory cache; one key is one image.
86    pub fn bytes(key: impl Into<Arc<str>>, bytes: impl Into<Arc<[u8]>>) -> Self {
87        Self {
88            source: ImageSource::Bytes {
89                key: key.into(),
90                bytes: bytes.into(),
91            },
92            max_width: u32::MAX,
93            max_height: u32::MAX,
94        }
95    }
96
97    /// Decodes the image no larger than `width` × `height` pixels, keeping
98    /// its aspect ratio: a thumbnail of a large photo holds a thumbnail's
99    /// memory. Pass the size it is drawn at, in device pixels.
100    pub fn max_size(mut self, width: u32, height: u32) -> Self {
101        self.max_width = width.max(1);
102        self.max_height = height.max(1);
103        self
104    }
105
106    /// [`max_size`](Self::max_size), unless the request already has one: a
107    /// composable that knows the size it draws at fills it in, and an app
108    /// that asked for a size keeps it.
109    pub fn or_max_size(self, width: u32, height: u32) -> Self {
110        if self.max_width == u32::MAX && self.max_height == u32::MAX {
111            self.max_size(width, height)
112        } else {
113            self
114        }
115    }
116}
117
118/// Where a request is.
119#[derive(Clone, Debug, PartialEq)]
120pub enum ImageLoadState {
121    /// The image is still being read and decoded.
122    Loading,
123    /// The image has loaded and is ready to draw.
124    Ready(ImageBitmap),
125    /// Loading failed for the stated reason.
126    Failed(ImageLoadError),
127}
128
129/// Why an image did not load.
130#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
131pub enum ImageLoadError {
132    /// The image bytes could not be read from their source. The text describes the cause.
133    #[error("could not read the image: {0}")]
134    Read(String),
135    /// The bytes were read but could not be decoded as an image. The text describes the cause.
136    #[error("could not decode the image: {0}")]
137    Decode(String),
138}
139
140/// How much the shared [`ImageLoader`] keeps.
141#[derive(Clone, Debug, PartialEq, Eq)]
142pub struct ImageCacheLimits {
143    /// The bytes of decoded pixels the memory cache holds.
144    pub memory_bytes: usize,
145    /// The bytes of downloaded files the disk cache holds.
146    pub disk_bytes: u64,
147}
148
149impl Default for ImageCacheLimits {
150    fn default() -> Self {
151        Self {
152            memory_bytes: 64 << 20,
153            disk_bytes: 128 << 20,
154        }
155    }
156}
157
158type LoadResult = Result<ImageBitmap, ImageLoadError>;
159
160#[cfg(not(target_arch = "wasm32"))]
161type LoadFuture = Pin<Box<dyn Future<Output = LoadResult> + Send>>;
162#[cfg(target_arch = "wasm32")]
163type LoadFuture = Pin<Box<dyn Future<Output = LoadResult>>>;
164
165/// What a loader shares with its loads: the browser's futures run on its one
166/// thread, so there they share through `Rc`.
167#[cfg(not(target_arch = "wasm32"))]
168type Handle<T> = Arc<T>;
169#[cfg(target_arch = "wasm32")]
170type Handle<T> = std::rc::Rc<T>;
171
172/// Loads, decodes and caches images. A clone is the same loader;
173/// [`image_loader`] is the one every [`rememberAsyncImage`] uses.
174#[derive(Clone)]
175pub struct ImageLoader {
176    inner: Handle<LoaderState>,
177}
178
179struct LoaderState {
180    memory: Mutex<MemoryCache>,
181    disk: Option<Arc<DiskCache>>,
182    in_flight: Mutex<HashMap<ImageRequest, WeakShared<LoadFuture>>>,
183}
184
185impl ImageLoader {
186    /// A loader that keeps downloaded files in `disk_dir`, or none on disk
187    /// when it is `None`.
188    pub fn new(limits: ImageCacheLimits, disk_dir: Option<PathBuf>) -> Self {
189        Self {
190            inner: Handle::new(LoaderState {
191                memory: Mutex::new(MemoryCache::new(limits.memory_bytes)),
192                disk: disk_dir.map(|dir| Arc::new(DiskCache::new(dir, limits.disk_bytes))),
193                in_flight: Mutex::new(HashMap::new()),
194            }),
195        }
196    }
197
198    /// The decoded image when the memory cache holds it.
199    pub fn cached(&self, request: &ImageRequest) -> Option<ImageBitmap> {
200        lock(&self.inner.memory).get(request)
201    }
202
203    /// Loads `request`, joining a load of the same request already under way.
204    /// The load stops when every future it returned is dropped.
205    pub fn load(
206        &self,
207        client: HttpClientRef,
208        request: &ImageRequest,
209    ) -> impl Future<Output = LoadResult> + use<> {
210        let shared = self.shared_load(client, request);
211        async move {
212            match shared {
213                Ok(bitmap) => Ok(bitmap),
214                Err(load) => load.await,
215            }
216        }
217    }
218
219    /// The cached image, or the load every caller of `request` shares.
220    fn shared_load(
221        &self,
222        client: HttpClientRef,
223        request: &ImageRequest,
224    ) -> Result<ImageBitmap, Shared<LoadFuture>> {
225        if let Some(bitmap) = self.cached(request) {
226            return Ok(bitmap);
227        }
228        let mut in_flight = lock(&self.inner.in_flight);
229        if let Some(load) = in_flight.get(request).and_then(WeakShared::upgrade) {
230            return Err(load);
231        }
232        let loader = self.clone();
233        let owned = request.clone();
234        let load: LoadFuture = Box::pin(async move {
235            let result = loader.fetch(client, &owned).await;
236            lock(&loader.inner.in_flight).remove(&owned);
237            if let Ok(bitmap) = &result {
238                lock(&loader.inner.memory).insert(owned, bitmap.clone());
239            }
240            result
241        });
242        let load = load.shared();
243        if let Some(weak) = load.downgrade() {
244            in_flight.insert(request.clone(), weak);
245        }
246        Err(load)
247    }
248
249    async fn fetch(&self, client: HttpClientRef, request: &ImageRequest) -> LoadResult {
250        let (width, height) = (request.max_width, request.max_height);
251        match &request.source {
252            ImageSource::Bytes { bytes, .. } => {
253                decode_off_thread(Encoded::Bytes(Arc::clone(bytes)), width, height).await
254            }
255            ImageSource::Url(url) if !is_remote(url) => {
256                decode_off_thread(Encoded::File(local_path(url)), width, height).await
257            }
258            ImageSource::Url(url) => {
259                if let Some(disk) = &self.inner.disk
260                    && let Some(cached) = decode_off_thread(
261                        Encoded::Cached(Arc::clone(disk), Arc::clone(url)),
262                        width,
263                        height,
264                    )
265                    .await
266                    .ok()
267                {
268                    return Ok(cached);
269                }
270                let bytes: Arc<[u8]> = client
271                    .get_bytes(url)
272                    .await
273                    .map_err(|error| ImageLoadError::Read(error.to_string()))?
274                    .into();
275                let encoded = match &self.inner.disk {
276                    Some(disk) => Encoded::Download {
277                        disk: Arc::clone(disk),
278                        url: Arc::clone(url),
279                        bytes,
280                    },
281                    None => Encoded::Bytes(bytes),
282                };
283                decode_off_thread(encoded, width, height).await
284            }
285        }
286    }
287}
288
289/// The loader every [`rememberAsyncImage`] shares, with the default limits
290/// and its disk cache in the application's cache folder.
291#[cfg(not(target_arch = "wasm32"))]
292pub fn image_loader() -> ImageLoader {
293    static LOADER: std::sync::OnceLock<ImageLoader> = std::sync::OnceLock::new();
294    LOADER.get_or_init(default_loader).clone()
295}
296
297/// The loader every [`rememberAsyncImage`] shares, with the default limits.
298/// The browser has no disk cache for it; its own HTTP cache stands in.
299#[cfg(target_arch = "wasm32")]
300pub fn image_loader() -> ImageLoader {
301    crate::thread_state::with(|state| {
302        state
303            .image_loader
304            .borrow_mut()
305            .get_or_insert_with(default_loader)
306            .clone()
307    })
308}
309
310fn default_loader() -> ImageLoader {
311    let disk_dir = if cfg!(target_arch = "wasm32") {
312        None
313    } else {
314        crate::host::application_directories()
315            .ok()
316            .map(|directories| directories.cache.join("images"))
317    };
318    ImageLoader::new(ImageCacheLimits::default(), disk_dir)
319}
320
321/// Loads `request` through [`image_loader`] while the caller stays in the
322/// composition. The caller reads the returned state, so it recomposes when
323/// the image is ready or failed. An image the memory cache holds is `Ready`
324/// in the first composition.
325///
326/// ```ignore
327/// match rememberAsyncImage(ImageRequest::new(url).max_size(240, 240)).value() {
328///     ImageLoadState::Ready(bitmap) => Image(BitmapPainter(bitmap), ..),
329///     ImageLoadState::Loading => Placeholder(),
330///     ImageLoadState::Failed(_) => BrokenImage(),
331/// }
332/// ```
333#[composable]
334pub fn rememberAsyncImage(request: ImageRequest) -> State<ImageLoadState> {
335    let loader = image_loader();
336    let cached = loader.cached(&request);
337    // Keyed: another request starts from its own cached image or `Loading`,
338    // never from the image of the request before it.
339    let state = rememberKeyed(request.clone(), move |_| {
340        mutableStateOf(cached.map_or(ImageLoadState::Loading, ImageLoadState::Ready))
341    });
342    let client = crate::http::local_http_client().current();
343    LaunchedEffectAsync(request.clone(), move |_| {
344        Box::pin(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    });
356    state.as_state()
357}
358
359fn is_remote(url: &str) -> bool {
360    url.starts_with("http://") || url.starts_with("https://")
361}
362
363fn local_path(url: &str) -> PathBuf {
364    PathBuf::from(url.strip_prefix("file://").unwrap_or(url))
365}
366
367/// Where the encoded image comes from, read on the decode worker.
368enum Encoded {
369    Bytes(Arc<[u8]>),
370    File(PathBuf),
371    /// The disk cache's copy, if it holds one.
372    Cached(Arc<DiskCache>, Arc<str>),
373    /// A download, which the worker also writes to the disk cache.
374    Download {
375        disk: Arc<DiskCache>,
376        url: Arc<str>,
377        bytes: Arc<[u8]>,
378    },
379}
380
381impl Encoded {
382    fn read(self) -> Result<Arc<[u8]>, ImageLoadError> {
383        match self {
384            Self::Bytes(bytes) => Ok(bytes),
385            Self::File(path) => std::fs::read(&path)
386                .map(Into::into)
387                .map_err(|error| ImageLoadError::Read(format!("{}: {error}", path.display()))),
388            Self::Cached(disk, url) => disk
389                .read(&url)
390                .map(Into::into)
391                .ok_or_else(|| ImageLoadError::Read(format!("{url} is not on disk"))),
392            Self::Download { disk, url, bytes } => {
393                disk.write(&url, &bytes);
394                Ok(bytes)
395            }
396        }
397    }
398}
399
400fn decode(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
401    let bytes = encoded.read()?;
402    let image = image::load_from_memory(&bytes)
403        .map_err(|error| ImageLoadError::Decode(error.to_string()))?;
404    let image = if image.width() > max_width || image.height() > max_height {
405        image.resize(max_width, max_height, image::imageops::FilterType::Triangle)
406    } else {
407        image
408    };
409    let rgba = image.into_rgba8();
410    let (width, height) = rgba.dimensions();
411    ImageBitmap::from_rgba8(width, height, rgba.into_raw())
412        .map_err(|error| ImageLoadError::Decode(error.to_string()))
413}
414
415/// Decodes on a worker thread, so a large image never holds a frame. The
416/// browser has no threads to hand the work to.
417#[cfg(not(target_arch = "wasm32"))]
418async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
419    let signal = Signal::new();
420    let done = signal.clone();
421    // Held while a caller waits: a job whose callers all left is skipped.
422    let waiting = Arc::new(());
423    let waiter = Arc::downgrade(&waiting);
424    let job: DecodeJob = Box::new(move || {
425        if waiter.strong_count() == 0 {
426            done.close();
427        } else {
428            done.set(decode(encoded, max_width, max_height));
429        }
430    });
431    if decode_queue().send(job).is_err() {
432        return Err(ImageLoadError::Decode("the decode threads stopped".into()));
433    }
434    let result = signal.wait().await;
435    drop(waiting);
436    result.unwrap_or_else(|| {
437        Err(ImageLoadError::Decode(
438            "the decode thread dropped the job".into(),
439        ))
440    })
441}
442
443#[cfg(target_arch = "wasm32")]
444async fn decode_off_thread(encoded: Encoded, max_width: u32, max_height: u32) -> LoadResult {
445    decode(encoded, max_width, max_height)
446}
447
448#[cfg(not(target_arch = "wasm32"))]
449type DecodeJob = Box<dyn FnOnce() + Send>;
450
451/// The queue of the decode threads: as many as the machine has cores, up to
452/// four, so a screen of thumbnails decodes side by side.
453#[cfg(not(target_arch = "wasm32"))]
454fn decode_queue() -> &'static std::sync::mpsc::Sender<DecodeJob> {
455    static QUEUE: std::sync::OnceLock<std::sync::mpsc::Sender<DecodeJob>> =
456        std::sync::OnceLock::new();
457    QUEUE.get_or_init(|| {
458        let (sender, receiver) = std::sync::mpsc::channel::<DecodeJob>();
459        let receiver = Arc::new(Mutex::new(receiver));
460        let threads = std::thread::available_parallelism().map_or(1, |cores| cores.get().min(4));
461        for index in 0..threads {
462            let receiver = Arc::clone(&receiver);
463            let started = std::thread::Builder::new()
464                .name(format!("cranpose-image-{index}"))
465                .spawn(move || {
466                    while let Ok(job) = lock(&receiver).recv() {
467                        job();
468                    }
469                });
470            if let Err(error) = started {
471                log::warn!("image loading: could not start a decode thread: {error}");
472            }
473        }
474        sender
475    })
476}
477
478/// Decoded images by request, the least recently used leaving first once
479/// their pixels pass the limit.
480struct MemoryCache {
481    entries: HashMap<ImageRequest, (ImageBitmap, u64)>,
482    bytes: usize,
483    limit: usize,
484    clock: u64,
485}
486
487impl MemoryCache {
488    fn new(limit: usize) -> Self {
489        Self {
490            entries: HashMap::new(),
491            bytes: 0,
492            limit,
493            clock: 0,
494        }
495    }
496
497    fn get(&mut self, request: &ImageRequest) -> Option<ImageBitmap> {
498        self.clock += 1;
499        let clock = self.clock;
500        self.entries.get_mut(request).map(|(bitmap, used)| {
501            *used = clock;
502            bitmap.clone()
503        })
504    }
505
506    fn insert(&mut self, request: ImageRequest, bitmap: ImageBitmap) {
507        let size = pixel_bytes(&bitmap);
508        if size > self.limit {
509            return;
510        }
511        self.clock += 1;
512        if let Some((old, _)) = self.entries.insert(request, (bitmap, self.clock)) {
513            self.bytes -= pixel_bytes(&old);
514        }
515        self.bytes += size;
516        while self.bytes > self.limit {
517            let Some(oldest) = self
518                .entries
519                .iter()
520                .min_by_key(|(_, (_, used))| *used)
521                .map(|(request, _)| request.clone())
522            else {
523                break;
524            };
525            if let Some((evicted, _)) = self.entries.remove(&oldest) {
526                self.bytes -= pixel_bytes(&evicted);
527            }
528        }
529    }
530}
531
532fn pixel_bytes(bitmap: &ImageBitmap) -> usize {
533    bitmap.width() as usize * bitmap.height() as usize * 4
534}
535
536/// Downloaded files, one per URL, named by the URL's hash and starting with
537/// the URL itself, so two URLs with one hash never read as each other. Past
538/// the limit the least recently read files go.
539struct DiskCache {
540    dir: PathBuf,
541    limit: u64,
542    /// The bytes the folder holds: scanned at the first write, then added up.
543    size: Mutex<Option<u64>>,
544}
545
546impl DiskCache {
547    fn new(dir: PathBuf, limit: u64) -> Self {
548        Self {
549            dir,
550            limit,
551            size: Mutex::new(None),
552        }
553    }
554
555    fn path(&self, url: &str) -> PathBuf {
556        self.dir.join(format!("{:016x}", fnv1a(url.as_bytes())))
557    }
558
559    fn read(&self, url: &str) -> Option<Vec<u8>> {
560        let path = self.path(url);
561        let mut contents = std::fs::read(&path).ok()?;
562        let header = contents.iter().position(|&byte| byte == b'\n')?;
563        if &contents[..header] != url.as_bytes() {
564            return None;
565        }
566        // Reading counts as use for the eviction order.
567        if let Ok(file) = std::fs::File::options().append(true).open(&path) {
568            let _ = file.set_modified(std::time::SystemTime::now());
569        }
570        contents.drain(..=header);
571        Some(contents)
572    }
573
574    fn write(&self, url: &str, bytes: &[u8]) {
575        if std::fs::create_dir_all(&self.dir).is_err() {
576            return;
577        }
578        let path = self.path(url);
579        let partial = path.with_extension("partial");
580        let mut contents = Vec::with_capacity(url.len() + 1 + bytes.len());
581        contents.extend_from_slice(url.as_bytes());
582        contents.push(b'\n');
583        contents.extend_from_slice(bytes);
584        let written = contents.len() as u64;
585        if std::fs::write(&partial, contents).is_err() || std::fs::rename(&partial, &path).is_err()
586        {
587            return;
588        }
589        let mut size = lock(&self.size);
590        let total = match *size {
591            Some(total) => total + written,
592            None => self.files().iter().map(|(_, len, _)| len).sum(),
593        };
594        *size = Some(if total > self.limit {
595            self.trim()
596        } else {
597            total
598        });
599    }
600
601    fn files(&self) -> Vec<(std::time::SystemTime, u64, PathBuf)> {
602        std::fs::read_dir(&self.dir)
603            .into_iter()
604            .flatten()
605            .filter_map(Result::ok)
606            .filter_map(|entry| {
607                let metadata = entry.metadata().ok()?;
608                Some((metadata.modified().ok()?, metadata.len(), entry.path()))
609            })
610            .collect()
611    }
612
613    /// Removes the least recently used files until the rest fit the limit,
614    /// and returns the bytes left.
615    fn trim(&self) -> u64 {
616        let mut files = self.files();
617        let mut total: u64 = files.iter().map(|(_, len, _)| len).sum();
618        files.sort_by_key(|(modified, _, _)| *modified);
619        for (_, len, path) in files {
620            if total <= self.limit {
621                break;
622            }
623            if std::fs::remove_file(&path).is_ok() {
624                total -= len;
625            }
626        }
627        total
628    }
629}
630
631/// FNV-1a: a hash that stays the same across builds, which a file name must.
632fn fnv1a(bytes: &[u8]) -> u64 {
633    bytes.iter().fold(0xcbf2_9ce4_8422_2325, |hash, &byte| {
634        (hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
635    })
636}
637
638fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
639    mutex.lock().unwrap_or_else(PoisonError::into_inner)
640}
cranpose_services/image_loading.rs

cranpose_services/
image_loading.rs
