mongodb/change_stream/
event.rs

1//! Contains the types related to a `ChangeStream` event.
2#[cfg(test)]
3use std::convert::TryInto;
4
5#[cfg(test)]
6use crate::{bson::Bson, bson_compat::RawError};
7use crate::{
8    bson::{DateTime, Document, RawBson, RawDocumentBuf, Timestamp},
9    cursor::common::CursorSpecification,
10    options::ChangeStreamOptions,
11};
12
13use serde::{Deserialize, Serialize};
14use serde_with::skip_serializing_none;
15
16/// An opaque token used for resuming an interrupted
17/// [`ChangeStream`](crate::change_stream::ChangeStream).
18///
19/// When starting a new change stream,
20/// [`crate::action::Watch::start_after`] and
21/// [`crate::action::Watch::resume_after`] fields can be specified
22/// with instances of `ResumeToken`.
23///
24/// See the documentation
25/// [here](https://www.mongodb.com/docs/manual/changeStreams/#change-stream-resume-token) for more
26/// information on resume tokens.
27#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
28pub struct ResumeToken(pub(crate) RawBson);
29
30impl ResumeToken {
31    pub(crate) fn initial(
32        options: Option<&ChangeStreamOptions>,
33        spec: &CursorSpecification,
34    ) -> Option<ResumeToken> {
35        match &spec.post_batch_resume_token {
36            // Token from initial response from `aggregate`
37            Some(token) if spec.is_empty => Some(token.clone()),
38            // Token from options passed to `watch`
39            _ => options
40                .and_then(|o| o.start_after.as_ref().or(o.resume_after.as_ref()))
41                .cloned(),
42        }
43    }
44
45    pub(crate) fn from_raw(doc: Option<RawDocumentBuf>) -> Option<ResumeToken> {
46        doc.map(|doc| ResumeToken(RawBson::Document(doc)))
47    }
48
49    #[cfg(test)]
50    pub(crate) fn parsed(self) -> std::result::Result<Bson, RawError> {
51        self.0.try_into()
52    }
53}
54
55/// A `ChangeStreamEvent` represents a
56/// [change event](https://www.mongodb.com/docs/manual/reference/change-events/) in the associated change stream.
57#[skip_serializing_none]
58#[derive(Debug, Serialize, Deserialize, PartialEq)]
59#[serde(rename_all = "camelCase")]
60#[non_exhaustive]
61pub struct ChangeStreamEvent<T> {
62    /// An opaque token for use when resuming an interrupted `ChangeStream`.
63    ///
64    /// See the documentation
65    /// [here](https://www.mongodb.com/docs/manual/changeStreams/#change-stream-resume-token) for
66    /// more information on resume tokens.
67    ///
68    /// Also see the documentation on [resuming a change
69    /// stream](https://www.mongodb.com/docs/manual/changeStreams/#resume-a-change-stream).
70    #[serde(rename = "_id")]
71    pub id: ResumeToken,
72
73    /// Describes the type of operation represented in this change notification.
74    pub operation_type: OperationType,
75
76    /// Identifies the collection or database on which the event occurred.
77    pub ns: Option<ChangeNamespace>,
78
79    /// The type of the newly created object.  Only included for `OperationType::Create`.
80    pub ns_type: Option<ChangeNamespaceType>,
81
82    /// The new name for the `ns` collection.  Only included for `OperationType::Rename`.
83    pub to: Option<ChangeNamespace>,
84
85    /// The identifier for the session associated with the transaction.
86    /// Only present if the operation is part of a multi-document transaction.
87    pub lsid: Option<Document>,
88
89    /// Together with the lsid, a number that helps uniquely identify a transaction.
90    /// Only present if the operation is part of a multi-document transaction.
91    pub txn_number: Option<i64>,
92
93    /// A `Document` that contains the `_id` of the document created or modified by the `insert`,
94    /// `replace`, `delete`, `update` operations (i.e. CRUD operations). For sharded collections,
95    /// also displays the full shard key for the document. The `_id` field is not repeated if it is
96    /// already a part of the shard key.
97    pub document_key: Option<Document>,
98
99    /// A description of the fields that were updated or removed by the update operation.
100    /// Only specified if `operation_type` is `OperationType::Update`.
101    pub update_description: Option<UpdateDescription>,
102
103    /// The cluster time at which the change occurred.
104    pub cluster_time: Option<Timestamp>,
105
106    /// The wall time from the mongod that the change event originated from.
107    pub wall_time: Option<DateTime>,
108
109    /// The `Document` created or modified by the `insert`, `replace`, `delete`, `update`
110    /// operations (i.e. CRUD operations).
111    ///
112    /// For `insert` and `replace` operations, this represents the new document created by the
113    /// operation.  For `delete` operations, this field is `None`.
114    ///
115    /// For `update` operations, this field only appears if you configured the change stream with
116    /// [`full_document`](crate::action::Watch::full_document) set to
117    /// [`UpdateLookup`](crate::options::FullDocumentType::UpdateLookup). This field then
118    /// represents the most current majority-committed version of the document modified by the
119    /// update operation.
120    pub full_document: Option<T>,
121
122    /// Contains the pre-image of the modified or deleted document if the pre-image is available
123    /// for the change event and either `Required` or `WhenAvailable` was specified for the
124    /// [`full_document_before_change`](
125    /// crate::action::Watch::full_document_before_change) option when creating the
126    /// change stream. If `WhenAvailable` was specified but the pre-image is unavailable, this
127    /// will be explicitly set to `None`.
128    pub full_document_before_change: Option<T>,
129}
130
131/// Describes which fields have been updated or removed from a document.
132#[skip_serializing_none]
133#[derive(Debug, Serialize, Deserialize, PartialEq)]
134#[serde(rename_all = "camelCase")]
135#[non_exhaustive]
136pub struct UpdateDescription {
137    /// A `Document` containing key:value pairs of names of the fields that were changed, and the
138    /// new value for those fields.
139    pub updated_fields: Document,
140
141    /// An array of field names that were removed from the `Document`.
142    pub removed_fields: Vec<String>,
143
144    /// Arrays that were truncated in the `Document`.
145    pub truncated_arrays: Option<Vec<TruncatedArray>>,
146
147    /// When an update event reports changes involving ambiguous fields, the disambiguatedPaths
148    /// document provides the path key with an array listing each path component.
149    /// Note: The disambiguatedPaths field is only available on change streams started with the
150    /// `show_expanded_events` option.
151    pub disambiguated_paths: Option<Document>,
152}
153
154/// Describes an array that has been truncated.
155#[derive(Debug, Serialize, Deserialize, PartialEq)]
156#[serde(rename_all = "camelCase")]
157#[non_exhaustive]
158pub struct TruncatedArray {
159    /// The field path of the array.
160    pub field: String,
161
162    /// The new size of the array.
163    pub new_size: i32,
164}
165
166/// The operation type represented in a given change notification.
167#[derive(Debug, Clone, PartialEq, Eq)]
168#[non_exhaustive]
169pub enum OperationType {
170    /// See [insert-event](https://www.mongodb.com/docs/manual/reference/change-events/#insert-event)
171    Insert,
172
173    /// See [update-event](https://www.mongodb.com/docs/manual/reference/change-events/#update-event)
174    Update,
175
176    /// See [replace-event](https://www.mongodb.com/docs/manual/reference/change-events/#replace-event)
177    Replace,
178
179    /// See [delete-event](https://www.mongodb.com/docs/manual/reference/change-events/#delete-event)
180    Delete,
181
182    /// See [drop-event](https://www.mongodb.com/docs/manual/reference/change-events/#drop-event)
183    Drop,
184
185    /// See [rename-event](https://www.mongodb.com/docs/manual/reference/change-events/#rename-event)
186    Rename,
187
188    /// See [dropdatabase-event](https://www.mongodb.com/docs/manual/reference/change-events/#dropdatabase-event)
189    DropDatabase,
190
191    /// See [invalidate-event](https://www.mongodb.com/docs/manual/reference/change-events/#invalidate-event)
192    Invalidate,
193
194    /// A catch-all for future event types.
195    Other(String),
196}
197
198#[derive(Serialize, Deserialize)]
199#[serde(rename_all = "camelCase")]
200enum OperationTypeHelper {
201    Insert,
202    Update,
203    Replace,
204    Delete,
205    Drop,
206    Rename,
207    DropDatabase,
208    Invalidate,
209}
210
211#[derive(Serialize, Deserialize)]
212#[serde(untagged)]
213enum OperationTypeWrapper<'a> {
214    Known(OperationTypeHelper),
215    Unknown(&'a str),
216}
217
218impl<'a> From<&'a OperationType> for OperationTypeWrapper<'a> {
219    fn from(src: &'a OperationType) -> Self {
220        match src {
221            OperationType::Insert => Self::Known(OperationTypeHelper::Insert),
222            OperationType::Update => Self::Known(OperationTypeHelper::Update),
223            OperationType::Replace => Self::Known(OperationTypeHelper::Replace),
224            OperationType::Delete => Self::Known(OperationTypeHelper::Delete),
225            OperationType::Drop => Self::Known(OperationTypeHelper::Drop),
226            OperationType::Rename => Self::Known(OperationTypeHelper::Rename),
227            OperationType::DropDatabase => Self::Known(OperationTypeHelper::DropDatabase),
228            OperationType::Invalidate => Self::Known(OperationTypeHelper::Invalidate),
229            OperationType::Other(s) => Self::Unknown(s),
230        }
231    }
232}
233
234impl From<OperationTypeWrapper<'_>> for OperationType {
235    fn from(src: OperationTypeWrapper) -> Self {
236        match src {
237            OperationTypeWrapper::Known(h) => match h {
238                OperationTypeHelper::Insert => Self::Insert,
239                OperationTypeHelper::Update => Self::Update,
240                OperationTypeHelper::Replace => Self::Replace,
241                OperationTypeHelper::Delete => Self::Delete,
242                OperationTypeHelper::Drop => Self::Drop,
243                OperationTypeHelper::Rename => Self::Rename,
244                OperationTypeHelper::DropDatabase => Self::DropDatabase,
245                OperationTypeHelper::Invalidate => Self::Invalidate,
246            },
247            OperationTypeWrapper::Unknown(s) => Self::Other(s.to_string()),
248        }
249    }
250}
251
252impl<'de> Deserialize<'de> for OperationType {
253    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
254    where
255        D: serde::Deserializer<'de>,
256    {
257        OperationTypeWrapper::deserialize(deserializer).map(OperationType::from)
258    }
259}
260
261impl Serialize for OperationType {
262    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
263    where
264        S: serde::Serializer,
265    {
266        OperationTypeWrapper::serialize(&self.into(), serializer)
267    }
268}
269
270/// Identifies the collection or database on which an event occurred.
271#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
272#[non_exhaustive]
273pub struct ChangeNamespace {
274    /// The name of the database in which the change occurred.
275    pub db: String,
276
277    /// The name of the collection in which the change occurred.
278    pub coll: Option<String>,
279}
280
281/// Identifies the type of object for a `create` event.
282#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
283#[non_exhaustive]
284pub enum ChangeNamespaceType {
285    /// A collection with no special options set.
286    Collection,
287    /// A timeseries collection.
288    Timeseries,
289    /// A view collection.
290    View,
291    /// Forward compatibility fallthrough.
292    #[serde(untagged)]
293    Other(String),
294}
mongodb/change_stream/event.rs

mongodb/change_stream/
event.rs
