Allow the asset processor to run single-threaded.

crates/bevy_asset/Cargo.toml

@@ -45,12 +45,16 @@ async-broadcast = { version = "0.7.2", default-features = false }
 async-fs = { version = "2.0", default-features = false }
 async-lock = { version = "3.0", default-features = false }
 bitflags = { version = "2.3", default-features = false }
+async-channel = { version = "2", default-features = false }
 crossbeam-channel = { version = "0.5", default-features = false, features = [
   "std",
 ] }
 downcast-rs = { version = "2", default-features = false }
 disqualified = { version = "1.0", default-features = false }
 either = { version = "1.13", default-features = false }
+futures-util = { version = "0.3", default-features = false, features = [
+  "alloc",
+] }
 futures-io = { version = "0.3", default-features = false }
 futures-lite = { version = "2.0.1", default-features = false }
 blake3 = { version = "1.5", default-features = false }
@@ -91,9 +95,6 @@ notify-debouncer-full = { version = "0.5.0", default-features = false, optional
 ureq = { version = "3", optional = true, default-features = false }
 blocking = { version = "1.6", optional = true }
 
-[dev-dependencies]
-async-channel = "2"
-
 [lints]
 workspace = true
 

crates/bevy_asset/src/io/embedded/embedded_watcher.rs

@@ -28,7 +28,7 @@ impl EmbeddedWatcher {
     pub fn new(
         dir: Dir,
         root_paths: Arc<RwLock<HashMap<Box<Path>, PathBuf>>>,
-        sender: crossbeam_channel::Sender<AssetSourceEvent>,
+        sender: async_channel::Sender<AssetSourceEvent>,
         debounce_wait_time: Duration,
     ) -> Self {
         let root = get_base_path();
@@ -50,7 +50,7 @@ impl AssetWatcher for EmbeddedWatcher {}
 /// binary-embedded Rust source files. This will read the contents of changed files from the file system and overwrite
 /// the initial static bytes from the file embedded in the binary.
 pub(crate) struct EmbeddedEventHandler {
-    sender: crossbeam_channel::Sender<AssetSourceEvent>,
+    sender: async_channel::Sender<AssetSourceEvent>,
     root_paths: Arc<RwLock<HashMap<Box<Path>, PathBuf>>>,
     root: PathBuf,
     dir: Dir,
@@ -90,7 +90,7 @@ impl FilesystemEventHandler for EmbeddedEventHandler {
                 }
             }
             self.last_event = Some(event.clone());
-            self.sender.send(event).unwrap();
+            self.sender.send_blocking(event).unwrap();
         }
     }
 }

crates/bevy_asset/src/io/file/file_watcher.rs

@@ -3,8 +3,8 @@ use crate::{
     path::normalize_path,
 };
 use alloc::borrow::ToOwned;
+use async_channel::Sender;
 use core::time::Duration;
-use crossbeam_channel::Sender;
 use notify_debouncer_full::{
     new_debouncer,
     notify::{
@@ -269,7 +269,7 @@ impl FilesystemEventHandler for FileEventHandler {
     fn handle(&mut self, _absolute_paths: &[PathBuf], event: AssetSourceEvent) {
         if self.last_event.as_ref() != Some(&event) {
             self.last_event = Some(event.clone());
-            self.sender.send(event).unwrap();
+            self.sender.send_blocking(event).unwrap();
         }
     }
 }

crates/bevy_asset/src/io/source.rs

@@ -125,7 +125,7 @@ pub struct AssetSourceBuilder {
     /// The [`AssetWatcher`] to use for unprocessed assets, if any.
     pub watcher: Option<
         Box<
-            dyn FnMut(crossbeam_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
+            dyn FnMut(async_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
                 + Send
                 + Sync,
         >,
@@ -138,7 +138,7 @@ pub struct AssetSourceBuilder {
     /// The [`AssetWatcher`] to use for processed assets, if any.
     pub processed_watcher: Option<
         Box<
-            dyn FnMut(crossbeam_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
+            dyn FnMut(async_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
                 + Send
                 + Sync,
         >,
@@ -174,7 +174,7 @@ impl AssetSourceBuilder {
         };
 
         if watch {
-            let (sender, receiver) = crossbeam_channel::unbounded();
+            let (sender, receiver) = async_channel::unbounded();
             match self.watcher.as_mut().and_then(|w| w(sender)) {
                 Some(w) => {
                     source.watcher = Some(w);
@@ -189,7 +189,7 @@ impl AssetSourceBuilder {
         }
 
         if watch_processed {
-            let (sender, receiver) = crossbeam_channel::unbounded();
+            let (sender, receiver) = async_channel::unbounded();
             match self.processed_watcher.as_mut().and_then(|w| w(sender)) {
                 Some(w) => {
                     source.processed_watcher = Some(w);
@@ -226,7 +226,7 @@ impl AssetSourceBuilder {
     /// Will use the given `watcher` function to construct unprocessed [`AssetWatcher`] instances.
     pub fn with_watcher(
         mut self,
-        watcher: impl FnMut(crossbeam_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
+        watcher: impl FnMut(async_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
             + Send
             + Sync
             + 'static,
@@ -256,7 +256,7 @@ impl AssetSourceBuilder {
     /// Will use the given `watcher` function to construct processed [`AssetWatcher`] instances.
     pub fn with_processed_watcher(
         mut self,
-        watcher: impl FnMut(crossbeam_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
+        watcher: impl FnMut(async_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
             + Send
             + Sync
             + 'static,
@@ -377,8 +377,8 @@ pub struct AssetSource {
     processed_writer: Option<Box<dyn ErasedAssetWriter>>,
     watcher: Option<Box<dyn AssetWatcher>>,
     processed_watcher: Option<Box<dyn AssetWatcher>>,
-    event_receiver: Option<crossbeam_channel::Receiver<AssetSourceEvent>>,
-    processed_event_receiver: Option<crossbeam_channel::Receiver<AssetSourceEvent>>,
+    event_receiver: Option<async_channel::Receiver<AssetSourceEvent>>,
+    processed_event_receiver: Option<async_channel::Receiver<AssetSourceEvent>>,
 }
 
 impl AssetSource {
@@ -429,15 +429,13 @@ impl AssetSource {
 
     /// Return's this source's unprocessed event receiver, if the source is currently watching for changes.
     #[inline]
-    pub fn event_receiver(&self) -> Option<&crossbeam_channel::Receiver<AssetSourceEvent>> {
+    pub fn event_receiver(&self) -> Option<&async_channel::Receiver<AssetSourceEvent>> {
         self.event_receiver.as_ref()
     }
 
     /// Return's this source's processed event receiver, if the source is currently watching for changes.
     #[inline]
-    pub fn processed_event_receiver(
-        &self,
-    ) -> Option<&crossbeam_channel::Receiver<AssetSourceEvent>> {
+    pub fn processed_event_receiver(&self) -> Option<&async_channel::Receiver<AssetSourceEvent>> {
         self.processed_event_receiver.as_ref()
     }
 
@@ -517,10 +515,9 @@ impl AssetSource {
     pub fn get_default_watcher(
         path: String,
         file_debounce_wait_time: Duration,
-    ) -> impl FnMut(crossbeam_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>>
-           + Send
-           + Sync {
-        move |sender: crossbeam_channel::Sender<AssetSourceEvent>| {
+    ) -> impl FnMut(async_channel::Sender<AssetSourceEvent>) -> Option<Box<dyn AssetWatcher>> + Send + Sync
+    {
+        move |sender: async_channel::Sender<AssetSourceEvent>| {
             #[cfg(all(
                 feature = "file_watcher",
                 not(target_arch = "wasm32"),

crates/bevy_asset/src/lib.rs

@@ -737,6 +737,7 @@ mod tests {
         vec,
         vec::Vec,
     };
+    use async_channel::{Receiver, Sender};
     use bevy_app::{App, TaskPoolPlugin, Update};
     use bevy_diagnostic::{DiagnosticsPlugin, DiagnosticsStore};
     use bevy_ecs::{
@@ -750,7 +751,6 @@ mod tests {
     };
     use bevy_reflect::TypePath;
     use core::time::Duration;
-    use crossbeam_channel::Sender;
     use serde::{Deserialize, Serialize};
     use std::path::{Path, PathBuf};
     use thiserror::Error;
@@ -2153,8 +2153,8 @@ mod tests {
     // we've selected the reader. The GatedReader blocks this process, so we need to wait until
     // we gate in the loader instead.
     struct GatedLoader {
-        in_loader_sender: async_channel::Sender<()>,
-        gate_receiver: async_channel::Receiver<()>,
+        in_loader_sender: Sender<()>,
+        gate_receiver: Receiver<()>,
     }
 
     impl AssetLoader for GatedLoader {
@@ -2381,7 +2381,7 @@ mod tests {
         // Sending an asset event should result in the asset being reloaded - resulting in a
         // "Modified" message.
         source_events
-            .send(AssetSourceEvent::ModifiedAsset(PathBuf::from(
+            .send_blocking(AssetSourceEvent::ModifiedAsset(PathBuf::from(
                 "abc.cool.ron",
             )))
             .unwrap();
@@ -2436,7 +2436,7 @@ mod tests {
 )"#,
         );
         source_events
-            .send(AssetSourceEvent::AddedAsset(PathBuf::from("abc.cool.ron")))
+            .send_blocking(AssetSourceEvent::AddedAsset(PathBuf::from("abc.cool.ron")))
             .unwrap();
 
         run_app_until(&mut app, |world| {