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:** Asynchronously publishes the events currently held by `aggregate`, in the order
38/// returned by [`AggregateRoot::pull_events`]. On success, the events are removed. If publishing
39/// fails, all pulled events are recorded back into the aggregate before the publisher error is
40/// returned; earlier successful publishes are not undone, so a retry can publish duplicates.
41///
42/// **日本語:** `aggregate` が保持するイベントを [`AggregateRoot::pull_events`] の返す順序で非同期に
43/// 発行します。成功時はイベントを取り除きます。発行に失敗した場合、パブリッシャーのエラーを
44/// 返す前に取り出した全イベントを集約へ記録し直します。それ以前に成功した発行は取り消されないため、
45/// 再試行すると重複発行される可能性があります。
46pub fn dispatch_events_async<'a, A, P>(
47    aggregate: &'a mut A,
48    publisher: &'a mut P,
49) -> Pin<Box<dyn Future<Output = Result<(), P::Error>> + Send + 'a>>
50where
51    A: AggregateRoot + Send + 'a,
52    A::Event: Send + Sync + 'a,
53    P: AsyncDomainEventPublisher<Event = A::Event> + Send + 'a,
54    P::Error: 'a,
55{
56    Box::pin(async move {
57        let snapshot = aggregate.pull_events();
58        let mut guard = EventRestorationGuard::new(aggregate, snapshot);
59
60        for event in guard.events() {
61            publisher.publish(event).await?
62        }
63
64        guard.commit();
65        Ok(())
66    })
67}
ddd_toolkit_core/async_dispatch.rs

ddd_toolkit_core/
async_dispatch.rs
