bevy_asset/io/
processor_gated.rs

1//! Tools for waiting for the [`AssetProcessor`] before loading an [`Asset`].
2//!
3//! [`AssetProcessor`]: crate::processor::AssetProcessor
4//! [`Asset`]: crate::Asset
5
6use crate::{
7    io::{
8        AssetReader, AssetReaderError, AssetSourceId, PathStream, Reader, ReaderNotSeekableError,
9        SeekableReader,
10    },
11    processor::{ProcessStatus, ProcessingState},
12    AssetPath,
13};
14use alloc::{borrow::ToOwned, boxed::Box, sync::Arc, vec::Vec};
15use async_lock::RwLockReadGuardArc;
16use core::{pin::Pin, task::Poll};
17use futures_io::AsyncRead;
18use std::path::Path;
19use tracing::trace;
20
21use super::ErasedAssetReader;
22
23/// An [`AssetReader`] that will prevent asset (and asset metadata) read futures from returning for a
24/// given path until that path has been processed by [`AssetProcessor`].
25///
26/// [`AssetProcessor`]: crate::processor::AssetProcessor
27pub(crate) struct ProcessorGatedReader {
28    reader: Arc<dyn ErasedAssetReader>,
29    source: AssetSourceId<'static>,
30    processing_state: Arc<ProcessingState>,
31}
32
33impl ProcessorGatedReader {
34    /// Creates a new [`ProcessorGatedReader`].
35    pub(crate) fn new(
36        source: AssetSourceId<'static>,
37        reader: Arc<dyn ErasedAssetReader>,
38        processing_state: Arc<ProcessingState>,
39    ) -> Self {
40        Self {
41            source,
42            reader,
43            processing_state,
44        }
45    }
46}
47
48impl AssetReader for ProcessorGatedReader {
49    async fn read<'a>(&'a self, path: &'a Path) -> Result<impl Reader + 'a, AssetReaderError> {
50        let asset_path = AssetPath::from(path.to_path_buf()).with_source(self.source.clone());
51        trace!("Waiting for processing to finish before reading {asset_path}");
52        let process_result = self
53            .processing_state
54            .wait_until_processed(asset_path.clone())
55            .await;
56        match process_result {
57            ProcessStatus::Processed => {}
58            ProcessStatus::Failed | ProcessStatus::NonExistent => {
59                return Err(AssetReaderError::NotFound(path.to_owned()));
60            }
61        }
62        trace!("Processing finished with {asset_path}, reading {process_result:?}",);
63        let lock = self
64            .processing_state
65            .get_transaction_lock(&asset_path)
66            .await?;
67        let asset_reader = self.reader.read(path).await?;
68        let reader = TransactionLockedReader::new(asset_reader, lock);
69        Ok(reader)
70    }
71
72    async fn read_meta<'a>(&'a self, path: &'a Path) -> Result<impl Reader + 'a, AssetReaderError> {
73        let asset_path = AssetPath::from(path.to_path_buf()).with_source(self.source.clone());
74        trace!("Waiting for processing to finish before reading meta for {asset_path}",);
75        let process_result = self
76            .processing_state
77            .wait_until_processed(asset_path.clone())
78            .await;
79        match process_result {
80            ProcessStatus::Processed => {}
81            ProcessStatus::Failed | ProcessStatus::NonExistent => {
82                return Err(AssetReaderError::NotFound(path.to_owned()));
83            }
84        }
85        trace!("Processing finished with {process_result:?}, reading meta for {asset_path}",);
86        let lock = self
87            .processing_state
88            .get_transaction_lock(&asset_path)
89            .await?;
90        let meta_reader = self.reader.read_meta(path).await?;
91        let reader = TransactionLockedReader::new(meta_reader, lock);
92        Ok(reader)
93    }
94
95    async fn read_directory<'a>(
96        &'a self,
97        path: &'a Path,
98    ) -> Result<Box<PathStream>, AssetReaderError> {
99        trace!(
100            "Waiting for processing to finish before reading directory {:?}",
101            path
102        );
103        self.processing_state.wait_until_finished().await;
104        trace!("Processing finished, reading directory {:?}", path);
105        let result = self.reader.read_directory(path).await?;
106        Ok(result)
107    }
108
109    async fn is_directory<'a>(&'a self, path: &'a Path) -> Result<bool, AssetReaderError> {
110        trace!(
111            "Waiting for processing to finish before reading directory {:?}",
112            path
113        );
114        self.processing_state.wait_until_finished().await;
115        trace!("Processing finished, getting directory status {:?}", path);
116        let result = self.reader.is_directory(path).await?;
117        Ok(result)
118    }
119}
120
121/// An [`AsyncRead`] impl that will hold its asset's transaction lock until [`TransactionLockedReader`] is dropped.
122pub struct TransactionLockedReader<'a> {
123    reader: Box<dyn Reader + 'a>,
124    _file_transaction_lock: RwLockReadGuardArc<()>,
125}
126
127impl<'a> TransactionLockedReader<'a> {
128    fn new(reader: Box<dyn Reader + 'a>, file_transaction_lock: RwLockReadGuardArc<()>) -> Self {
129        Self {
130            reader,
131            _file_transaction_lock: file_transaction_lock,
132        }
133    }
134}
135
136impl AsyncRead for TransactionLockedReader<'_> {
137    fn poll_read(
138        mut self: Pin<&mut Self>,
139        cx: &mut core::task::Context<'_>,
140        buf: &mut [u8],
141    ) -> Poll<futures_io::Result<usize>> {
142        Pin::new(&mut self.reader).poll_read(cx, buf)
143    }
144}
145
146impl Reader for TransactionLockedReader<'_> {
147    fn read_to_end<'a>(
148        &'a mut self,
149        buf: &'a mut Vec<u8>,
150    ) -> stackfuture::StackFuture<'a, std::io::Result<usize>, { super::STACK_FUTURE_SIZE }> {
151        self.reader.read_to_end(buf)
152    }
153
154    fn seekable(&mut self) -> Result<&mut dyn SeekableReader, ReaderNotSeekableError> {
155        self.reader.seekable()
156    }
157}
bevy_asset/io/processor_gated.rs

bevy_asset/io/
processor_gated.rs
