refactor: remove tokio

CHANGELOG.md

@@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - This release supports **Bevy 0.17**.
 
+## [0.11.1] - 2025-12-21
+
+- This release supports **Bevy 0.17**.
+
+### Fixed
+
+- Removed the need for tokio channels internally to reduce dependency graph
+- Re-applied `async-compat` to native futures due to tokio reactor conflicts on native (particularly with reqwest).
+
 ## [0.11.0] - 2025-12-12
 
 - This release supports **Bevy 0.17**.

Cargo.toml

@@ -61,8 +61,6 @@ clippy.wildcard_dependencies = "warn"
 [lib]
 
 [dependencies]
-tokio = { version = "1.48.0", default-features = false, features = ["sync"] }
-
 bevy_ecs = { version = "0.17.3", default-features = false, features = [
   "multi_threaded",
   "async_executor",
@@ -84,6 +82,7 @@ bevy = { version = "0.17.3", default-features = false, features = [
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 futures-timer = "3.0.3"
+async-compat = "0.2.5"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 futures-timer = { version = "3.0.3", features = ["wasm-bindgen"] }

src/receiver.rs

@@ -2,9 +2,9 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 
+use futures::channel::mpsc;
+use futures::channel::oneshot;
 use futures::task::AtomicWaker;
-use tokio::sync::mpsc;
-use tokio::sync::oneshot::{self};
 
 /// A channel that catches an [`AsyncTask`](crate::AsyncTask) result.
 #[derive(Debug)]
@@ -18,13 +18,13 @@ impl<T> AsyncReceiver<T> {
     /// Poll the current thread waiting for the async result.
     pub fn try_recv(&mut self) -> Option<T> {
         match self.receiver.try_recv() {
-            Ok(t) => {
+            Ok(Some(t)) => {
                 self.receiver.close();
                 self.received.store(true, Ordering::Relaxed);
                 self.waker.wake(); // Wake the sender to drop
                 Some(t)
             }
-            Err(_) => None,
+            Ok(None) | Err(_) => None,
         }
     }
 }
@@ -41,16 +41,20 @@ pub struct AsyncStreamReceiver<T> {
 impl<T> AsyncStreamReceiver<T> {
     /// Returns whether the stream has finished producing items.
     pub fn is_finished(&self) -> bool {
-        self.finished.load(Ordering::Relaxed) && self.receiver.is_empty()
+        self.finished.load(Ordering::Relaxed)
     }
 
     /// Try to receive the next item from the stream without blocking.
     /// Returns `Some(item)` if an item is available, `None` otherwise.
     pub fn try_recv(&mut self) -> Option<T> {
-        match self.receiver.try_recv() {
-            Ok(item) => Some(item),
+        match self.receiver.try_next() {
+            Ok(Some(item)) => Some(item),
             Err(_) => {
-                // Check if we're truly finished (no more items will come)
+                // No message yet, and sender is not dropped.
+                None
+            }
+            Ok(None) => {
+                // Sender is closed, stream exhausted
                 if self.finished.load(Ordering::Relaxed) {
                     // Signal to the producer that we're done
                     self.received.store(true, Ordering::Relaxed);

src/stream.rs

@@ -9,8 +9,9 @@ use bevy_tasks::ConditionalSendFuture;
 use bevy_tasks::futures_lite::Stream;
 use bevy_tasks::futures_lite::StreamExt;
 use bevy_tasks::futures_lite::stream;
+use futures::SinkExt;
+use futures::channel::mpsc;
 use futures::task::AtomicWaker;
-use tokio::sync::mpsc;
 
 use crate::ConditionalSendStream;
 use crate::receiver::AsyncStreamReceiver;
@@ -59,7 +60,7 @@ impl<T: ConditionalSend + 'static> AsyncStream<T> {
         impl ConditionalSendFuture<Output = ()>,
         AsyncStreamReceiver<T>,
     ) {
-        let (tx, rx) = mpsc::unbounded_channel();
+        let (mut tx, rx) = mpsc::unbounded();
         let finished = Arc::new(AtomicBool::new(false));
         let finished_clone = Arc::clone(&finished);
         let waker = Arc::new(AtomicWaker::new());
@@ -69,8 +70,9 @@ impl<T: ConditionalSend + 'static> AsyncStream<T> {
 
         let fut = async move {
             let mut stream = self.stream;
+
             while let Some(item) = stream.next().await {
-                if tx.send(item).is_err() {
+                if tx.send(item).await.is_err() {
                     // Receiver dropped, exit early
                     break;
                 }

src/task.rs

@@ -7,10 +7,12 @@ use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::task::Poll;
 
+#[cfg(not(target_arch = "wasm32"))]
+use async_compat::CompatExt;
 use bevy_tasks::ConditionalSend;
 use bevy_tasks::ConditionalSendFuture;
+use futures::channel::oneshot;
 use futures::task::AtomicWaker;
-use tokio::sync::oneshot;
 
 use crate::AsyncReceiver;
 use crate::error::TimeoutError;
@@ -91,7 +93,10 @@ where
             let waker = waker.clone();
             let received = received.clone();
             async move {
+                #[cfg(target_arch = "wasm32")]
                 let result = self.fut.await;
+                #[cfg(not(target_arch = "wasm32"))]
+                let result = self.fut.compat().await;
 
                 if let Ok(()) = tx.send(result) {
                     // Wait for the receiver to get the result before dropping.
@@ -188,7 +193,10 @@ where
             let waker = waker.clone();
             let received = received.clone();
             async move {
+                #[cfg(target_arch = "wasm32")]
                 let result = timeout(self.timeout, self.fut).await;
+                #[cfg(not(target_arch = "wasm32"))]
+                let result = timeout(self.timeout, self.fut.compat()).await;
 
                 if let Ok(()) = tx.send(result) {
                     // Wait for the receiver to get the result before dropping.