Replace crossbeam_channel with async_channel for hot reloading.

Author: andriyDev

crates/bevy_asset/Cargo.toml

@@ -45,6 +45,7 @@ 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",
 ] }
@@ -91,9 +92,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/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::{

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| {

crates/bevy_asset/src/processor/mod.rs

@@ -274,7 +274,8 @@ impl AssetProcessor {
 
             for source in self.data.sources.iter_processed() {
                 if let Some(receiver) = source.event_receiver() {
-                    for event in receiver.try_iter() {
+                    // TODO: Handle TryRecvError::Closed.
+                    while let Ok(event) = receiver.try_recv() {
                         if !started_processing {
                             self.set_state(ProcessorState::Processing).await;
                             started_processing = true;

crates/bevy_asset/src/server/mod.rs

@@ -1836,14 +1836,14 @@ pub fn handle_internal_asset_events(world: &mut World) {
             match server.data.mode {
                 AssetServerMode::Unprocessed => {
                     if let Some(receiver) = source.event_receiver() {
-                        for event in receiver.try_iter() {
+                        while let Ok(event) = receiver.try_recv() {
                             handle_event(source.id(), event);
                         }
                     }
                 }
                 AssetServerMode::Processed => {
                     if let Some(receiver) = source.processed_event_receiver() {
-                        for event in receiver.try_iter() {
+                        while let Ok(event) = receiver.try_recv() {
                             handle_event(source.id(), event);
                         }
                     }