ddd_toolkit_core/
async_dispatch.rs

1use crate::{AggregateRoot, AsyncDomainEventPublisher};
2use std::future::Future;
3use std::pin::Pin;
4
5struct EventRestorationGuard<'a, A: AggregateRoot> {
6    aggregate: &'a mut A,
7    snapshot: Option<Vec<A::Event>>,
8}
9
10impl<'a, A: AggregateRoot> EventRestorationGuard<'a, A> {
11    fn new(aggregate: &'a mut A, snapshot: Vec<A::Event>) -> Self {
12        Self {
13            aggregate,
14            snapshot: Some(snapshot),
15        }
16    }
17
18    fn events(&self) -> impl Iterator<Item = &A::Event> {
19        self.snapshot.as_deref().into_iter().flatten()
20    }
21
22    fn commit(&mut self) {
23        self.snapshot.take();
24    }
25}
26
27impl<A: AggregateRoot> Drop for EventRestorationGuard<'_, A> {
28    fn drop(&mut self) {
29        if let Some(snapshot) = self.snapshot.take() {
30            for event in snapshot {
31                self.aggregate.record_event(event);
32            }
33        }
34    }
35}
36
37/// **English:**<br>
38/// Asynchronously publishes the events currently held by `aggregate`, in the order
39/// returned by [`AggregateRoot::pull_events`]. On success, the events are removed. If publishing
40/// fails, all pulled events are recorded back into the aggregate before the publisher error is
41/// returned; earlier successful publishes are not undone, so a retry can publish duplicates.
42///
43/// **日本語:**<br>
44/// `aggregate` が保持するイベントを [`AggregateRoot::pull_events`] の返す順序で非同期に
45/// 発行します。成功時はイベントを取り除きます。発行に失敗した場合、パブリッシャーのエラーを
46/// 返す前に取り出した全イベントを集約へ記録し直します。それ以前に成功した発行は取り消されないため、
47/// 再試行すると重複発行される可能性があります。
48pub fn dispatch_events_async<'a, A, P>(
49    aggregate: &'a mut A,
50    publisher: &'a mut P,
51) -> Pin<Box<dyn Future<Output = Result<(), P::Error>> + Send + 'a>>
52where
53    A: AggregateRoot + Send + 'a,
54    A::Event: Send + Sync + 'a,
55    P: AsyncDomainEventPublisher<Event = A::Event> + Send + 'a,
56    P::Error: 'a,
57{
58    Box::pin(async move {
59        let snapshot = aggregate.pull_events();
60        let mut guard = EventRestorationGuard::new(aggregate, snapshot);
61
62        for event in guard.events() {
63            publisher.publish(event).await?
64        }
65
66        guard.commit();
67        Ok(())
68    })
69}
ddd_toolkit_core/async_dispatch.rs

ddd_toolkit_core/
async_dispatch.rs
