Struct Processor 

pub struct Processor<Key, Req, State = State, Res = (), H: Call<Req, State, Res> = TowerHandler<Req, State, Res>> { /* private fields */ }

The message dispatcher.

It holds handlers keyed by MessageKey and a set of global handlers, and routes each message through the matching chain.

Implementations

impl<Key, Req, State, Res, H: Call<Req, State, Res>> Processor<Key, Req, State, Res, H>

where Key: Eq + Hash, State: Send,

pub fn new() -> Self

Create an empty processor.

pub fn handler_count(&self) -> usize

The number of registered handlers, including globals.

pub fn handler_statistics( &self, module_name_of: fn(&H) -> &'static str, ) -> HashMap<&'static str, usize>

Count the handlers grouped by module, dropping the duplicate global copies.

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,

Build a processor from builder functions, filtering by group and splitting globals.

pub fn register(&mut self, message_key: Key, handler: H)

Register a handler for a message key.

pub fn register_global(&mut self, handler: H)

Register a global handler that runs for every message.

pub fn resolve(&mut self)

where H: Clone,

Clone the global handlers into every key list and sort each list by priority.

After resolving, the processor should not be modified anymore.

pub async fn process_bundle( &self, bundle: Bundle<Req, State, Res>, key: Key, ) -> Bundle<Req, State, Res>

where Key: Eq + Hash,

Run a bundle through the handler chain for the given key, stopping on StopFlag.

pub fn process<Item, InnerStream, AssembleBundle, ConsumeBundle>( self: Arc<Self>, stream: InnerStream, assemble_bundle: AssembleBundle, consume_bundle: ConsumeBundle, ) -> impl Stream<Item = Bundle<Req, State, Res>>

where Key: Clone + Eq + Hash + Send + 'static, Req: Send + 'static, Res: Send + 'static, State: Send + 'static, Item: MessageKey<Key> + Into<Req> + Send + 'static, InnerStream: Stream<Item = Item> + Send + 'static, AssembleBundle: Fn(Req) -> Bundle<Req, State, Res> + Send + 'static, ConsumeBundle: Fn(Bundle<Req, State, Res>) -> Bundle<Req, State, Res> + Clone + Send + 'static,

Process a stream of messages, assembling a bundle per item and consuming the result.

impl<Key, Req, Target, Res> Processor<Key, Req, Target, Res, SyncHandler<Req, Target, Res>>

where Key: Eq + Hash, Req: Send + 'static, Target: Send + 'static, Res: Send + 'static + Mul<Output = Res>,

pub fn new_with_dual_group<SubState>( target_builders: &[fn() -> (Key, SyncHandler<Req, Target, Res>)], source_builders: &[fn() -> (Key, SyncHandler<Req, SubState, Res>)], groups: &HashSet<String>, target_group_of: fn(&SyncHandler<Req, Target, Res>) -> &'static str, source_group_of: fn(&SyncHandler<Req, SubState, Res>) -> &'static str, is_all: impl Fn(&Key) -> bool, ) -> Self

where SubState: Send + 'static, Target: WithSubState<SubState>,

Build a processor from two handler sets (target and source states), merging them.

The source handlers are written against SubState but run against Target. This is only sound for SyncHandler, whose Call impl reinterprets &mut Bundle<Req, Target, Res> as &mut Bundle<Req, SubState, Res> via the WithSubState layout guarantee.

TowerHandler and AsyncHandler erase the handler behind a trait object (BoxCloneService / Arc<dyn Call>), so they cannot be transmuted between state types; the dual-state trick is unique to SyncHandler, which stores raw pointers and monomorphized function pointers. The layout assumption is checked by assert_sync_handler_layout in extend.

pub fn extend<SubState>( &mut self, processor_source: Processor<Key, Req, SubState, Res, SyncHandler<Req, SubState, Res>>, )

where SubState: Send + 'static, Target: WithSubState<SubState>,

Merge the handlers of a source-state processor into this target-state processor.

Each source handler is transmuted to run against Target. This assumes the WithSubState layout guarantee and is checked by assert_sync_handler_layout.