use std::collections::HashMap;
use crate::{
app::SystemStage,
assets::{
deps::Dependencies,
storage::{Assets, RawAssetHandle},
upload::Asset,
},
ecs::{
plugin::Plugin,
resources::Resources,
system::{Local, Res, ResMut},
},
};
const STUCK_AFTER_TICKS: u32 = 300;
fn should_warn_stuck(ticks: u32) -> bool {
ticks >= STUCK_AFTER_TICKS && ticks.is_multiple_of(STUCK_AFTER_TICKS)
}
pub struct AssetPlugin<B, T: Asset<B>> {
_marker: std::marker::PhantomData<(B, T)>,
}
impl<B, T: Asset<B>> AssetPlugin<B, T> {
pub fn new() -> Self {
Self { _marker: std::marker::PhantomData }
}
}
impl<B, T> Plugin for AssetPlugin<B, T>
where
B: 'static + Send + Sync,
T: Asset<B>,
{
fn build(&self, app: &mut crate::app::App) {
app.try_insert_resource(Assets::<T>::new());
app.add_system(SystemStage::AssetSync, sync_assets::<B, T>);
}
}
fn sync_assets<B, T>(
mut assets: ResMut<Assets<T>>,
backend: Option<Res<B>>,
mut blocked_ticks: Local<u32>,
mut pending_ticks: Local<HashMap<RawAssetHandle, u32>>,
world: &hecs::World,
resources: &Resources,
) where
B: 'static + Send + Sync,
T: Asset<B>,
{
let Some(backend) = backend else {
log_waiting::<B, T>(&assets, "backend", &mut blocked_ticks);
return;
};
let Some(deps) = T::Deps::try_gather(world, resources) else {
log_waiting::<B, T>(&assets, "dependencies", &mut blocked_ticks);
return;
};
*blocked_ticks = 0;
for handle in assets.take_removed() {
pending_ticks.remove(&handle);
}
let dirty = assets.take_dirty();
let mut still_pending = Vec::new();
let results: Vec<(RawAssetHandle, Option<String>, Option<Option<T::Processed>>)> = dirty
.iter()
.map(|&handle| match assets.get_source_quiet(handle) {
None => (handle, None, None),
Some(source) => {
let name = assets.name_for_handle(handle).map(String::from);
let outcome = source.upload(&backend, &deps);
(handle, name, Some(outcome))
}
})
.collect();
for (handle, name, outcome) in results {
let label = name.as_deref().map(|n| format!(" ({n})")).unwrap_or_default();
match outcome {
None => {
tracing::debug!(
"{}: handle {:?} was in the dirty queue but the source asset is already \
gone (inserted and removed in the same tick?)",
std::any::type_name::<T>(),
handle
);
pending_ticks.remove(&handle);
}
Some(None) => {
let ticks = pending_ticks.entry(handle).or_insert(0);
*ticks += 1;
if should_warn_stuck(*ticks) {
tracing::warn!(
"{}: {:?}{} has not uploaded after {} ticks — upload() may be \
unconditionally returning None, or a Deps resource it needs is never \
actually going to appear. Still retrying every tick.",
std::any::type_name::<T>(),
handle,
label,
*ticks
);
} else {
tracing::debug!(
"{}: {:?}{} upload returned None — a required dependency is not yet \
ready, requeued for next tick",
std::any::type_name::<T>(),
handle,
label,
);
}
still_pending.push(handle);
}
Some(Some(value)) => {
assets.set_processed(handle, value);
pending_ticks.remove(&handle);
tracing::debug!(
"{}: uploaded {:?}{}",
std::any::type_name::<T>(),
handle,
label
);
}
}
}
if !still_pending.is_empty() {
tracing::debug!(
"{}: {} handle(s) still pending upload (waiting on dependencies)",
std::any::type_name::<T>(),
still_pending.len()
);
}
assets.requeue(still_pending);
}
fn log_waiting<B, T>(assets: &Assets<T>, what: &str, blocked_ticks: &mut u32)
where
B: 'static + Send + Sync,
T: Asset<B>,
{
if assets.dirty_is_empty() {
*blocked_ticks = 0;
return;
}
*blocked_ticks += 1;
if should_warn_stuck(*blocked_ticks) {
tracing::warn!(
"{}: {} asset(s) have been queued for {} ticks, still waiting on {what} before \
upload can begin — if {what} is never going to appear, this pipeline will wait \
forever.",
std::any::type_name::<T>(),
assets.dirty_len(),
*blocked_ticks,
);
} else {
tracing::debug!(
"{}: {} asset(s) queued but waiting on {what} before upload can begin",
std::any::type_name::<T>(),
assets.dirty_len()
);
}
}