Function replay_aggregate 

pub fn replay_aggregate<A>(
    id: &<A as Entity>::Id,
    history: &StreamHistory<<A as AggregateRoot>::Event>,
) -> Result<RehydratedAggregate<A>, ReplayError<<A as AggregateReplay>::InitializationError, <A as AggregateReplay>::ApplicationError>>
where
    A: AggregateReplay,

English:
Initializes A for id and applies every event in history in stored order. Returns the reconstructed aggregate and history revision. The first hook error is wrapped in ReplayError; the local aggregate is dropped on error. No storage or publishing is performed, and hook panics propagate. The caller must supply history belonging to id.

This counter restores two historical increments without queuing new events.

日本語:
id の A を初期化し、history の全イベントを保存順に適用します。 復元した集約と履歴リビジョンを返します。最初のフックエラーを ReplayError で包み、 エラー時はローカルな集約を破棄します。保存や発行は行わず、フックのパニックは伝播します。 呼び出し側は id に属する履歴を指定する必要があります。

このカウンターは新規イベントを保留せず、履歴の加算 2 件を復元します。

Example / 使用例

use std::convert::Infallible;
use ddd_toolkit_core::{
    AggregateReplay, AggregateRoot, AggregateSnapshot, DomainEvent, Entity,
    SnapshotUse, StreamHistory, StreamRevision, replay_aggregate,
};
struct Increment(u64);
impl DomainEvent for Increment {}
struct Counter { id: u64, total: u64, pending: Vec<Increment> }
impl Entity for Counter {
    type Id = u64;
    fn id(&self) -> &u64 { &self.id }
}
impl AggregateRoot for Counter {
    type Event = Increment;
    fn record_event(&mut self, event: Increment) { self.pending.push(event); }
    fn pull_events(&mut self) -> Vec<Increment> { std::mem::take(&mut self.pending) }
}
impl AggregateReplay for Counter {
    type InitializationError = Infallible;
    type ApplicationError = &'static str;
    type Snapshot = u64;
    type SnapshotValidationError = Infallible;
    type SnapshotStateError = Infallible;
    fn initial_state(id: &u64) -> Result<Self, Infallible> {
        Ok(Self { id: *id, total: 0, pending: vec![] })
    }
    fn apply_event(&mut self, event: &Increment) -> Result<(), &'static str> {
        self.total = self.total.checked_add(event.0).ok_or("counter overflow")?;
        Ok(())
    }
    fn classify_snapshot(_: &AggregateSnapshot<u64>, _: StreamRevision)
        -> Result<SnapshotUse, Infallible> { Ok(SnapshotUse::Applicable) }
    fn from_snapshot(id: &u64, snapshot: AggregateSnapshot<u64>) -> Result<Self, Infallible> {
        Ok(Self { id: *id, total: snapshot.into_parts().0, pending: vec![] })
    }
}
let history = StreamHistory::try_new(vec![Increment(2), Increment(3)], StreamRevision::new(1))
    .unwrap();
let mut restored = replay_aggregate::<Counter>(&7, &history).unwrap();
assert_eq!(restored.aggregate.total, 5);
assert_eq!(restored.revision, StreamRevision::new(1));
assert!(restored.aggregate.pull_events().is_empty());

Examples found in repository

examples/event_sourcing.rs (line 182)

fn main() {
    let mut store = MemoryEvents::default();
    assert!(matches!(store.load(&1).unwrap(), LoadResult::NoStream(_)));
    let first = store
        .append(&1, ExpectedRevision::NoStream(NoStream), vec![Added::V1(2)])
        .unwrap();
    assert_eq!(first.resulting_revision, StreamRevision::new(0));
    let mut replayed = replay_aggregate::<Counter>(&1, &history(&store)).unwrap();
    assert_eq!(replayed.aggregate.value, 2);
    assert!(replayed.aggregate.pull_events().is_empty());
    let mut snapshots = MemorySnapshots::default();
    snapshots
        .save(
            &1,
            AggregateSnapshot::new(replayed.aggregate.value, replayed.revision),
        )
        .unwrap();
    store
        .append(
            &1,
            ExpectedRevision::Revision(replayed.revision),
            vec![Added::V2 { amount: 3 }],
        )
        .unwrap();
    assert!(matches!(
        store.append(
            &1,
            ExpectedRevision::Revision(replayed.revision),
            vec![Added::V1(99)]
        ),
        Err(AppendError::Conflict(_))
    ));
    let history = history(&store);
    let mut restored =
        replay_with_optional_snapshot::<Counter>(&1, &history, snapshots.load(&1).unwrap())
            .unwrap();
    let full = replay_aggregate::<Counter>(&1, &history).unwrap();
    assert_eq!(restored.aggregate.value, 5);
    assert_eq!(restored.aggregate.value, full.aggregate.value);
    assert_eq!(restored.revision, StreamRevision::new(1));
    assert!(restored.aggregate.pull_events().is_empty());
    println!("event_sourcing: upcast v1; rejected stale write; snapshot + tail = 5");
}