lumen_core/
command.rs

1//! Bounded command queue for off-main-thread mutations of ECS state.
2//!
3//! - Capacity defaults to [`DEFAULT_QUEUE_CAPACITY`] (8192) and is configurable via [`CommandQueue::with_capacity`].
4//! - Overflow drops the command, logs, and emits a [`CommandQueueOverflow`] message.
5//! - The [`TickStage::CommandDrain`](crate::tick::TickStage::CommandDrain) stage drains the queue on the main thread between `Input` and `Systems`.
6
7use crate::property_store::{PropertyKey, PropertyStore, PropertyValue};
8use bevy_ecs::prelude::*;
9use crossbeam_channel::{Receiver, Sender, TrySendError, bounded};
10use std::any::{Any, TypeId};
11use std::collections::HashMap;
12use std::sync::Arc;
13
14/// Default queue capacity. Override via [`CommandQueue::with_capacity`].
15pub const DEFAULT_QUEUE_CAPACITY: usize = 8192;
16
17/// A deferred command applied during [`TickStage::CommandDrain`](crate::tick::TickStage::CommandDrain).
18pub enum Command {
19    /// Carries an asynchronously decoded asset payload.
20    AssetLoaded {
21        /// Identifier supplied by the requester when the asset was requested.
22        callback_id: u64,
23        /// Decoded payload (image bytes, etc.) typed by the consumer.
24        payload: Box<dyn Any + Send>,
25    },
26    /// Carries an HTTP response payload.
27    NetworkResponse {
28        /// Identifier supplied at request time.
29        callback_id: u64,
30        /// HTTP status code.
31        status: u16,
32        /// Response body bytes.
33        body: Vec<u8>,
34    },
35    /// Carries a script-produced state mutation.
36    ScriptUpdate(Box<dyn Any + Send>),
37    /// Writes a typed value into the [`PropertyStore`].
38    ///
39    /// Drained in [`TickStage::CommandDrain`](crate::tick::TickStage::CommandDrain) via [`apply_property_commands`].
40    /// The per-`App` queue is the intended home for external property writes, but the global `EXTERNAL_TX` /
41    /// `EXTERNAL_RX` ring in `property_store` is still what the C-ABI crate and the async tasks use; this variant has
42    /// no producer yet.
43    SetProperty {
44        /// Target property key.
45        key: PropertyKey,
46        /// New value to write.
47        value: PropertyValue,
48    },
49    /// Strongly-typed custom command, dispatched via [`CommandRegistry`].
50    ///
51    /// Replaces blind `Custom(Box<dyn Any>)` downcasts: the receiving plugin registers a handler
52    /// for its concrete payload type with [`crate::app::App::register_command`] and the drain
53    /// dispatches by [`TypeId`].
54    Typed {
55        /// Concrete payload type id used to route the command to its registered handler.
56        type_id: TypeId,
57        /// Boxed payload.
58        payload: Box<dyn Any + Send>,
59    },
60    /// Free-form variant retained for backward compatibility with callers that have not yet migrated to [`Self::Typed`].
61    Custom(Box<dyn Any + Send>),
62}
63
64/// Message emitted when [`CommandQueue::try_push`] returns `Err` because the queue is full.
65#[derive(Message, Clone, Copy, Debug)]
66pub struct CommandQueueOverflow;
67
68/// Producer-side ECS resource. Cloneable for sharing across worker threads.
69#[derive(Resource, Clone)]
70pub struct CommandQueue {
71    tx: Sender<Command>,
72}
73
74impl CommandQueue {
75    /// Returns a `(CommandQueue, CommandReceiver)` pair with [`DEFAULT_QUEUE_CAPACITY`].
76    /// Insert the queue as a resource; pass the receiver to the drain system.
77    pub fn new() -> (Self, CommandReceiver) {
78        Self::with_capacity(DEFAULT_QUEUE_CAPACITY)
79    }
80
81    /// Returns a `(CommandQueue, CommandReceiver)` pair with the explicit capacity `cap`.
82    pub fn with_capacity(cap: usize) -> (Self, CommandReceiver) {
83        let (tx, rx) = bounded(cap);
84        (Self { tx }, CommandReceiver { rx })
85    }
86
87    /// Sends `cmd` non-blockingly. Returns `Err(TrySendError::Full)` when the channel is full; never blocks.
88    pub fn try_push(&self, cmd: Command) -> Result<(), TrySendError<Command>> {
89        self.tx.try_send(cmd)
90    }
91
92    /// Borrows the underlying sender for adapter code (the C-ABI crate, async tasks) that needs to clone the channel across threads.
93    pub fn sender(&self) -> &Sender<Command> {
94        &self.tx
95    }
96}
97
98/// Consumer-side ECS resource, registered as a non-Send resource since the drain runs on the main thread.
99#[derive(Resource)]
100pub struct CommandReceiver {
101    rx: Receiver<Command>,
102}
103
104impl CommandReceiver {
105    /// Returns an iterator that yields commands via `try_recv` and stops as soon as the channel is empty; never blocks.
106    pub fn drain(&mut self) -> impl Iterator<Item = Command> + '_ {
107        std::iter::from_fn(|| self.rx.try_recv().ok())
108    }
109
110    /// Borrows the underlying receiver for adapter code.
111    pub fn receiver(&self) -> &Receiver<Command> {
112        &self.rx
113    }
114}
115
116/// Handler invoked when a [`Command::Typed`] with a matching [`TypeId`] is drained.
117pub type CommandHandlerFn = Arc<dyn Fn(&mut World, Box<dyn Any + Send>) + Send + Sync>;
118
119/// Resource holding the per-`TypeId` handler table for [`Command::Typed`].
120///
121/// Populated by [`crate::app::App::register_command`]. Plugins that author new command kinds register their handler once
122/// at build-time and then push `Command::Typed { type_id: TypeId::of::<MyCmd>(), payload }` from any thread.
123#[derive(Resource, Default, Clone)]
124pub struct CommandRegistry {
125    handlers: HashMap<TypeId, CommandHandlerFn>,
126}
127
128impl CommandRegistry {
129    /// Registers `handler` for payloads of type `T`. Subsequent registrations for the same type overwrite the prior entry.
130    pub fn register<T, F>(&mut self, handler: F)
131    where
132        T: Any + Send,
133        F: Fn(&mut World, Box<T>) + Send + Sync + 'static,
134    {
135        let f: CommandHandlerFn = Arc::new(move |world, payload| {
136            if let Ok(typed) = payload.downcast::<T>() {
137                handler(world, typed);
138            }
139        });
140        self.handlers.insert(TypeId::of::<T>(), f);
141    }
142
143    /// Returns the registered handler for `type_id`, if any.
144    pub fn lookup(&self, type_id: &TypeId) -> Option<CommandHandlerFn> {
145        self.handlers.get(type_id).cloned()
146    }
147}
148
149/// Drains [`Command::SetProperty`] and [`Command::Typed`] entries from the [`CommandReceiver`] and applies them to the
150/// [`PropertyStore`] / [`CommandRegistry`] respectively.
151///
152/// Other [`Command`] variants are intentionally dropped by this drain - their owning plugins install dedicated drains.
153///
154/// Not auto-installed by [`crate::app::App::new`]. `lumen-runtime` registers it in the
155/// [`crate::tick::TickStage::CommandDrain`] stage when it builds an app; an embedder assembling its own `App`
156/// adds it the same way.
157pub fn apply_property_commands(world: &mut World) {
158    let mut to_apply: Vec<(PropertyKey, PropertyValue)> = Vec::new();
159    let mut typed_dispatch: Vec<(TypeId, Box<dyn Any + Send>)> = Vec::new();
160    if let Some(mut recv) = world.get_resource_mut::<CommandReceiver>() {
161        for cmd in recv.drain() {
162            match cmd {
163                Command::SetProperty { key, value } => to_apply.push((key, value)),
164                Command::Typed { type_id, payload } => typed_dispatch.push((type_id, payload)),
165                _ => {}
166            }
167        }
168    }
169    if let Some(mut store) = world.get_resource_mut::<PropertyStore>() {
170        for (k, v) in to_apply {
171            store.set(k, v);
172        }
173    }
174    let registry = world.get_resource::<CommandRegistry>().cloned();
175    if let Some(registry) = registry {
176        for (tid, payload) in typed_dispatch {
177            if let Some(handler) = registry.lookup(&tid) {
178                handler(world, payload);
179            }
180        }
181    }
182}
lumen_core/command.rs

lumen_core/
command.rs
