Store the processing state as an Arc.

Author: andriyDev

crates/bevy_asset/src/io/processor_gated.rs

@@ -1,6 +1,6 @@
 use crate::{
     io::{AssetReader, AssetReaderError, AssetSourceId, PathStream, Reader},
-    processor::{AssetProcessorData, ProcessStatus},
+    processor::{ProcessStatus, ProcessingState},
     AssetPath,
 };
 use alloc::{borrow::ToOwned, boxed::Box, sync::Arc, vec::Vec};
@@ -16,23 +16,23 @@ use super::{AsyncSeekForward, ErasedAssetReader};
 /// given path until that path has been processed by [`AssetProcessor`].
 ///
 /// [`AssetProcessor`]: crate::processor::AssetProcessor
-pub struct ProcessorGatedReader {
+pub(crate) struct ProcessorGatedReader {
     reader: Box<dyn ErasedAssetReader>,
     source: AssetSourceId<'static>,
-    processor_data: Arc<AssetProcessorData>,
+    processing_state: Arc<ProcessingState>,
 }
 
 impl ProcessorGatedReader {
     /// Creates a new [`ProcessorGatedReader`].
-    pub fn new(
+    pub(crate) fn new(
         source: AssetSourceId<'static>,
         reader: Box<dyn ErasedAssetReader>,
-        processor_data: Arc<AssetProcessorData>,
+        processing_state: Arc<ProcessingState>,
     ) -> Self {
         Self {
             source,
-            processor_data,
             reader,
+            processing_state,
         }
     }
 }
@@ -42,7 +42,6 @@ impl AssetReader for ProcessorGatedReader {
         let asset_path = AssetPath::from(path.to_path_buf()).with_source(self.source.clone());
         trace!("Waiting for processing to finish before reading {asset_path}");
         let process_result = self
-            .processor_data
             .processing_state
             .wait_until_processed(asset_path.clone())
             .await;
@@ -54,7 +53,6 @@ impl AssetReader for ProcessorGatedReader {
         }
         trace!("Processing finished with {asset_path}, reading {process_result:?}",);
         let lock = self
-            .processor_data
             .processing_state
             .get_transaction_lock(&asset_path)
             .await?;
@@ -67,7 +65,6 @@ impl AssetReader for ProcessorGatedReader {
         let asset_path = AssetPath::from(path.to_path_buf()).with_source(self.source.clone());
         trace!("Waiting for processing to finish before reading meta for {asset_path}",);
         let process_result = self
-            .processor_data
             .processing_state
             .wait_until_processed(asset_path.clone())
             .await;
@@ -79,7 +76,6 @@ impl AssetReader for ProcessorGatedReader {
         }
         trace!("Processing finished with {process_result:?}, reading meta for {asset_path}",);
         let lock = self
-            .processor_data
             .processing_state
             .get_transaction_lock(&asset_path)
             .await?;
@@ -96,10 +92,7 @@ impl AssetReader for ProcessorGatedReader {
             "Waiting for processing to finish before reading directory {:?}",
             path
         );
-        self.processor_data
-            .processing_state
-            .wait_until_finished()
-            .await;
+        self.processing_state.wait_until_finished().await;
         trace!("Processing finished, reading directory {:?}", path);
         let result = self.reader.read_directory(path).await?;
         Ok(result)
@@ -110,10 +103,7 @@ impl AssetReader for ProcessorGatedReader {
             "Waiting for processing to finish before reading directory {:?}",
             path
         );
-        self.processor_data
-            .processing_state
-            .wait_until_finished()
-            .await;
+        self.processing_state.wait_until_finished().await;
         trace!("Processing finished, getting directory status {:?}", path);
         let result = self.reader.is_directory(path).await?;
         Ok(result)

crates/bevy_asset/src/io/source.rs

@@ -1,6 +1,6 @@
 use crate::{
     io::{processor_gated::ProcessorGatedReader, AssetSourceEvent, AssetWatcher},
-    processor::AssetProcessorData,
+    processor::ProcessingState,
 };
 use alloc::{
     boxed::Box,
@@ -560,12 +560,12 @@ impl AssetSource {
 
     /// This will cause processed [`AssetReader`](crate::io::AssetReader) futures (such as [`AssetReader::read`](crate::io::AssetReader::read)) to wait until
     /// the [`AssetProcessor`](crate::AssetProcessor) has finished processing the requested asset.
-    pub fn gate_on_processor(&mut self, processor_data: Arc<AssetProcessorData>) {
+    pub(crate) fn gate_on_processor(&mut self, processing_state: Arc<ProcessingState>) {
         if let Some(reader) = self.processed_reader.take() {
             self.processed_reader = Some(Box::new(ProcessorGatedReader::new(
                 self.id(),
                 reader,
-                processor_data,
+                processing_state,
             )));
         }
     }
@@ -622,9 +622,9 @@ impl AssetSources {
 
     /// This will cause processed [`AssetReader`](crate::io::AssetReader) futures (such as [`AssetReader::read`](crate::io::AssetReader::read)) to wait until
     /// the [`AssetProcessor`](crate::AssetProcessor) has finished processing the requested asset.
-    pub fn gate_on_processor(&mut self, processor_data: Arc<AssetProcessorData>) {
+    pub(crate) fn gate_on_processor(&mut self, processing_state: Arc<ProcessingState>) {
         for source in self.iter_processed_mut() {
-            source.gate_on_processor(processor_data.clone());
+            source.gate_on_processor(processing_state.clone());
         }
     }
 }

crates/bevy_asset/src/lib.rs

@@ -390,7 +390,7 @@ impl Plugin for AssetPlugin {
                         let mut builders = app.world_mut().resource_mut::<AssetSourceBuilders>();
                         let processor = AssetProcessor::new(&mut builders);
                         let mut sources = builders.build_sources(false, watch);
-                        sources.gate_on_processor(processor.data.clone());
+                        sources.gate_on_processor(processor.data.processing_state.clone());
                         // the main asset server shares loaders with the processor asset server
                         app.insert_resource(AssetServer::new_with_loaders(
                             sources,

crates/bevy_asset/src/processor/mod.rs

@@ -105,7 +105,7 @@ pub struct AssetProcessor {
 /// Internal data stored inside an [`AssetProcessor`].
 pub struct AssetProcessorData {
     /// The state of processing.
-    pub(crate) processing_state: ProcessingState,
+    pub(crate) processing_state: Arc<ProcessingState>,
     /// The factory that creates the transaction log.
     ///
     /// Note: we use a regular Mutex instead of an async mutex since we expect users to only set
@@ -139,7 +139,7 @@ impl AssetProcessor {
         let data = Arc::new(AssetProcessorData::new(source.build_sources(true, false)));
         // The asset processor uses its own asset server with its own id space
         let mut sources = source.build_sources(false, false);
-        sources.gate_on_processor(data.clone());
+        sources.gate_on_processor(data.processing_state.clone());
         let server = AssetServer::new_with_meta_check(
             sources,
             AssetServerMode::Processed,
@@ -1212,7 +1212,7 @@ impl AssetProcessorData {
     /// Initializes a new [`AssetProcessorData`] using the given [`AssetSources`].
     pub fn new(source: AssetSources) -> Self {
         AssetProcessorData {
-            processing_state: ProcessingState::new(),
+            processing_state: Arc::new(ProcessingState::new()),
             sources: source,
             log_factory: Mutex::new(Some(Box::new(FileTransactionLogFactory::default()))),
             log: Default::default(),