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

ddd_toolkit_core/cqrs/
async_query_handler.rs
