bevy-async-runner 0.1.2

Bevy Async Runner simplifies working with asynchronous code in the Bevy game engine. It provides a mechanism to schedule and execute async tasks and provide their result to any system.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

use std::future::Future;
use bevy::prelude::{warn, trace, Commands, IntoSystem, Resource, SystemInput};
use bevy::tasks::IoTaskPool;
use maybe_sync::{MaybeSend, MaybeSync};
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender, UnboundedReceiver};
use tokio::sync::mpsc::error::TryRecvError;

type ExecuteSystemFn = Box<dyn FnOnce(&mut Commands) + Send + Sync>;

#[derive(Resource)]
pub struct AsyncRunner {
    channel: (
        UnboundedSender<ExecuteSystemFn>,
        UnboundedReceiver<ExecuteSystemFn>
    ),
    #[cfg(all(not(feature = "tokio-runtime"), not(feature = "tokio-runtime-multi-thread")))]
    pool: &'static IoTaskPool,
    #[cfg(any(feature = "tokio-runtime-multi-thread", feature = "tokio-runtime"))]
    pool: tokio::runtime::Runtime,
}

impl AsyncRunner {
    pub fn new() -> AsyncRunner {

        #[cfg(all(not(feature = "tokio-runtime"), not(feature = "tokio-runtime-multi-thread")))]
        let pool = {
            trace!("Fetching IoTaskPool runtime via IoTaskPool::get()");
            IoTaskPool::get()
        };
        #[cfg(feature = "tokio-runtime-multi-thread")]
        let pool= {
            trace!("Starting tokio runtime via Builder::new_multi_thread()");
            tokio::runtime::Builder::new_multi_thread()
                .enable_all()
                .build()
                .unwrap()
        };
        #[cfg(all(not(feature = "tokio-runtime-multi-thread"), feature = "tokio-runtime"))]
        let pool = {
            trace!("Starting tokio runtime via Builder::new_current_thread()");
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap()
        };

        AsyncRunner {
            channel: unbounded_channel(),
            pool
        }
    }

    /// Takes a join handle, and runs an ECS system with its value once completed
    pub fn schedule<
        S: IntoSystem<I, (), M> + Send + Sync + 'static,
        M: 'static,
        I: SystemInput<Inner<'static>: Send + Sync> + Send + Sync + 'static
    >(
        &self,
        task: impl Future<Output = I::Inner<'static>> + MaybeSend + MaybeSync + 'static,
        system: S,
    ) {
        let task = self.pool.spawn((async |sender: UnboundedSender<ExecuteSystemFn>| {
            let result = task.await;

            let boxed_result = Box::new(result);

            let execute = move |commands: &mut Commands| {
                commands.run_system_cached_with(
                    system,
                    *boxed_result
                );
            };

            sender.send(Box::new(execute)).unwrap();
        })(self.channel.0.clone()));

        // IoTaskPool won't finish it unless we detach
        #[cfg(all(not(feature = "tokio-runtime"), not(feature = "tokio-runtime-multi-thread")))]
        task.detach();
    }

    /// Loop over all completed join handles and run the systems
    pub fn run(&mut self, mut commands: Commands) {
        loop {
            match self.channel.1.try_recv() {
                Ok(execute) => {
                    execute(&mut commands);
                }
                Err(e) => {
                    match e {
                        TryRecvError::Empty => {}
                        TryRecvError::Disconnected => {
                            warn!("AsyncRunner communication channel terminated");
                            break;
                        }
                    }

                    break;
                }
            }
        }
    }
}