Struct StreamRevision

Source

pub struct StreamRevision(/* private fields */);

Expand description

English:
Zero-based event position; the first event in a stream has revision 0.

日本語:
ゼロ始まりのイベント位置です。ストリームの最初のイベントはリビジョン 0 です。

Implementations§

Source §

impl StreamRevision

Source

pub const fn new(value: u64) -> StreamRevision

English:
Wraps value without checking whether a corresponding stream exists.

日本語:
対応するストリームの存在を確認せず value をラップします。

Examples found in repository ?

examples/event_sourcing.rs (line 96)

85    fn load(&self, id: &u64) -> Result<LoadResult<Added>, Infallible> {
86        match self.0.get(id) {
87            None => Ok(LoadResult::NoStream(NoStream)),
88            Some(events) => {
89                // Real adapters deserialize and select migrations at this boundary.
90                let current = events
91                    .iter()
92                    .cloned()
93                    .map(|event| ToV2.upcast(event).unwrap())
94                    .collect();
95                Ok(LoadResult::Found(
96                    StreamHistory::try_new(current, StreamRevision::new(events.len() as u64 - 1))
97                        .unwrap(),
98                ))
99            }
100        }
101    }
102    fn append(
103        &mut self,
104        id: &u64,
105        expected: ExpectedRevision,
106        events: Vec<Added>,
107    ) -> Result<AppendResult, AppendError<Infallible>> {
108        if events.is_empty() {
109            return Err(AppendError::EmptyBatch);
110        }
111        let current = self
112            .0
113            .get(id)
114            .map(|events| StreamRevision::new(events.len() as u64 - 1));
115        let matches = match (expected, current) {
116            (ExpectedRevision::NoStream(_), None) => true,
117            (ExpectedRevision::Revision(want), Some(have)) => want == have,
118            _ => false,
119        };
120        if !matches {
121            return Err(AppendError::Conflict(RevisionConflict {
122                expected,
123                current: current.map_or(
124                    CurrentRevision::NoStream(NoStream),
125                    CurrentRevision::Revision,
126                ),
127            }));
128        }
129        let revision = match current {
130            None => StreamRevision::new(events.len() as u64 - 1),
131            Some(revision) => StreamRevision::new(
132                revision
133                    .value()
134                    .checked_add(events.len() as u64)
135                    .ok_or(AppendError::RevisionExhausted { current: revision })?,
136            ),
137        };
138        // Exclusive access makes the check + append atomic only in this process.
139        self.0.entry(*id).or_default().extend(events);
140        Ok(AppendResult {
141            resulting_revision: revision,
142        })
143    }
144}
145#[derive(Default)]
146struct MemorySnapshots(HashMap<u64, AggregateSnapshot<u32>>);
147impl SnapshotStore for MemorySnapshots {
148    type Aggregate = Counter;
149    type State = u32;
150    type Error = Infallible;
151    fn load(
152        &self,
153        id: &u64,
154    ) -> Result<Option<AggregateSnapshot<u32>>, SnapshotLoadError<Infallible>> {
155        Ok(self
156            .0
157            .get(id)
158            .map(|snapshot| AggregateSnapshot::new(*snapshot.state(), snapshot.revision())))
159    }
160    fn save(
161        &mut self,
162        id: &u64,
163        snapshot: AggregateSnapshot<u32>,
164    ) -> Result<(), SnapshotSaveError<Infallible>> {
165        self.0.insert(*id, snapshot);
166        Ok(())
167    }
168}
169fn history(store: &MemoryEvents) -> StreamHistory<Added> {
170    let LoadResult::Found(history) = store.load(&1).unwrap() else {
171        panic!("stream missing");
172    };
173    history
174}
175fn main() {
176    let mut store = MemoryEvents::default();
177    assert!(matches!(store.load(&1).unwrap(), LoadResult::NoStream(_)));
178    let first = store
179        .append(&1, ExpectedRevision::NoStream(NoStream), vec![Added::V1(2)])
180        .unwrap();
181    assert_eq!(first.resulting_revision, StreamRevision::new(0));
182    let mut replayed = replay_aggregate::<Counter>(&1, &history(&store)).unwrap();
183    assert_eq!(replayed.aggregate.value, 2);
184    assert!(replayed.aggregate.pull_events().is_empty());
185    let mut snapshots = MemorySnapshots::default();
186    snapshots
187        .save(
188            &1,
189            AggregateSnapshot::new(replayed.aggregate.value, replayed.revision),
190        )
191        .unwrap();
192    store
193        .append(
194            &1,
195            ExpectedRevision::Revision(replayed.revision),
196            vec![Added::V2 { amount: 3 }],
197        )
198        .unwrap();
199    assert!(matches!(
200        store.append(
201            &1,
202            ExpectedRevision::Revision(replayed.revision),
203            vec![Added::V1(99)]
204        ),
205        Err(AppendError::Conflict(_))
206    ));
207    let history = history(&store);
208    let mut restored =
209        replay_with_optional_snapshot::<Counter>(&1, &history, snapshots.load(&1).unwrap())
210            .unwrap();
211    let full = replay_aggregate::<Counter>(&1, &history).unwrap();
212    assert_eq!(restored.aggregate.value, 5);
213    assert_eq!(restored.aggregate.value, full.aggregate.value);
214    assert_eq!(restored.revision, StreamRevision::new(1));
215    assert!(restored.aggregate.pull_events().is_empty());
216    println!("event_sourcing: upcast v1; rejected stale write; snapshot + tail = 5");
217}

More examples

Hide additional examples

examples/reliable_delivery.rs (line 106)

65    fn save_and_enqueue<'a>(
66        &'a mut self,
67        account: &'a Account,
68        expected: ExpectedRevision,
69        events: Vec<Credited>,
70    ) -> Pin<Box<dyn Future<Output = Result<AppendResult, AppendError<Self::Error>>> + Send + 'a>>
71    {
72        Box::pin(async move {
73            if events.is_empty() {
74                return Err(AppendError::EmptyBatch);
75            }
76            if self
77                .saved
78                .as_ref()
79                .is_some_and(|saved| saved.id != account.id)
80            {
81                return Err(AppendError::Adapter("demo stores only one account"));
82            }
83            let matches = match (expected, self.revision) {
84                (ExpectedRevision::NoStream(_), None) => true,
85                (ExpectedRevision::Revision(want), Some(have)) => want == have,
86                _ => false,
87            };
88            if !matches {
89                return Err(AppendError::Conflict(RevisionConflict {
90                    expected,
91                    current: self.revision.map_or(
92                        CurrentRevision::NoStream(NoStream),
93                        CurrentRevision::Revision,
94                    ),
95                }));
96            }
97            let first = match self.revision {
98                None => 0,
99                Some(current) => current
100                    .value()
101                    .checked_add(1)
102                    .ok_or(AppendError::RevisionExhausted { current })?,
103            };
104            let last = first.checked_add(events.len() as u64 - 1).ok_or(
105                AppendError::RevisionExhausted {
106                    current: StreamRevision::new(first),
107                },
108            )?;
109            let mut snapshot = account.clone();
110            snapshot.pull_events();
111            self.saved = Some(snapshot);
112            self.revision = Some(StreamRevision::new(last));
113            for (offset, event) in events.into_iter().enumerate() {
114                self.queue.push_back((
115                    OutboxMessageId::new(format!("{}:{}", account.id, first + offset as u64)),
116                    event,
117                ));
118            }
119            Ok(AppendResult {
120                resulting_revision: StreamRevision::new(last),
121            })
122        })
123    }

Source

pub const fn value(self) -> u64

English:
Returns the zero-based event position.

日本語:
ゼロ始まりのイベント位置を返します。

Examples found in repository ?

examples/event_sourcing.rs (line 133)

102    fn append(
103        &mut self,
104        id: &u64,
105        expected: ExpectedRevision,
106        events: Vec<Added>,
107    ) -> Result<AppendResult, AppendError<Infallible>> {
108        if events.is_empty() {
109            return Err(AppendError::EmptyBatch);
110        }
111        let current = self
112            .0
113            .get(id)
114            .map(|events| StreamRevision::new(events.len() as u64 - 1));
115        let matches = match (expected, current) {
116            (ExpectedRevision::NoStream(_), None) => true,
117            (ExpectedRevision::Revision(want), Some(have)) => want == have,
118            _ => false,
119        };
120        if !matches {
121            return Err(AppendError::Conflict(RevisionConflict {
122                expected,
123                current: current.map_or(
124                    CurrentRevision::NoStream(NoStream),
125                    CurrentRevision::Revision,
126                ),
127            }));
128        }
129        let revision = match current {
130            None => StreamRevision::new(events.len() as u64 - 1),
131            Some(revision) => StreamRevision::new(
132                revision
133                    .value()
134                    .checked_add(events.len() as u64)
135                    .ok_or(AppendError::RevisionExhausted { current: revision })?,
136            ),
137        };
138        // Exclusive access makes the check + append atomic only in this process.
139        self.0.entry(*id).or_default().extend(events);
140        Ok(AppendResult {
141            resulting_revision: revision,
142        })
143    }

More examples

Hide additional examples

examples/reliable_delivery.rs (line 100)

65    fn save_and_enqueue<'a>(
66        &'a mut self,
67        account: &'a Account,
68        expected: ExpectedRevision,
69        events: Vec<Credited>,
70    ) -> Pin<Box<dyn Future<Output = Result<AppendResult, AppendError<Self::Error>>> + Send + 'a>>
71    {
72        Box::pin(async move {
73            if events.is_empty() {
74                return Err(AppendError::EmptyBatch);
75            }
76            if self
77                .saved
78                .as_ref()
79                .is_some_and(|saved| saved.id != account.id)
80            {
81                return Err(AppendError::Adapter("demo stores only one account"));
82            }
83            let matches = match (expected, self.revision) {
84                (ExpectedRevision::NoStream(_), None) => true,
85                (ExpectedRevision::Revision(want), Some(have)) => want == have,
86                _ => false,
87            };
88            if !matches {
89                return Err(AppendError::Conflict(RevisionConflict {
90                    expected,
91                    current: self.revision.map_or(
92                        CurrentRevision::NoStream(NoStream),
93                        CurrentRevision::Revision,
94                    ),
95                }));
96            }
97            let first = match self.revision {
98                None => 0,
99                Some(current) => current
100                    .value()
101                    .checked_add(1)
102                    .ok_or(AppendError::RevisionExhausted { current })?,
103            };
104            let last = first.checked_add(events.len() as u64 - 1).ok_or(
105                AppendError::RevisionExhausted {
106                    current: StreamRevision::new(first),
107                },
108            )?;
109            let mut snapshot = account.clone();
110            snapshot.pull_events();
111            self.saved = Some(snapshot);
112            self.revision = Some(StreamRevision::new(last));
113            for (offset, event) in events.into_iter().enumerate() {
114                self.queue.push_back((
115                    OutboxMessageId::new(format!("{}:{}", account.id, first + offset as u64)),
116                    event,
117                ));
118            }
119            Ok(AppendResult {
120                resulting_revision: StreamRevision::new(last),
121            })
122        })
123    }