ddd_toolkit_core/cqrs/
async_query_handler.rs

1#![allow(clippy::type_complexity)]
2
3use crate::cqrs::OperationState;
4use crate::{
5    AggregateRoot, AsyncDomainEventPublisher, AsyncRepository, Entity, LifecycleCompletionError,
6    LifecycleOperationError, dispatch_events_async,
7};
8use std::future::Future;
9use std::pin::Pin;
10
11struct AttemptGuard<'a> {
12    state: &'a mut OperationState,
13    succeeded: bool,
14}
15
16impl<'a> AttemptGuard<'a> {
17    fn new(state: &'a mut OperationState) -> Self {
18        Self {
19            state,
20            succeeded: false,
21        }
22    }
23
24    fn succeed(&mut self) {
25        *self.state = OperationState::Succeeded;
26        self.succeeded = true;
27    }
28}
29
30impl Drop for AttemptGuard<'_> {
31    fn drop(&mut self) {
32        if !self.succeeded {
33            *self.state = OperationState::FailedConsumed;
34        }
35    }
36}
37
38/// **English:**<br>
39/// Coordinates asynchronous aggregate persistence and event dispatch. Each operation may be attempted once; dropping an in-progress future consumes that attempt.
40///
41/// **日本語:**<br>
42/// 非同期の集約永続化とイベント配信を調整します。各操作は一度だけ試行でき、実行中の future を破棄すると試行は消費済みになります。
43pub struct AsyncLifecycleContext<'storage, A, R, P>
44where
45    A: AggregateRoot + Send + 'storage,
46    R: AsyncRepository<Aggregate = A> + Send + 'storage,
47    P: AsyncDomainEventPublisher<Event = A::Event> + Send + 'storage,
48    A::Event: Send + Sync + 'storage,
49    R::Error: Send + 'storage,
50    P::Error: Send + 'storage,
51{
52    aggregate: &'storage mut A,
53    repository: &'storage mut R,
54    publisher: &'storage mut P,
55    save_state: OperationState,
56    dispatch_state: OperationState,
57}
58impl<'storage, A, R, P> AsyncLifecycleContext<'storage, A, R, P>
59where
60    A: AggregateRoot + Send + 'storage,
61    R: AsyncRepository<Aggregate = A> + Send + 'storage,
62    P: AsyncDomainEventPublisher<Event = A::Event> + Send + 'storage,
63    A::Event: Send + Sync + 'storage,
64    R::Error: Send + 'storage,
65    P::Error: Send + 'storage,
66{
67    /// **English:**<br>
68    /// Creates a lifecycle context borrowing the aggregate, repository, and publisher mutably.
69    ///
70    /// **日本語:**<br>
71    /// 集約、リポジトリ、パブリッシャーを可変借用するライフサイクルコンテキストを作成します。
72    pub fn new(
73        aggregate: &'storage mut A,
74        repository: &'storage mut R,
75        publisher: &'storage mut P,
76    ) -> Self {
77        Self {
78            aggregate,
79            repository,
80            publisher,
81            save_state: OperationState::Pending,
82            dispatch_state: OperationState::Pending,
83        }
84    }
85
86    /// **English:**<br>
87    /// Attempts repository persistence once; adapter errors are returned and the attempt cannot be retried.
88    ///
89    /// **日本語:**<br>
90    /// リポジトリへの保存を一度試行します。アダプターエラーを返し、再試行はできません。
91    pub fn save<'ctx>(
92        &'ctx mut self,
93    ) -> Pin<Box<dyn Future<Output = Result<(), LifecycleOperationError<R::Error>>> + Send + 'ctx>>
94    where
95        'storage: 'ctx,
96        A: Send + 'ctx,
97        R: Send + 'ctx,
98        R::Error: Send + 'ctx,
99    {
100        if self.save_state != OperationState::Pending {
101            return Box::pin(async { Err(LifecycleOperationError::AlreadySaved) });
102        }
103        self.save_state = OperationState::FailedConsumed;
104        let state = &mut self.save_state;
105        let repository: &'ctx mut R = &mut *self.repository;
106        let aggregate: &'ctx mut A = &mut *self.aggregate;
107        Box::pin(async move {
108            let mut guard = AttemptGuard::new(state);
109            match repository.save(&*aggregate).await {
110                Ok(()) => {
111                    guard.succeed();
112                    Ok(())
113                }
114                Err(error) => Err(LifecycleOperationError::Adapter(error)),
115            }
116        })
117    }
118
119    /// **English:**<br>
120    /// Attempts event dispatch once; adapter errors are returned and the attempt cannot be retried.
121    ///
122    /// **日本語:**<br>
123    /// イベント配信を一度試行します。アダプターエラーを返し、再試行はできません。
124    pub fn dispatch<'ctx>(
125        &'ctx mut self,
126    ) -> Pin<Box<dyn Future<Output = Result<(), LifecycleOperationError<P::Error>>> + Send + 'ctx>>
127    where
128        'storage: 'ctx,
129        A: Send + 'ctx,
130        P: Send + 'ctx,
131        A::Event: Send + Sync + 'ctx,
132        P::Error: Send + 'ctx,
133        P::Error: 'ctx,
134    {
135        if self.dispatch_state != OperationState::Pending {
136            return Box::pin(async { Err(LifecycleOperationError::AlreadyDispatched) });
137        }
138        self.dispatch_state = OperationState::FailedConsumed;
139        let state = &mut self.dispatch_state;
140        let aggregate: &'ctx mut A = &mut *self.aggregate;
141        let publisher: &'ctx mut P = &mut *self.publisher;
142        Box::pin(async move {
143            let mut guard = AttemptGuard::new(state);
144            match dispatch_events_async(aggregate, publisher).await {
145                Ok(()) => {
146                    guard.succeed();
147                    Ok(())
148                }
149                Err(error) => Err(LifecycleOperationError::Adapter(error)),
150            }
151        })
152    }
153
154    /// **English:**<br>
155    /// Consumes the context and succeeds only when both save and dispatch succeeded.
156    ///
157    /// **日本語:**<br>
158    /// コンテキストを消費し、保存と配信の両方が成功した場合にのみ成功します。
159    pub fn complete(self) -> Result<(), LifecycleCompletionError> {
160        match (self.save_state, self.dispatch_state) {
161            (OperationState::Succeeded, OperationState::Succeeded) => Ok(()),
162            (OperationState::FailedConsumed, _) => {
163                Err(LifecycleCompletionError::SaveFailedConsumed)
164            }
165            (_, OperationState::FailedConsumed) => {
166                Err(LifecycleCompletionError::DispatchFailedConsumed)
167            }
168            (OperationState::Pending, _) => Err(LifecycleCompletionError::MissingSave),
169            (_, OperationState::Pending) => Err(LifecycleCompletionError::MissingDispatch),
170        }
171    }
172}
173
174/// **English:**<br>
175/// Defines asynchronous work performed with a mutable asynchronous lifecycle context.
176///
177/// **日本語:**<br>
178/// 非同期ライフサイクルコンテキストを使って実行する非同期処理を定義します。
179pub trait AsyncLifecycleCallback<A, R, P>
180where
181    A: AggregateRoot + Send,
182    R: AsyncRepository<Aggregate = A> + Send,
183    P: AsyncDomainEventPublisher<Event = A::Event> + Send,
184    A::Event: Send + Sync,
185    R::Error: Send,
186    P::Error: Send,
187{
188    type Output;
189    type Error;
190
191    fn call<'storage, 'callback>(
192        self,
193        context: &'callback mut AsyncLifecycleContext<'storage, A, R, P>,
194    ) -> Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send + 'callback>>
195    where
196        'storage: 'callback,
197        A: 'storage,
198        R: 'storage,
199        P: 'storage,
200        A::Event: 'storage,
201        R::Error: 'storage,
202        P::Error: 'storage,
203        Self: Send + 'callback,
204        Self::Output: Send + 'callback,
205        Self::Error: Send + 'callback;
206}
207
208/// **English:**<br>
209/// Asynchronously handles a command against an aggregate using a lifecycle callback.
210///
211/// **日本語:**<br>
212/// ライフサイクルコールバックを使って集約に対するコマンドを非同期に処理します。
213pub trait AsyncCommandHandler {
214    type Command;
215    type Aggregate: AggregateRoot;
216    type Repository: AsyncRepository<Aggregate = Self::Aggregate>;
217    type Publisher: AsyncDomainEventPublisher<Event = <Self::Aggregate as AggregateRoot>::Event>;
218    type Output;
219    type Error;
220
221    fn handle<'a, F>(
222        &'a mut self,
223        id: &'a <Self::Aggregate as Entity>::Id,
224        command: Self::Command,
225        callback: F,
226    ) -> Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send + 'a>>
227    where
228        Self: Send + 'a,
229        Self::Command: Send + 'a,
230        Self::Aggregate: Send + 'a,
231        <Self::Aggregate as Entity>::Id: Sync + 'a,
232        <Self::Aggregate as AggregateRoot>::Event: Send + Sync + 'a,
233        Self::Repository: Send + 'a,
234        <Self::Repository as AsyncRepository>::Error: Send + 'a,
235        Self::Publisher: Send + 'a,
236        <Self::Publisher as AsyncDomainEventPublisher>::Error: Send + 'a,
237        Self::Output: Send + 'a,
238        Self::Error: From<LifecycleOperationError<<Self::Repository as AsyncRepository>::Error>>
239            + From<LifecycleOperationError<<Self::Publisher as AsyncDomainEventPublisher>::Error>>
240            + From<LifecycleCompletionError>
241            + Send
242            + 'a,
243        F: AsyncLifecycleCallback<
244                Self::Aggregate,
245                Self::Repository,
246                Self::Publisher,
247                Output = Self::Output,
248                Error = Self::Error,
249            > + Send
250            + 'a;
251}
252
253/// **English:**<br>
254/// Asynchronously creates an aggregate from a command and runs the lifecycle callback.
255///
256/// **日本語:**<br>
257/// コマンドから集約を非同期に作成し、ライフサイクルコールバックを実行します。
258pub trait AsyncAggregateCreator {
259    type Command;
260    type Aggregate: AggregateRoot;
261    type Repository: AsyncRepository<Aggregate = Self::Aggregate>;
262    type Publisher: AsyncDomainEventPublisher<Event = <Self::Aggregate as AggregateRoot>::Event>;
263    type Output;
264    type Error;
265
266    fn create<'a, F>(
267        &'a mut self,
268        command: Self::Command,
269        callback: F,
270    ) -> Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send + 'a>>
271    where
272        Self: Send + 'a,
273        Self::Command: Send + 'a,
274        Self::Aggregate: Send + 'a,
275        <Self::Aggregate as AggregateRoot>::Event: Send + Sync + 'a,
276        Self::Repository: Send + 'a,
277        <Self::Repository as AsyncRepository>::Error: Send + 'a,
278        Self::Publisher: Send + 'a,
279        <Self::Publisher as AsyncDomainEventPublisher>::Error: Send + 'a,
280        Self::Output: Send + 'a,
281        Self::Error: From<LifecycleOperationError<<Self::Repository as AsyncRepository>::Error>>
282            + From<LifecycleOperationError<<Self::Publisher as AsyncDomainEventPublisher>::Error>>
283            + From<LifecycleCompletionError>
284            + Send
285            + 'a,
286        F: AsyncLifecycleCallback<
287                Self::Aggregate,
288                Self::Repository,
289                Self::Publisher,
290                Output = Self::Output,
291                Error = Self::Error,
292            > + Send
293            + 'a;
294}
295
296/// **English:**<br>
297/// Handles a query asynchronously without requiring aggregate lifecycle operations.
298///
299/// **日本語:**<br>
300/// 集約ライフサイクル操作を伴わずにクエリを非同期処理します。
301pub trait AsyncQueryHandler {
302    type Query;
303    type Result;
304    type Error;
305
306    fn handle<'a>(
307        &'a self,
308        query: Self::Query,
309    ) -> Pin<Box<dyn Future<Output = Result<Self::Result, Self::Error>> + Send + 'a>>
310    where
311        Self: Sync + 'a,
312        Self::Query: Send + 'a,
313        Self::Result: Send + 'a,
314        Self::Error: Send + 'a;
315}
ddd_toolkit_core/cqrs/async_query_handler.rs

ddd_toolkit_core/cqrs/
async_query_handler.rs
