Added Method to Allow Pipelined Asset Loading (#10565)

# Objective

- Fixes #10518

## Solution

I've added a method to `LoadContext`, `load_direct_with_reader`, which
mirrors the behaviour of `load_direct` with a single key difference: it
is provided with the `Reader` by the caller, rather than getting it from
the contained `AssetServer`. This allows for an `AssetLoader` to process
its `Reader` stream, and then directly hand the results off to the
`LoadContext` to handle further loading. The outer `AssetLoader` can
control how the `Reader` is interpreted by providing a relevant
`AssetPath`.

For example, a Gzip decompression loader could process the asset
`images/my_image.png.gz` by decompressing the bytes, then handing the
decompressed result to the `LoadContext` with the new path
`images/my_image.png.gz/my_image.png`. This intuitively reflects the
nature of contained assets, whilst avoiding unintended behaviour, since
the generated path cannot be a real file path (a file and folder of the
same name cannot coexist in most file-systems).

```rust
#[derive(Asset, TypePath)]
pub struct GzAsset {
    pub uncompressed: ErasedLoadedAsset,
}

#[derive(Default)]
pub struct GzAssetLoader;

impl AssetLoader for GzAssetLoader {
    type Asset = GzAsset;
    type Settings = ();
    type Error = GzAssetLoaderError;
    fn load<'a>(
        &'a self,
        reader: &'a mut Reader,
        _settings: &'a (),
        load_context: &'a mut LoadContext,
    ) -> BoxedFuture<'a, Result<Self::Asset, Self::Error>> {
        Box::pin(async move {
            let compressed_path = load_context.path();
            let file_name = compressed_path
                .file_name()
                .ok_or(GzAssetLoaderError::IndeterminateFilePath)?
                .to_string_lossy();
            let uncompressed_file_name = file_name
                .strip_suffix(".gz")
                .ok_or(GzAssetLoaderError::IndeterminateFilePath)?;
            let contained_path = compressed_path.join(uncompressed_file_name);

            let mut bytes_compressed = Vec::new();

            reader.read_to_end(&mut bytes_compressed).await?;

            let mut decoder = GzDecoder::new(bytes_compressed.as_slice());

            let mut bytes_uncompressed = Vec::new();

            decoder.read_to_end(&mut bytes_uncompressed)?;

            // Now that we have decompressed the asset, let's pass it back to the
            // context to continue loading

            let mut reader = VecReader::new(bytes_uncompressed);

            let uncompressed = load_context
                .load_direct_with_reader(&mut reader, contained_path)
                .await?;

            Ok(GzAsset { uncompressed })
        })
    }

    fn extensions(&self) -> &[&str] {
        &["gz"]
    }
}
```

Because this example is so prudent, I've included an
`asset_decompression` example which implements this exact behaviour:

```rust
fn main() {
    App::new()
        .add_plugins(DefaultPlugins)
        .init_asset::<GzAsset>()
        .init_asset_loader::<GzAssetLoader>()
        .add_systems(Startup, setup)
        .add_systems(Update, decompress::<Image>)
        .run();
}

fn setup(mut commands: Commands, asset_server: Res<AssetServer>) {
    commands.spawn(Camera2dBundle::default());

    commands.spawn((
        Compressed::<Image> {
            compressed: asset_server.load("data/compressed_image.png.gz"),
            ..default()
        },
        Sprite::default(),
        TransformBundle::default(),
        VisibilityBundle::default(),
    ));
}

fn decompress<A: Asset>(
    mut commands: Commands,
    asset_server: Res<AssetServer>,
    mut compressed_assets: ResMut<Assets<GzAsset>>,
    query: Query<(Entity, &Compressed<A>)>,
) {
    for (entity, Compressed { compressed, .. }) in query.iter() {
        let Some(GzAsset { uncompressed }) = compressed_assets.remove(compressed) else {
            continue;
        };

        let uncompressed = uncompressed.take::<A>().unwrap();

        commands
            .entity(entity)
            .remove::<Compressed<A>>()
            .insert(asset_server.add(uncompressed));
    }
}
```

A key limitation to this design is how to type the internally loaded
asset, since the example `GzAssetLoader` is unaware of the internal
asset type `A`. As such, in this example I store the contained asset as
an `ErasedLoadedAsset`, and leave it up to the consumer of the `GzAsset`
to handle typing the final result, which is the purpose of the
`decompress` system. This limitation can be worked around by providing
type information to the `GzAssetLoader`, such as `GzAssetLoader<Image,
ImageAssetLoader>`, but this would require registering the asset loader
for every possible decompression target.

Aside from this limitation, nested asset containerisation works as an
end user would expect; if the user registers a `TarAssetLoader`, and a
`GzAssetLoader`, then they can load assets with compound
containerisation, such as `images.tar.gz`.

---

## Changelog

- Added `LoadContext::load_direct_with_reader`
- Added `asset_decompression` example

## Notes

- While I believe my implementation of a Gzip asset loader is
reasonable, I haven't included it as a public feature of `bevy_asset` to
keep the scope of this PR as focussed as possible.
- I have included `flate2` as a `dev-dependency` for the example; it is
not included in the main dependency graph.

Cargo.toml

@@ -282,6 +282,7 @@ bevy_internal = { path = "crates/bevy_internal", version = "0.12.0", default-fea
 [dev-dependencies]
 rand = "0.8.0"
 ron = "0.8.0"
+flate2 = "1.0"
 serde = { version = "1", features = ["derive"] }
 bytemuck = "1.7"
 # Needed to poll Task examples
@@ -1077,6 +1078,17 @@ description = "Demonstrates various methods to load assets"
 category = "Assets"
 wasm = false
 
+[[example]]
+name = "asset_decompression"
+path = "examples/asset/asset_decompression.rs"
+doc-scrape-examples = true
+
+[package.metadata.example.asset_decompression]
+name = "Asset Decompression"
+description = "Demonstrates loading a compressed asset"
+category = "Assets"
+wasm = false
+
 [[example]]
 name = "custom_asset"
 path = "examples/asset/custom_asset.rs"

crates/bevy_asset/src/loader.rs

@@ -562,6 +562,62 @@ impl<'a> LoadContext<'a> {
         self.loader_dependencies.insert(path, hash);
         Ok(loaded_asset)
     }
+
+    /// Loads the asset at the given `path` directly from the provided `reader`. This is an async function that will wait until the asset is fully loaded before
+    /// returning. Use this if you need the _value_ of another asset in order to load the current asset, and that value comes from your [`Reader`].
+    /// For example, if you are deriving a new asset from the referenced asset, or you are building a collection of assets. This will add the `path` as a
+    /// "load dependency".
+    ///
+    /// If the current loader is used in a [`Process`] "asset preprocessor", such as a [`LoadAndSave`] preprocessor,
+    /// changing a "load dependency" will result in re-processing of the asset.
+    ///
+    /// [`Process`]: crate::processor::Process
+    /// [`LoadAndSave`]: crate::processor::LoadAndSave
+    pub async fn load_direct_with_reader<'b>(
+        &mut self,
+        reader: &mut Reader<'_>,
+        path: impl Into<AssetPath<'b>>,
+    ) -> Result<ErasedLoadedAsset, LoadDirectError> {
+        let path = path.into().into_owned();
+
+        let loader = self
+            .asset_server
+            .get_path_asset_loader(&path)
+            .await
+            .map_err(|error| LoadDirectError {
+                dependency: path.clone(),
+                error: error.into(),
+            })?;
+
+        let meta = loader.default_meta();
+
+        let loaded_asset = self
+            .asset_server
+            .load_with_meta_loader_and_reader(
+                &path,
+                meta,
+                &*loader,
+                reader,
+                false,
+                self.populate_hashes,
+            )
+            .await
+            .map_err(|error| LoadDirectError {
+                dependency: path.clone(),
+                error,
+            })?;
+
+        let info = loaded_asset
+            .meta
+            .as_ref()
+            .and_then(|m| m.processed_info().as_ref());
+
+        let hash = info.map(|i| i.full_hash).unwrap_or_default();
+
+        self.loader_dependencies.insert(path, hash);
+
+        Ok(loaded_asset)
+    }
 }
 
 /// An error produced when calling [`LoadContext::read_asset_bytes`]

examples/README.md

@@ -181,6 +181,7 @@ Example | Description
 
 Example | Description
 --- | ---
+[Asset Decompression](../examples/asset/asset_decompression.rs) | Demonstrates loading a compressed asset
 [Asset Loading](../examples/asset/asset_loading.rs) | Demonstrates various methods to load assets
 [Asset Processing](../examples/asset/processing/asset_processing.rs) | Demonstrates how to process and load custom assets
 [Custom Asset](../examples/asset/custom_asset.rs) | Implements a custom asset loader

examples/asset/asset_decompression.rs

@@ -0,0 +1,138 @@
+//! Implements loader for a Gzip compressed asset.
+
+use bevy::utils::thiserror;
+use bevy::{
+    asset::{
+        io::{Reader, VecReader},
+        AssetLoader, AsyncReadExt, ErasedLoadedAsset, LoadContext, LoadDirectError,
+    },
+    prelude::*,
+    reflect::TypePath,
+    utils::BoxedFuture,
+};
+use flate2::read::GzDecoder;
+use std::io::prelude::*;
+use std::marker::PhantomData;
+use thiserror::Error;
+
+#[derive(Asset, TypePath)]
+pub struct GzAsset {
+    pub uncompressed: ErasedLoadedAsset,
+}
+
+#[derive(Default)]
+pub struct GzAssetLoader;
+
+/// Possible errors that can be produced by [`GzAssetLoader`]
+#[non_exhaustive]
+#[derive(Debug, Error)]
+pub enum GzAssetLoaderError {
+    /// An [IO](std::io) Error
+    #[error("Could not load asset: {0}")]
+    Io(#[from] std::io::Error),
+    /// An error caused when the asset path cannot be used ot determine the uncompressed asset type.
+    #[error("Could not determine file path of uncompressed asset")]
+    IndeterminateFilePath,
+    /// An error caused by the internal asset loader.
+    #[error("Could not load contained asset: {0}")]
+    LoadDirectError(#[from] LoadDirectError),
+}
+
+impl AssetLoader for GzAssetLoader {
+    type Asset = GzAsset;
+    type Settings = ();
+    type Error = GzAssetLoaderError;
+    fn load<'a>(
+        &'a self,
+        reader: &'a mut Reader,
+        _settings: &'a (),
+        load_context: &'a mut LoadContext,
+    ) -> BoxedFuture<'a, Result<Self::Asset, Self::Error>> {
+        Box::pin(async move {
+            let compressed_path = load_context.path();
+            let file_name = compressed_path
+                .file_name()
+                .ok_or(GzAssetLoaderError::IndeterminateFilePath)?
+                .to_string_lossy();
+            let uncompressed_file_name = file_name
+                .strip_suffix(".gz")
+                .ok_or(GzAssetLoaderError::IndeterminateFilePath)?;
+            let contained_path = compressed_path.join(uncompressed_file_name);
+
+            let mut bytes_compressed = Vec::new();
+
+            reader.read_to_end(&mut bytes_compressed).await?;
+
+            let mut decoder = GzDecoder::new(bytes_compressed.as_slice());
+
+            let mut bytes_uncompressed = Vec::new();
+
+            decoder.read_to_end(&mut bytes_uncompressed)?;
+
+            // Now that we have decompressed the asset, let's pass it back to the
+            // context to continue loading
+
+            let mut reader = VecReader::new(bytes_uncompressed);
+
+            let uncompressed = load_context
+                .load_direct_with_reader(&mut reader, contained_path)
+                .await?;
+
+            Ok(GzAsset { uncompressed })
+        })
+    }
+
+    fn extensions(&self) -> &[&str] {
+        &["gz"]
+    }
+}
+
+#[derive(Component, Default)]
+struct Compressed<T> {
+    compressed: Handle<GzAsset>,
+    _phantom: PhantomData<T>,
+}
+
+fn main() {
+    App::new()
+        .add_plugins(DefaultPlugins)
+        .init_asset::<GzAsset>()
+        .init_asset_loader::<GzAssetLoader>()
+        .add_systems(Startup, setup)
+        .add_systems(Update, decompress::<Image>)
+        .run();
+}
+
+fn setup(mut commands: Commands, asset_server: Res<AssetServer>) {
+    commands.spawn(Camera2dBundle::default());
+
+    commands.spawn((
+        Compressed::<Image> {
+            compressed: asset_server.load("data/compressed_image.png.gz"),
+            ..default()
+        },
+        Sprite::default(),
+        TransformBundle::default(),
+        VisibilityBundle::default(),
+    ));
+}
+
+fn decompress<A: Asset>(
+    mut commands: Commands,
+    asset_server: Res<AssetServer>,
+    mut compressed_assets: ResMut<Assets<GzAsset>>,
+    query: Query<(Entity, &Compressed<A>)>,
+) {
+    for (entity, Compressed { compressed, .. }) in query.iter() {
+        let Some(GzAsset { uncompressed }) = compressed_assets.remove(compressed) else {
+            continue;
+        };
+
+        let uncompressed = uncompressed.take::<A>().unwrap();
+
+        commands
+            .entity(entity)
+            .remove::<Compressed<A>>()
+            .insert(asset_server.add(uncompressed));
+    }
+}