ddd_toolkit_core/
dispatch.rs

1use crate::{AggregateRoot, DomainEventPublisher};
2
3struct EventRestorationGuard<'a, A: AggregateRoot> {
4    aggregate: &'a mut A,
5    snapshot: Option<Vec<A::Event>>,
6}
7
8impl<'a, A: AggregateRoot> EventRestorationGuard<'a, A> {
9    fn new(aggregate: &'a mut A, snapshot: Vec<A::Event>) -> Self {
10        Self {
11            aggregate,
12            snapshot: Some(snapshot),
13        }
14    }
15
16    fn events(&self) -> impl Iterator<Item = &A::Event> {
17        self.snapshot.as_deref().into_iter().flatten()
18    }
19
20    fn commit(&mut self) {
21        self.snapshot.take();
22    }
23}
24
25impl<A: AggregateRoot> Drop for EventRestorationGuard<'_, A> {
26    fn drop(&mut self) {
27        if let Some(snapshot) = self.snapshot.take() {
28            for event in snapshot {
29                self.aggregate.record_event(event);
30            }
31        }
32    }
33}
34
35/// **English:**<br>
36/// Publishes the events currently held by `aggregate` through `publisher`.
37///
38/// Events are passed to [`DomainEventPublisher::publish`] in the order returned by
39/// [`AggregateRoot::pull_events`]. If a publish operation returns an error, all events pulled at
40/// the start of this call are recorded back into the aggregate before that error is returned.
41/// This restoration does not undo publishes that already succeeded. If every publish succeeds,
42/// the events are removed from the aggregate and the function returns `Ok(())`.
43///
44/// The function may panic if the aggregate's [`AggregateRoot::pull_events`] or
45/// [`AggregateRoot::record_event`] implementation panics. It does not impose additional
46/// synchronization; the aggregate and publisher must be mutably borrowed by the caller for the
47/// duration of the dispatch.
48///
49/// **日本語:**<br>
50/// `aggregate` が現在保持しているイベントを `publisher` 経由で発行します。
51///
52/// イベントは [`AggregateRoot::pull_events`] が返した順序で [`DomainEventPublisher::publish`]
53/// に渡されます。発行処理がエラーを返すと、この呼び出しの開始時に取り出したすべての
54/// イベントを集約へ記録し直してから、そのエラーを返します。すでに成功した発行処理は
55/// 取り消されません。すべての発行に成功した場合、イベントは集約から削除され、`Ok(())`
56/// を返します。
57///
58/// 集約の [`AggregateRoot::pull_events`] または [`AggregateRoot::record_event`] の実装が
59/// パニックすると、この関数もパニックする可能性があります。この関数は追加の同期を
60/// 行わないため、呼び出し側はディスパッチ中、集約とパブリッシャーを可変借用します。
61pub fn dispatch_events<A, P>(aggregate: &mut A, publisher: &mut P) -> Result<(), P::Error>
62where
63    A: AggregateRoot,
64    P: DomainEventPublisher<Event = A::Event>,
65{
66    let snapshot = aggregate.pull_events();
67    let mut guard = EventRestorationGuard::new(aggregate, snapshot);
68
69    for event in guard.events() {
70        publisher.publish(event)?;
71    }
72
73    guard.commit();
74    Ok(())
75}
ddd_toolkit_core/dispatch.rs

ddd_toolkit_core/
dispatch.rs
