EventDispatcher, EventManager, polling et handlers
Bootcode IWA-S04 â Semaine 20, Jour 4
1. Comprendre le flux complet d'un événement de domaine asynchrone
De l'agrégat jusqu'au handler, en passant par la queue Postgres
2. Identifier les handlers existants et les événements qu'ils traitent
Cartographier les handlers du codebase
3. Comprendre le rĂŽle du MessageModule et du polling
Le polling toutes les 3s lit la queue et dispatch aux handlers
4. Savoir créer un nouveau handler pour un événement existant
Classe @injectable() avec une méthode handle(event)
Rappel : événements synchrones vs asynchrones
@Handle dans l'agrégat vs handlers entre composants
Le flux complet asynchrone
agrĂ©gat â EventDispatcher â DB queue â polling â EventManager â Handler
Lire le code : PostgresEventRepository, MessageModule, EventManager
Les trois briques du systĂšme asynchrone
Un vrai handler : UserRoleUpgradedHandler
Classe @injectable() avec une méthode handle(event)
Le pattern webhook
PersistWebhook â WebhookPersisted â handler filtre et traite
Deux types d'événements, deux mécanismes
âĄ
Synchrone (@Handle)
Dans l'agrégat. Exécuté immédiatement pendant la commande.
đš
Asynchrone (handlers)
Entre composants. Persisté en DB, traité plus tard par polling.
L'agrégat réagit à ses propres événements pendant l'exécution de la commande.
export class UserAggregate extends AggregateRoot {
private email: string;
upgradeRole(newRole: Role) {
this.role = newRole;
this.raise(new UserRoleUpgraded(this.id, newRole));
}
// @Handle = réaction synchrone DANS l'agrégat
@Handle(UserRoleUpgraded)
private onRoleUpgraded(event: UserRoleUpgraded) {
// Mettre Ă jour un compteur interne, etc.
this.lastRoleChange = new Date();
}
}
đĄ @Handle = l'agrĂ©gat se met Ă jour lui-mĂȘme. Pas de communication entre composants.
De l'agrégat jusqu'au handler, en passant par la queue Postgres.
1ïžâŁ AgrĂ©gat
user.upgradeRole(newRole) â raise(UserRoleUpgraded)
2ïžâŁ EventDispatcher
Persiste l'événement dans la queue Postgres (table events)
3ïžâŁ DB queue (Postgres)
Les Ă©vĂ©nements attendent d'ĂȘtre traitĂ©s (status: pending)
4ïžâŁ Polling (toutes les 3s)
Le MessageModule lit les événements pending
5ïžâŁ EventManager
Dispatch chaque événement aux handlers enregistrés
6ïžâŁ Handler
UserRoleUpgradedHandler.handle(event) â action (email, webhook...)
DĂ©couplage
L'agrégat ne sait pas qui va traiter l'événement. Il ne connaßt pas les handlers.
RĂ©silience
Si un handler plante, l'événement reste en queue. Il sera retraité au prochain polling.
Performance
La requĂȘte HTTP rĂ©pond immĂ©diatement. Les side effects (email, webhook) se font en arriĂšre-plan.
Idempotence
Chaque événement a un refId unique. Un handler peut détecter les doublons.
Persiste et lit les événements dans la table events.
export class PostgresEventRepository implements EventRepository {
constructor(@inject(DataSource) private ds: DataSource) {}
async save(event: DomainEvent) {
await this.ds.getRepository(EventEntity)
.save({ refId: event.refId, type: event.type, payload: event, status: 'pending' });
}
async findPending(limit: number) {
return this.ds.getRepository(EventEntity)
.find({ where: { status: 'pending' }, take: limit });
}
}
MessageModule
Le poller. Toutes les 3s, lit les événements pending et les passe à l'EventManager.
Marque les événements processing puis done.
EventManager
Le dispatcher. Connaßt la liste des handlers enregistrés par type d'événement.
Pour chaque événement, appelle handler.handle(event).
// MessageModule â la boucle de polling
async poll() {
const events = await this.repo.findPending(50);
for (const event of events) {
await this.eventManager.dispatch(event);
await this.repo.markDone(event.refId);
}
}
// setInterval(poll, 3000) au démarrage
Un vrai handler du codebase : classe @injectable() avec une méthode handle(event).
@injectable()
export class UserRoleUpgradedHandler implements DomainEventHandler<UserRoleUpgraded> {
constructor(
@inject(TYPES.Mailer) private mailer: Mailer,
@inject(TYPES.UserRepository) private userRepo: UserRepository,
) {}
async handle(event: UserRoleUpgraded) {
const user = await this.userRepo.findById(event.userId);
await this.mailer.send(user.email, 'Votre rÎle a été mis à jour');
}
}
đĄ Le handler reçoit ses dĂ©pendances par injection (Inversify). Il ne connaĂźt pas l'agrĂ©gat qui a Ă©mis l'Ă©vĂ©nement.
Le handler doit ĂȘtre enregistrĂ© dans l'EventManager via configureHandler.
// Dans le Build du domaine
export class UserBuild implements Build {
build(container: Container) {
// Enregistrer le handler dans le container
container.bind(UserRoleUpgradedHandler).toSelf();
// L'enregistrer dans l'EventManager
const eventManager = container.get(EventManager);
eventManager.configureHandler(
UserRoleUpgraded, // le type d'événement
container.get(UserRoleUpgradedHandler) // l'instance
);
}
}
â ïž Si tu oublies configureHandler, l'Ă©vĂ©nement sera persistĂ© mais jamais traitĂ©. Silencieusement.
Persiste d'abord, traite ensuite â idempotence par refId.
1ïžâŁ PersistWebhook (use case)
Reçoit le webhook HTTP. Persiste le payload immédiatement avec un refId unique.
raise(WebhookPersisted) â rĂ©pond 200 OK tout de suite
2ïžâŁ WebhookPersisted (Ă©vĂ©nement)
Persisté en queue Postgres. Attend le polling.
3ïžâŁ WebhookHandler (handler)
Filtre par type de webhook. Traite seulement si pas déjà traité (refId).
Exécute la logique métier (créer user, mettre à jour commande...)
đĄ Pourquoi ? Si le handler plante, le webhook n'est pas perdu. Le refId garantit qu'un retraitement ne crĂ©e pas de doublon.
â Confondre @Handle et DomainEventHandler
@Handle = synchrone, dans l'agrégat. DomainEventHandler = asynchrone, entre composants.
â ïž Oublier d'enregistrer le handler
Sans configureHandler dans l'EventManager, l'événement est persisté mais jamais traité.
â Ne pas comprendre le polling
Dessiner le flux au tableau : agrĂ©gat â DB â polling (3s) â EventManager â handler.
â
ĂvĂ©nements synchrones = dans l'agrĂ©gat (@Handle). Asynchrones = entre composants (handlers).
â Le polling toutes les 3s lit la queue Postgres et dispatch aux handlers enregistrĂ©s.
â
Un handler est une classe @injectable() avec une méthode handle(event).
â
Le pattern webhook : persiste d'abord, traite ensuite â idempotence par refId.
Demain : le flux complet, de A Ă Z đ