evoq_event_type_registry (evoq v1.30.1)
View SourceEvent type registry for evoq.
Maintains a mapping of event types to interested handlers. Uses pg (process groups) for efficient pub/sub routing.
This is the key to per-event-type subscriptions: - Handlers register interest in specific event types - When events are published, only interested handlers receive them - Scales to millions of events without per-stream overhead
Registrants
A registrant is a handler process (event handler or projection), joined to one pg group per event type, or a process manager module, registered by the PM router with register_process_manager/2 and held in this process's state. Every registration is announced to the store subscriptions on this node as {new_registrant, Registrant, Types, Info}, so each one backfills that registrant's history to it alone (evoq #4). A process manager is a registrant of its own: a type only a process manager consumes is routed like any other (evoq #2).
Info carries the registrant's identity, which outlives the process (a handler's module; a process manager's {pm, Module}), so a restarted handler is recognised as the same consumer and is not handed again as new what its last incarnation already had; and from, the global position its history starts at (a projection's checkpoint plus one), so a backfill does not read and route what it already holds.
Summary
Functions
Every event type something consumes: a handler or a process manager.
Every registered process manager module and its event types.
The handler processes registered for an event type (pg members).
The process manager modules registered for an event type.
Register a handler pid for an event type. Same as register_all([EventType], HandlerPid).
Register a handler pid for all its event types in one step.
register_all/2 with the registrant's identity and starting point.
Refused: {error, not_supported}. A module is never registered as a handler; start one with evoq_event_handler:start_link/2 (evoq #3).
Register a store subscription listener.
Register a process manager module for its event types.
Start the event type registry.
Unregister a handler pid from an event type.
Refused: {error, not_supported}. A module is never registered as a handler; start one with evoq_event_handler:start_link/2 (evoq #3).
Unregister a store subscription listener.
Withdraw a process manager module.
Types
-type identity() :: term().
-type registration_opts() :: #{identity => identity(), from => non_neg_integer()}.
Functions
-spec get_all_event_types() -> [binary()].
Every event type something consumes: a handler or a process manager.
Every registered process manager module and its event types.
The PM router rebuilds its routing from this when it restarts: the registrations live here, in a process that outlives it (evoq#9).
The handler processes registered for an event type (pg members).
The process manager modules registered for an event type.
Register a handler pid for an event type. Same as register_all([EventType], HandlerPid).
Register a handler pid for all its event types in one step.
Store subscription listeners get ONE {new_registrant, HandlerPid, Types} message naming every type of the registration, whether or not another handler already holds it, so each subscription backfills the history of those types to this handler alone, in one pass, in global order (evoq #4). Registering types one by one made one backfill per type, and a projection that checkpoints on the global position then skipped every event of its second type below the first type's last one.
-spec register_all([binary()], pid(), registration_opts()) -> ok.
register_all/2 with the registrant's identity and starting point.
identity names the consumer across restarts (default: the pid itself, which never restarts); evoq_event_handler and evoq_projection pass their module. from is the global position the registrant's history starts at (default 0); evoq_projection passes its checkpoint plus one.
Refused: {error, not_supported}. A module is never registered as a handler; start one with evoq_event_handler:start_link/2 (evoq #3).
-spec register_listener(pid()) -> {ok, [{registrant(), [binary()], identity()}]}.
Register a store subscription listener.
Atomically returns every registration that exists now, as {Registrant, Types}, AND subscribes the listener for every later one. This is race-free: no registration can execute between returning the current ones and subscribing for notifications, because both happen in the same gen_server call. Only registrants on this node are returned and announced: a remote handler is served by its own node's subscription.
Each registration is {Registrant, Types, Identity}. The listener receives {new_registrant, Registrant, Types, #{identity, from}} for every later registration (see register_all/3 and register_process_manager/2) and {registrant_left, Registrant, Types} when one withdraws.
Register a process manager module for its event types.
Called by the PM router. The module becomes a registrant of its own, {pm, Module}: store subscriptions route its types to the PM router for it whether or not a handler consumes them (evoq #2), and backfill their history to it alone. Registering again replaces its types.
Start the event type registry.
Unregister a handler pid from an event type.
Refused: {error, not_supported}. A module is never registered as a handler; start one with evoq_event_handler:start_link/2 (evoq #3).
-spec unregister_listener(pid()) -> ok.
Unregister a store subscription listener.
-spec unregister_process_manager(module()) -> ok.
Withdraw a process manager module.