Add more event fields: lsid, txnNumber and disambiguatedPaths

Author: arthurprs

src/change_stream/event.rs

@@ -74,6 +74,14 @@ pub struct ChangeStreamEvent<T> {
     /// The new name for the `ns` collection.  Only included for `OperationType::Rename`.
     pub to: Option<ChangeNamespace>,
 
+    /// The identifier for the session associated with the transaction.
+    /// Only present if the operation is part of a multi-document transaction.
+    pub lsid: Option<Document>,
+
+    /// Together with the lsid, a number that helps uniquely identify a transaction.
+    /// Only present if the operation is part of a multi-document transaction.
+    pub txn_number: Option<i64>,
+
     /// A `Document` that contains the `_id` of the document created or modified by the `insert`,
     /// `replace`, `delete`, `update` operations (i.e. CRUD operations). For sharded collections,
     /// also displays the full shard key for the document. The `_id` field is not repeated if it is
@@ -126,6 +134,12 @@ pub struct UpdateDescription {
 
     /// Arrays that were truncated in the `Document`.
     pub truncated_arrays: Option<Vec<TruncatedArray>>,
+
+    /// When an update event reports changes involving ambiguous fields, the disambiguatedPaths
+    /// document provides the path key with an array listing each path component.
+    /// Note: The disambiguatedPaths field is only available on change streams started with the
+    /// showExpandedEvents option
+    pub disambiguated_paths: Option<Document>,
 }
 
 /// Describes an array that has been truncated.

src/test/change_stream.rs

@@ -659,6 +659,48 @@ async fn split_large_event() -> Result<()> {
     Ok(())
 }
 
+/// Test that transaction fields are parsed correctly
+#[tokio::test]
+async fn transaction_fields() -> Result<()> {
+    let (client, coll, mut stream) =
+        match init_stream("chang_stream_transaction_fields", true).await? {
+            Some(t) => t,
+            None => return Ok(()),
+        };
+    if client.is_sharded() {
+        log_uncaptured("skipping change stream test on unsupported topology");
+        return Ok(());
+    }
+    if !client.supports_transactions() {
+        log_uncaptured(
+            "skipping change stream transaction_fields test due to lack of transaction support",
+        );
+        return Ok(());
+    }
+
+    let mut session = client.start_session().await.unwrap();
+    let session_id = session.id().get("id").cloned();
+    assert!(session_id.is_some());
+    session.start_transaction().await.unwrap();
+    coll.insert_one(doc! {"_id": 1})
+        .session(&mut session)
+        .await?;
+    session.commit_transaction().await.unwrap();
+
+    let next_event = stream.next().await.transpose()?;
+    assert!(matches!(next_event,
+        Some(ChangeStreamEvent {
+            operation_type: OperationType::Insert,
+            document_key: Some(key),
+            lsid: Some(lsid),
+            txn_number: Some(1),
+            ..
+        }) if key == doc! { "_id": 1 } && lsid.get("id") == session_id.as_ref()
+    ));
+
+    Ok(())
+}
+
 // Regression test: `Collection::watch` uses the type parameter.  This is not flagged as a test to
 // run because it's just asserting that this compiles.
 #[allow(unreachable_code, unused_variables, clippy::diverging_sub_expression)]