ygopro_handler/
processor.rs

1//! The message dispatcher: routes messages to handlers by key.
2//!
3//! A [`Processor`] holds a set of handlers keyed by message type (a [`MessageKey`]),
4//! plus a set of global handlers that run for every message. It dispatches each message
5//! through the matching handler chain and combines their responses.
6
7use std::hash::Hash;
8use std::marker::PhantomData;
9
10use futures::Stream;
11use futures::StreamExt;
12use hashbrown::HashMap;
13use hashbrown::HashSet;
14use ygopro_data::complex;
15use ygopro_data::every_client_to_server_flat_message;
16use ygopro_data::every_game_message_flat_message;
17use ygopro_data::every_server_to_client_flat_message;
18use ygopro_data::message::ctos;
19use ygopro_data::message::game_message as gm;
20use ygopro_data::message::stoc;
21
22use crate::handler::Bundle;
23use crate::handler::Call;
24use crate::handler::sync_handler::SyncHandler;
25use crate::handler::sync_handler::WithSubState;
26use crate::extract::Request;
27
28/// Clone the global handlers into every key list and sort each list by priority.
29///
30/// After resolving, the handler set should be considered frozen; mutating it afterwards
31/// would break the ordering or the global-handler copies.
32pub fn resolve_globals<K, H: Clone>(handlers: &mut HashMap<K, Vec<H>>, global_handlers: &[H], key: impl Fn(&H) -> u8) {
33    for list in handlers.values_mut() {
34        list.extend(global_handlers.iter().cloned());
35        list.sort_unstable_by_key(&key);
36    }
37}
38
39/// A pure flag whose only meaning is its name.
40/// Used as a handler key it resolves to 0, letting the call site register
41/// the handler as a global one via `register_global`.
42#[derive(Debug)]
43pub struct All {
44    _private: (),
45}
46
47impl ygopro_data::message::PureMessage for All {}
48
49impl ygopro_data::message::Message for All {
50    fn message_type() -> ygopro_data::message::all::MessageType {
51        ygopro_data::message::all::MessageType::Other("all", 0)
52    }
53}
54
55/// The key that [`Processor`] used to look up the handlers for a message.
56pub trait MessageKey<Key> {
57    /// Get the message key.
58    fn message_key(&self) -> Key;
59}
60
61impl MessageKey<u8> for u8 {
62    fn message_key(&self) -> u8 {
63        *self
64    }
65}
66
67impl MessageKey<u8> for complex::Complex<ctos::Message> {
68    fn message_key(&self) -> u8 {
69        self.data[0]
70    }
71}
72
73impl MessageKey<u8> for complex::Complex<stoc::Message> {
74    fn message_key(&self) -> u8 {
75        self.data[0]
76    }
77}
78
79impl MessageKey<u8> for complex::Complex<gm::Message> {
80    fn message_key(&self) -> u8 {
81        self.data[0]
82    }
83}
84
85macro_rules! impl_message_key {
86    ($message_mod:ident, $($variant:ident = $flag:literal),* $(,)?) => {
87        impl $crate::processor::MessageKey<u8> for $message_mod::MessageType {
88            fn message_key(&self) -> u8 {
89                u8::from(self)
90            }
91        }
92        impl $crate::processor::MessageKey<u8> for $message_mod::Message {
93            fn message_key(&self) -> u8 {
94                match self {
95                    $($message_mod::Message::$variant(_) => $flag),*
96                }
97            }
98        }
99        $(
100            impl $crate::processor::MessageKey<u8> for $message_mod::$variant {
101                fn message_key(&self) -> u8 {
102                    $flag
103                }
104            }
105        )*
106    };
107}
108
109macro_rules! impl_ctos_message_key { ($($rest:tt)*) => { impl_message_key!(ctos, $($rest)*); }; }
110macro_rules! impl_stoc_message_key { ($($rest:tt)*) => { impl_message_key!(stoc, $($rest)*); }; }
111macro_rules! impl_gm_message_key { ($($rest:tt)*) => { impl_message_key!(gm, $($rest)*); }; }
112
113every_client_to_server_flat_message!(impl_ctos_message_key);
114every_server_to_client_flat_message!(impl_stoc_message_key);
115every_game_message_flat_message!(impl_gm_message_key);
116
117impl<Message, Extra> MessageKey<u8> for Request<Message, Extra>
118where
119    Message: MessageKey<u8>,
120{
121    fn message_key(&self) -> u8 {
122        self.message.message_key()
123    }
124}
125
126/// The message dispatcher.
127///
128/// It holds handlers keyed by [`MessageKey`] and a set of global handlers, and routes
129/// each message through the matching chain.
130pub struct Processor<Key, Req, State = crate::handler::State, Res = (), H: Call<Req, State, Res> = crate::handler::tower_handler::TowerHandler<Req, State, Res>> {
131    handlers: HashMap<Key, Vec<H>>,
132    global_handlers: Vec<H>,
133    _phantom: PhantomData<fn(Req, State, Res)>,
134}
135
136impl<Key, Req, State, Res, H: Call<Req, State, Res>> Processor<Key, Req, State, Res, H>
137where
138    Key: Eq + Hash,
139    State: Send,
140{
141    /// Create an empty processor.
142    pub fn new() -> Self {
143        Self {
144            handlers: HashMap::new(),
145            global_handlers: Vec::new(),
146            _phantom: PhantomData,
147        }
148    }
149
150    /// The number of registered handlers, including globals.
151    pub fn handler_count(&self) -> usize {
152        self.handlers.values().map(|handlers| handlers.len()).sum::<usize>() + self.global_handlers.len()
153    }
154
155    /// Count the handlers grouped by module, dropping the duplicate global copies.
156    pub fn handler_statistics(&self, module_name_of: fn(&H) -> &'static str) -> hashbrown::HashMap<&'static str, usize> {
157        let mut handler_counts = hashbrown::HashMap::new();
158        for handlers in self.handlers.values() {
159            for handler in handlers {
160                *handler_counts.entry(module_name_of(handler)).or_insert(0) += 1;
161            }
162        }
163        let mut global_counts = hashbrown::HashMap::new();
164        for handler in &self.global_handlers {
165            *global_counts.entry(module_name_of(handler)).or_insert(0) += 1;
166        }
167        // resolve() clones each global handler into every key list, so a module
168        // counts its own globals once per key list; drop the duplicate copies.
169        let key_list_count = self.handlers.len();
170        for (module_name, global_count) in global_counts {
171            let duplicated_copies = global_count * key_list_count.saturating_sub(1);
172            if let Some(count) = handler_counts.get_mut(&module_name) {
173                *count -= duplicated_copies;
174            }
175        }
176        handler_counts
177    }
178
179    /// Build a processor from builder functions, filtering by group and splitting globals.
180    pub fn new_with_groups(builders: &[fn() -> (Key, H)], groups: &HashSet<String>, group_of: fn(&H) -> &'static str, is_all: impl Fn(&Key) -> bool) -> Self where H: Clone {
181        let mut processor = Self::new();
182        for build in builders {
183            let (key, handler) = build();
184            if !groups.is_empty() && !groups.contains(group_of(&handler)) {
185                continue;
186            }
187            if is_all(&key) {
188                processor.register_global(handler);
189            } else {
190                processor.register(key, handler);
191            }
192        }
193        processor.resolve();
194        processor
195    }
196
197    /// Register a handler for a message key.
198    pub fn register(&mut self, message_key: Key, handler: H) {
199        self.handlers.entry(message_key).or_default().push(handler);
200    }
201
202    /// Register a global handler that runs for every message.
203    pub fn register_global(&mut self, handler: H) {
204        self.global_handlers.push(handler);
205    }
206
207    /// Clone the global handlers into every key list and sort each list by priority.
208    ///
209    /// After resolving, the processor should not be modified anymore.
210    pub fn resolve(&mut self) where H: Clone {
211        resolve_globals(&mut self.handlers, &self.global_handlers, |h| h.priority());
212    }
213
214    /// Run a bundle through the handler chain for the given key, stopping on [`StopFlag`](crate::handler::StopFlag).
215    pub async fn process_bundle(&self, bundle: Bundle<Req, State, Res>, key: Key) -> Bundle<Req, State, Res>
216    where
217        Key: Eq + Hash,
218    {
219        let handlers = self.handlers.get(&key).unwrap_or(&self.global_handlers);
220        let mut bundle = bundle;
221        for handler in handlers {
222            bundle = handler.call(bundle).await;
223            if bundle.stop_flag.0 { break }
224        }
225        bundle
226    }
227
228    /// Process a stream of messages, assembling a bundle per item and consuming the result.
229    pub fn process<Item, InnerStream, AssembleBundle, ConsumeBundle>(
230        self: std::sync::Arc<Self>,
231        stream: InnerStream,
232        assemble_bundle: AssembleBundle,
233        consume_bundle: ConsumeBundle,
234    ) -> impl Stream<Item = Bundle<Req, State, Res>>
235    where
236        Key: Clone + Eq + Hash + Send + 'static,
237        Req: Send + 'static,
238        Res: Send + 'static,
239        State: Send + 'static,
240        Item: MessageKey<Key> + Into<Req> + Send + 'static,
241        InnerStream: Stream<Item = Item> + Send + 'static,
242        AssembleBundle: Fn(Req) -> Bundle<Req, State, Res> + Send + 'static,
243        ConsumeBundle: Fn(Bundle<Req, State, Res>) -> Bundle<Req, State, Res> + Clone + Send + 'static,
244    {
245        stream.then(move |item| {
246            let key = item.message_key();
247            let request: Req = item.into();
248            let bundle = assemble_bundle(request);
249            let processor = self.clone();
250            let consume_bundle = consume_bundle.clone();
251            async move {
252                let bundle = processor.process_bundle(bundle, key).await;
253                let bundle = consume_bundle(bundle);
254                bundle
255            }
256        })
257    }
258}
259
260impl<Key, Req, Target, Res> Processor<Key, Req, Target, Res, SyncHandler<Req, Target, Res>>
261where
262    Key: Eq + Hash,
263    Req: Send + 'static,
264    Target: Send + 'static,
265    Res: Send + 'static + std::ops::Mul<Output = Res>,
266{
267    /// Build a processor from two handler sets (target and source states), merging them.
268    ///
269    /// The source handlers are written against `SubState` but run against `Target`. This is
270    /// only sound for [`SyncHandler`], whose `Call` impl reinterprets
271    /// `&mut Bundle<Req, Target, Res>` as `&mut Bundle<Req, SubState, Res>` via the
272    /// [`WithSubState`] layout guarantee.
273    ///
274    /// [`TowerHandler`](crate::handler::tower_handler::TowerHandler) and
275    /// [`AsyncHandler`](crate::handler::async_handler::AsyncHandler) erase the handler
276    /// behind a trait object (`BoxCloneService` / `Arc<dyn Call>`), so they cannot be
277    /// transmuted between state types; the dual-state trick is unique to `SyncHandler`,
278    /// which stores raw pointers and monomorphized function pointers. The layout
279    /// assumption is checked by
280    /// [`assert_sync_handler_layout`](crate::handler::sync_handler::assert_sync_handler_layout)
281    /// in [`extend`](Processor::extend).
282    pub fn new_with_dual_group<SubState>(
283        target_builders: &[fn() -> (Key, SyncHandler<Req, Target, Res>)],
284        source_builders: &[fn() -> (Key, SyncHandler<Req, SubState, Res>)],
285        groups: &HashSet<String>,
286        target_group_of: fn(&SyncHandler<Req, Target, Res>) -> &'static str,
287        source_group_of: fn(&SyncHandler<Req, SubState, Res>) -> &'static str,
288        is_all: impl Fn(&Key) -> bool,
289    ) -> Self
290    where
291        SubState: Send + 'static,
292        Target: WithSubState<SubState>,
293    {
294        let mut processor = Self::new();
295        for build in target_builders {
296            let (key, handler) = build();
297            if !groups.is_empty() && !groups.contains(target_group_of(&handler)) { continue; }
298            if is_all(&key) { processor.register_global(handler); } else { processor.register(key, handler); };
299        }
300        let mut source_processor = Processor::<Key, Req, SubState, Res, SyncHandler<Req, SubState, Res>>::new();
301        for build in source_builders {
302            let (key, handler) = build();
303            if !groups.is_empty() && !groups.contains(source_group_of(&handler)) { continue; }
304            if is_all(&key) { source_processor.register_global(handler); } else { source_processor.register(key, handler); };
305        }
306        processor.extend(source_processor);
307        processor.resolve();
308        processor
309    }
310
311    /// Merge the handlers of a source-state processor into this target-state processor.
312    ///
313    /// Each source handler is transmuted to run against `Target`. This assumes the
314    /// [`WithSubState`] layout guarantee and is
315    /// checked by
316    /// [`assert_sync_handler_layout`](crate::handler::sync_handler::assert_sync_handler_layout).
317    pub fn extend<SubState>(&mut self, processor_source: Processor<Key, Req, SubState, Res, SyncHandler<Req, SubState, Res>>)
318    where
319        SubState: Send + 'static,
320        Target: WithSubState<SubState>,
321    {
322        crate::handler::sync_handler::assert_sync_handler_layout::<Req, SubState, Target, Res>();
323        for (key, handlers) in processor_source.handlers {
324            self.handlers.entry(key).or_default().extend(handlers.into_iter()
325                .map(|handler| unsafe { std::mem::transmute::<SyncHandler<Req, SubState, Res>, SyncHandler<Req, Target, Res>>(handler) }));
326        }
327        self.global_handlers.extend(processor_source.global_handlers.into_iter()
328            .map(|handler| unsafe { std::mem::transmute::<SyncHandler<Req, SubState, Res>, SyncHandler<Req, Target, Res>>(handler) }));
329    }
330}
331
332/// Create a bundle from a request, with default state and response.
333pub fn default_bundle<Req, State: Default, Res: Default>(request: Req) -> Bundle<Req, State, Res> {
334    Bundle {
335        request,
336        state: State::default(),
337        response: Res::default(),
338        stop_flag: Default::default()
339    }
340}
341
342#[cfg(test)]
343mod tests {
344    use super::MessageKey;
345    use ygopro_data::message::ctos;
346
347    #[test]
348    fn join_game_is_message_key() {
349        let join = ctos::JoinGame {
350            version: 0x1338,
351            gameid: 0,
352            pass: ygopro_data::string::FixedLengthString::from(""),
353        };
354        let key: u8 = join.message_key();
355        assert_eq!(key, 18);
356    }
357}
ygopro_handler/processor.rs

ygopro_handler/
processor.rs
