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; earlier successful publishes are not undone, so a retry can publish duplicates.
11///
12/// **日本語:**<br>
13/// `aggregate` が保持するイベントを [`AggregateRoot::pull_events`] の返す順序で非同期に
14/// 発行します。成功時はイベントを取り除きます。発行に失敗した場合、パブリッシャーのエラーを
15/// 返す前に取り出した全イベントを集約へ記録し直します。それ以前に成功した発行は取り消されないため、
16/// 再試行すると重複発行される可能性があります。
17pub fn dispatch_events_async<'a, A, P>(
18    aggregate: &'a mut A,
19    publisher: &'a mut P,
20) -> Pin<Box<dyn Future<Output = Result<(), P::Error>> + Send + 'a>>
21where
22    A: AggregateRoot + Send + 'a,
23    A::Event: Send + Sync + 'a,
24    P: AsyncDomainEventPublisher<Event = A::Event> + Send + 'a,
25    P::Error: 'a,
26{
27    Box::pin(async move {
28        let snapshot = aggregate.pull_events();
29        let mut guard = EventRestorationGuard::new(aggregate, snapshot);
30
31        for event in guard.events() {
32            publisher.publish(event).await?
33        }
34
35        guard.commit();
36        Ok(())
37    })
38}
ddd_toolkit_core/async_dispatch.rs

ddd_toolkit_core/
async_dispatch.rs
