keket 0.20.3

Database-like Asset management on top of ECS storage
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use crate::{
    database::path::{AssetPath, AssetPathStatic},
    fetch::{AssetAwaitsAsyncFetch, AssetFetch},
};
use anput::{
    bundle::DynamicBundle,
    third_party::time::{Duration, Instant},
    world::World,
};
use std::{collections::BTreeSet, error::Error, sync::RwLock};

/// Strategy for throttling asset fetches during maintenance ticks.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ThrottledAssetFetchStrategy {
    /// Limits fetches to specified maximum number per maintenance tick.
    Number(usize),
    /// Limits fetches to specified maximum duration per maintenance tick.
    Duration(Duration),
}

pub struct ThrottledAssetFetch<Fetch: AssetFetch> {
    fetch: RwLock<Fetch>,
    strategy: ThrottledAssetFetchStrategy,
    awaiting: RwLock<BTreeSet<AssetPathStatic>>,
}

impl<Fetch: AssetFetch> ThrottledAssetFetch<Fetch> {
    pub fn new(fetch: Fetch, strategy: ThrottledAssetFetchStrategy) -> Self {
        Self {
            fetch: RwLock::new(fetch),
            strategy,
            awaiting: Default::default(),
        }
    }
}

impl<Fetch: AssetFetch> AssetFetch for ThrottledAssetFetch<Fetch> {
    fn load_bytes(&self, path: AssetPath) -> Result<DynamicBundle, Box<dyn Error>> {
        let path: AssetPathStatic = path.into_static();
        self.awaiting
            .write()
            .map_err(|error| {
                format!(
                    "Failed to get write access to inner fetch engine in throttled fetch for asset: `{path}`. Error: {error}"
                )
            })?
            .insert(path);
        let mut bundle = DynamicBundle::default();
        let _ = bundle.add_component(AssetAwaitsAsyncFetch);
        Ok(bundle)
    }

    fn maintain(&mut self, storage: &mut World) -> Result<(), Box<dyn Error>> {
        self.fetch
            .write()
            .map_err(|error| format!("Failed throttled fetch engine maintainance. Error: {error}"))?
            .maintain(storage)?;

        let mut number = 0;
        let timer = Instant::now();
        let mut awaiting = self.awaiting.write().map_err(|error| {
            format!(
                "Failed to get write access to awaiting fetches during throttled fetch maintainance. Error: {error}")
            }
        )?;
        while let Some(path) = awaiting.pop_first() {
            let bundle = self.fetch
                .write()
                .map_err(|error| {
                    format!(
                        "Failed to get write access to inner fetch engine in throttled fetch for asset: `{path}`. Error: {error}"
                    )
                })?
                .load_bytes(path.clone());
            match bundle {
                Ok(bundle) => {
                    if let Some(entity) = storage.find_by::<true, _>(&path) {
                        storage.remove::<(AssetAwaitsAsyncFetch,)>(entity)?;
                        storage.insert(entity, bundle)?;
                    }
                }
                Err(e) => {
                    if let Some(entity) = storage.find_by::<true, _>(&path) {
                        storage.remove::<(AssetAwaitsAsyncFetch,)>(entity)?;
                    }
                    return Err(format!(
                        "Throttled fetch execution of `{path}` asset failed with error: {e}"
                    )
                    .into());
                }
            }
            number += 1;
            match self.strategy {
                ThrottledAssetFetchStrategy::Number(max_per_tick) => {
                    if number >= max_per_tick {
                        break;
                    }
                }
                ThrottledAssetFetchStrategy::Duration(max_duration) => {
                    if timer.elapsed() >= max_duration {
                        break;
                    }
                }
            }
        }

        Ok(())
    }
}