Function dispatch_events

Source

pub fn dispatch_events<A, P>(
    aggregate: &mut A,
    publisher: &mut P,
) -> Result<(), <P as DomainEventPublisher>::Error>where
    A: AggregateRoot,
    P: DomainEventPublisher<Event = <A as AggregateRoot>::Event>,

Expand description

English:
Publishes the events currently held by aggregate through publisher.

Events are passed to DomainEventPublisher::publish in the order returned by AggregateRoot::pull_events. If a publish operation returns an error, all events pulled at the start of this call are recorded back into the aggregate before that error is returned. If every publish succeeds, the events are removed from the aggregate and the function returns Ok(()).

If DomainEventPublisher::publish panics and unwinds, a Drop guard records all pulled events back into the aggregate before unwinding continues. Restoration does not undo external side effects of publishing, including publishes that already succeeded; retrying can publish duplicates. Restoration may fail if destructors do not run (for example, with panic=abort) or if AggregateRoot::record_event itself panics during restoration.

The function may panic if the aggregate’s AggregateRoot::pull_events or AggregateRoot::record_event implementation panics. It does not impose additional synchronization; the aggregate and publisher must be mutably borrowed by the caller for the duration of the dispatch.

日本語:
aggregate が現在保持しているイベントを publisher 経由で発行します。

イベントは AggregateRoot::pull_events が返した順序で DomainEventPublisher::publish に渡されます。発行処理がエラーを返すと、この呼び出しの開始時に取り出したすべてのイベントを集約へ記録し直してから、そのエラーを返します。すべての発行に成功した場合、イベントは集約から削除され、Ok(()) を返します。

DomainEventPublisher::publish のパニックがアンワインドする場合、Dropガードが取り出した全イベントを集約へ記録し直してから、アンワインドが継続します。復元は、すでに成功した発行を含め、発行による外部副作用を取り消しません。再試行すると重複発行される可能性があります。panic=abort などでデストラクタが実行されない場合や、復元中に AggregateRoot::record_event 自体がパニックする場合、復元は保証されません。

集約の AggregateRoot::pull_events または AggregateRoot::record_event の実装がパニックすると、この関数もパニックする可能性があります。この関数は追加の同期を行わないため、呼び出し側はディスパッチ中、集約とパブリッシャーを可変借用します。

Examples found in repository ?

examples/dispatch_retry.rs (line 35)

14fn main() {
15    let mut order = Order {
16        id: 1,
17        total: 4200,
18        events: vec![],
19    };
20    order.record_event(Placed(1));
21    order.record_event(Placed(2));
22    let mut store = Store::default();
23    store.save(&order).unwrap();
24    let mut loaded = store.find_by_id(&1).unwrap().unwrap();
25    assert_eq!(loaded.total, 4200);
26    assert!(loaded.pull_events().is_empty());
27
28    // Saving succeeded. A later publisher error cannot undo that save.
29    // 保存と配信は別操作。後の配信失敗で保存がロールバックされるわけではない。
30    let mut publisher = Publisher {
31        fail_at: Some(1),
32        ..Publisher::default()
33    };
34    assert_eq!(
35        dispatch_events(&mut order, &mut publisher),
36        Err("broker down")
37    );
38    assert!(store.exists(&1).unwrap()); // Publication failure does not roll back saving.
39    assert_eq!(order.events, vec![Placed(1), Placed(2)]);
40    // The whole batch was restored, including the event already delivered.
41    // 配信済みイベントも含めたバッチ全体が復元される。受信側には重複対策が必要。
42    publisher.fail_at = None;
43    dispatch_events(&mut order, &mut publisher).unwrap();
44    assert!(order.pull_events().is_empty());
45    assert_eq!(publisher.accepted, vec![Placed(1), Placed(1), Placed(2)]);
46    println!("dispatch_retry: saved total=4200; retry delivered [1, 1, 2]");
47}

More examples

Hide additional examples

examples/getting_started.rs (line 107)

84fn main() {
85    let mut order = Order::draft(1);
86    assert_eq!(order.confirm(), Err(OrderError::EmptyOrder));
87    assert_eq!(order.add_item(0), Err(OrderError::ZeroPrice));
88    order.add_item(1200).unwrap();
89    order.add_item(3000).unwrap();
90    order.confirm().unwrap();
91    assert_eq!(order.add_item(100), Err(OrderError::AlreadyConfirmed));
92    assert_eq!(order.confirm(), Err(OrderError::AlreadyConfirmed));
93    assert_eq!(order.total_yen, 4200);
94    assert_eq!(order.events.len(), 1); // Rejected operations did not emit more events.
95
96    // Persistence and notification are separate, explicitly ordered operations.
97    // 保存と通知は別操作。順番はアプリケーションが明示する。
98    let mut orders = MemoryOrders::default();
99    orders.save(&order).unwrap();
100    let saved = orders.find_by_id(&1).unwrap().unwrap();
101    assert!(saved.confirmed);
102    assert_eq!(saved.total_yen, 4200);
103    assert!(saved.events.is_empty()); // Persisted order state excludes pending notifications.
104    assert_eq!(order.events.len(), 1); // Saving did not consume the original queue.
105
106    let mut notifications = Notifications::default();
107    dispatch_events(&mut order, &mut notifications).unwrap();
108    assert_eq!(
109        notifications.sent,
110        vec![OrderConfirmed {
111            order_id: 1,
112            total_yen: 4200
113        }]
114    );
115    assert!(order.events.is_empty());
116    println!("getting_started: order #1 confirmed; saved total=4200 yen; notifications=1");
117}

dispatch_events

Function dispatch_events Copy item path

Examples found in repository?

Function dispatch_events

Examples found in repository ?