ddd_toolkit_core/
async_dispatch.rs

1use crate::event_restoration_guard::EventRestorationGuard;
2use crate::{AggregateRoot, AsyncDomainEventPublisher};
3use std::future::Future;
4use std::pin::Pin;
5
6/// **English:**<br>
7/// Asynchronously publishes the events currently held by `aggregate`, in the order
8/// returned by [`AggregateRoot::pull_events`]. On success, the events are removed. If publishing
9/// fails, all pulled events are recorded back into the aggregate before the publisher error is
10/// returned.
11///
12/// If [`AsyncDomainEventPublisher::publish`] or its returned future panics and unwinds, a Drop
13/// guard records all pulled events back into the aggregate before unwinding continues. The guard
14/// also restores all pulled events when the dispatch future is dropped after events have been
15/// pulled but before dispatch completes. Dropping the dispatch future without polling it does not
16/// pull any events.
17///
18/// Restoration does not undo external side effects of publishing, including publishes that already
19/// succeeded; retrying can publish duplicates. Restoration may fail if destructors do not run
20/// (for example, with `panic=abort`) or if [`AggregateRoot::record_event`] itself panics during
21/// restoration. The aggregate's [`AggregateRoot::pull_events`] or [`AggregateRoot::record_event`]
22/// implementation may also panic.
23///
24/// **日本語:**<br>
25/// `aggregate` が保持するイベントを [`AggregateRoot::pull_events`] の返す順序で非同期に
26/// 発行します。成功時はイベントを取り除きます。発行に失敗した場合、パブリッシャーのエラーを
27/// 返す前に取り出した全イベントを集約へ記録し直します。
28///
29/// [`AsyncDomainEventPublisher::publish`] またはそれが返すfutureのパニックがアンワインド
30/// する場合、Dropガードが取り出した全イベントを集約へ記録し直してから、アンワインドが
31/// 継続します。イベント取り出し後、ディスパッチ完了前にディスパッチのfutureがdropされた
32/// 場合も、ガードが取り出した全イベントを復元します。ディスパッチのfutureが一度もpoll
33/// されなければ、イベントは取り出されません。
34///
35/// 復元は、すでに成功した発行を含め、発行による外部副作用を取り消しません。再試行すると
36/// 重複発行される可能性があります。`panic=abort` などでデストラクタが実行されない場合や、
37/// 復元中に [`AggregateRoot::record_event`] 自体がパニックする場合、復元は保証されません。
38/// 集約の [`AggregateRoot::pull_events`] または [`AggregateRoot::record_event`] の実装も
39/// パニックする可能性があります。
40pub fn dispatch_events_async<'a, A, P>(
41    aggregate: &'a mut A,
42    publisher: &'a mut P,
43) -> Pin<Box<dyn Future<Output = Result<(), P::Error>> + Send + 'a>>
44where
45    A: AggregateRoot + Send + 'a,
46    A::Event: Send + Sync + 'a,
47    P: AsyncDomainEventPublisher<Event = A::Event> + Send + 'a,
48    P::Error: 'a,
49{
50    Box::pin(async move {
51        let snapshot = aggregate.pull_events();
52        let mut guard = EventRestorationGuard::new(aggregate, snapshot);
53
54        for event in guard.events() {
55            publisher.publish(event).await?
56        }
57
58        guard.commit();
59        Ok(())
60    })
61}
ddd_toolkit_core/async_dispatch.rs

ddd_toolkit_core/
async_dispatch.rs
