de_net: Support custom task spawner (#561)

This makes the crate compatible with Bevy, needed because Bevy uses
custom task executor.

Author: Indy2222

crates/connector/src/game/mod.rs

@@ -21,7 +21,12 @@ mod state;
 ///   automatically added to the game as if they sent [`de_net::ToGame::Join`].
 pub(crate) async fn startup(net: Network, owner: SocketAddr) {
     let port = net.port();
-    let (outputs, inputs, errors) = de_net::startup(net);
+    let (outputs, inputs, errors) = de_net::startup(
+        |t| {
+            task::spawn(t);
+        },
+        net,
+    );
 
     let (server_sender, server_receiver) = bounded(16);
     task::spawn(ereceiver::run(port, errors, server_sender.clone()));

crates/connector/src/server.rs

@@ -1,6 +1,7 @@
 use std::net::SocketAddr;
 
 use anyhow::Context;
+use async_std::task;
 use de_net::{
     self, FromServer, MessageDecoder, MessageReceiver, MessageSender, Network, OutMessage, Peers,
     ToServer,
@@ -19,7 +20,12 @@ pub(crate) struct MainServer {
 impl MainServer {
     /// Setup the server & startup its network stack.
     pub(crate) fn start(net: Network) -> Self {
-        let (outputs, inputs, _) = de_net::startup(net);
+        let (outputs, inputs, _) = de_net::startup(
+            |t| {
+                task::spawn(t);
+            },
+            net,
+        );
         Self { outputs, inputs }
     }
 

crates/net/src/tasks/mod.rs

@@ -55,12 +55,13 @@
 //! data. The user communicates with these via [`MessageSender`] and
 //! [`MessageReceiver`] respectively.
 
-use async_std::{channel::bounded, task};
+use async_std::channel::bounded;
 pub use communicator::{
     ConnErrorReceiver, ConnectionError, InMessage, MessageDecoder, MessageReceiver, MessageSender,
     OutMessage, OutMessageBuilder,
 };
 pub(crate) use dsender::OutDatagram;
+use futures::future::BoxFuture;
 use tracing::info;
 
 use crate::{
@@ -89,69 +90,82 @@ const CHANNEL_CAPACITY: usize = 1024;
 /// closed. Once the [`MessageSender`], [`MessageReceiver`], and
 /// [`ConnErrorReceiver`] are all dropped, the networking stack will terminate
 /// completely.
-pub fn startup(network: Network) -> (MessageSender, MessageReceiver, ConnErrorReceiver) {
+///
+/// # Arguments
+///
+/// * `spawn` - async task spawner.
+///
+/// * `network` - network communication will happen over this socket.
+pub fn startup<S>(spawn: S, network: Network) -> (MessageSender, MessageReceiver, ConnErrorReceiver)
+where
+    S: Fn(BoxFuture<'static, ()>),
+{
     let port = network.port();
     info!("Starting up network stack on port {port}...");
 
     let messages = Messages::new(network);
 
     let (out_datagrams_sender, out_datagrams_receiver) = bounded(16);
-    task::spawn(dsender::run(port, out_datagrams_receiver, messages.clone()));
+    spawn(Box::pin(dsender::run(
+        port,
+        out_datagrams_receiver,
+        messages.clone(),
+    )));
 
     let (in_system_datagrams_sender, in_system_datagrams_receiver) = bounded(16);
     let (in_user_datagrams_sender, in_user_datagrams_receiver) = bounded(16);
-    task::spawn(dreceiver::run(
+    spawn(Box::pin(dreceiver::run(
         port,
         in_system_datagrams_sender,
         in_user_datagrams_sender,
         messages,
-    ));
+    )));
 
     let resends = Resends::new();
     let (sreceiver_cancellation_sender, sreceiver_cancellation_receiver) = cancellation();
-    task::spawn(sreceiver::run(
+    spawn(Box::pin(sreceiver::run(
         port,
         sreceiver_cancellation_receiver,
         in_system_datagrams_receiver,
         resends.clone(),
-    ));
+    )));
 
     let (inputs_sender, inputs_receiver) = bounded(CHANNEL_CAPACITY);
     let (confirmer_cancellation_sender, confirmer_cancellation_receiver) = cancellation();
     let confirms = Confirmations::new();
-    task::spawn(ureceiver::run(
+    spawn(Box::pin(ureceiver::run(
         port,
         confirmer_cancellation_sender,
         in_user_datagrams_receiver,
         inputs_sender,
         confirms.clone(),
-    ));
+    )));
 
     let (outputs_sender, outputs_receiver) = bounded(CHANNEL_CAPACITY);
     let (errors_sender, errors_receiver) = bounded(CHANNEL_CAPACITY);
     let (resender_cancellation_sender, resender_cancellation_receiver) = cancellation();
-    task::spawn(resender::run(
+    spawn(Box::pin(resender::run(
         port,
         resender_cancellation_receiver,
         sreceiver_cancellation_sender,
         out_datagrams_sender.clone(),
         errors_sender,
         resends.clone(),
-    ));
+    )));
 
-    task::spawn(confirmer::run(
+    spawn(Box::pin(confirmer::run(
         port,
         confirmer_cancellation_receiver,
         out_datagrams_sender.clone(),
         confirms,
-    ));
-    task::spawn(usender::run(
+    )));
+    spawn(Box::pin(usender::run(
         port,
         resender_cancellation_sender,
         out_datagrams_sender,
         outputs_receiver,
         resends,
-    ));
+    )));
 
     (
         MessageSender(outputs_sender),