From f07932834bc5d7c91cc71cac9ec4844780e46dfe Mon Sep 17 00:00:00 2001
From: Abraham Egnor <abraham.egnor@mongodb.com>
Date: Wed, 18 May 2022 12:23:34 -0400
Subject: [PATCH 1/3] initial poc

---
 src/change_stream/event.rs                    |  5 ++-
 .../unified/change-streams.json               | 42 +++++++++++++++++++
 2 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/src/change_stream/event.rs b/src/change_stream/event.rs
index 9a154cef7..6bcf5a34f 100644
--- a/src/change_stream/event.rs
+++ b/src/change_stream/event.rs
@@ -6,7 +6,7 @@ use crate::{cursor::CursorSpecification, options::ChangeStreamOptions};
 
 #[cfg(test)]
 use bson::Bson;
-use bson::{Document, RawBson, RawDocumentBuf, Timestamp};
+use bson::{Document, RawBson, RawDocumentBuf, Timestamp, DateTime};
 use serde::{Deserialize, Serialize};
 
 /// An opaque token used for resuming an interrupted
@@ -87,6 +87,9 @@ pub struct ChangeStreamEvent<T> {
     /// The cluster time at which the change occurred.
     pub cluster_time: Option<Timestamp>,
 
+    /// The wall time from the mongod that the change event originated from.
+    pub wall_time: Option<DateTime>,
+
     /// The `Document` created or modified by the `insert`, `replace`, `delete`, `update`
     /// operations (i.e. CRUD operations).
     ///
diff --git a/src/test/spec/json/change-streams/unified/change-streams.json b/src/test/spec/json/change-streams/unified/change-streams.json
index dfc86a913..4faf2c8e0 100644
--- a/src/test/spec/json/change-streams/unified/change-streams.json
+++ b/src/test/spec/json/change-streams/unified/change-streams.json
@@ -1428,6 +1428,48 @@
           ]
         }
       ]
+    },
+    {
+      "description": "Test wallTime field is set in a change event",
+      "runOnRequirements": [
+        {
+          "minServerVersion": "6.0.0"
+        }
+      ],
+      "operations": [
+        {
+          "name": "createChangeStream",
+          "object": "collection0",
+          "arguments": {
+            "pipeline": []
+          },
+          "saveResultAsEntity": "changeStream0"
+        },
+        {
+          "name": "insertOne",
+          "object": "collection0",
+          "arguments": {
+            "document": {
+              "_id": 1,
+              "a": 1
+            }
+          }
+        },
+        {
+          "name": "iterateUntilDocumentOrError",
+          "object": "changeStream0",
+          "expectResult": {
+            "operationType": "insert",
+            "ns": {
+              "db": "database0",
+              "coll": "collection0"
+            },
+            "wallTime": {
+              "$$exists": true
+            }
+          }
+        }
+      ]
     }
   ]
 }

From d9cea81c17caaf118dd3f69487009a2a891dc79e Mon Sep 17 00:00:00 2001
From: Abraham Egnor <abraham.egnor@mongodb.com>
Date: Wed, 18 May 2022 12:26:00 -0400
Subject: [PATCH 2/3] rustfmt

---
 src/change_stream/event.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/change_stream/event.rs b/src/change_stream/event.rs
index 6bcf5a34f..2feee4fdd 100644
--- a/src/change_stream/event.rs
+++ b/src/change_stream/event.rs
@@ -6,7 +6,7 @@ use crate::{cursor::CursorSpecification, options::ChangeStreamOptions};
 
 #[cfg(test)]
 use bson::Bson;
-use bson::{Document, RawBson, RawDocumentBuf, Timestamp, DateTime};
+use bson::{DateTime, Document, RawBson, RawDocumentBuf, Timestamp};
 use serde::{Deserialize, Serialize};
 
 /// An opaque token used for resuming an interrupted

From cae92c9b2deabc9ec9ad0d51e538864925ed0990 Mon Sep 17 00:00:00 2001
From: Abraham Egnor <abraham.egnor@mongodb.com>
Date: Wed, 18 May 2022 12:27:32 -0400
Subject: [PATCH 3/3] include yaml

---
 .../change-streams/unified/change-streams.yml | 21 +++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/src/test/spec/json/change-streams/unified/change-streams.yml b/src/test/spec/json/change-streams/unified/change-streams.yml
index 629fbe1bf..afaa77ca7 100644
--- a/src/test/spec/json/change-streams/unified/change-streams.yml
+++ b/src/test/spec/json/change-streams/unified/change-streams.yml
@@ -737,3 +737,24 @@ tests:
                 pipeline: [ { $changeStream: {} } ]
               commandName: aggregate
               databaseName: *database0
+
+  - description: "Test wallTime field is set in a change event"
+    runOnRequirements:
+      - minServerVersion: "6.0.0"
+    operations:
+      - name: createChangeStream
+        object: *collection0
+        arguments: { pipeline: [] }
+        saveResultAsEntity: &changeStream0 changeStream0
+      - name: insertOne
+        object: *collection0
+        arguments:
+          document: { "_id": 1, "a": 1 }
+      - name: iterateUntilDocumentOrError
+        object: *changeStream0
+        expectResult:
+          operationType: "insert"
+          ns:
+            db: *database0
+            coll: *collection0
+          wallTime: { $$exists: true }