CSHARP-4150: Always report 'wallTime' in the change stream event output. (#802)

Author: JamesKovacs

src/MongoDB.Driver.Core/ChangeStreamDocument.cs

@@ -13,6 +13,7 @@
 * limitations under the License.
 */
 
+using System;
 using MongoDB.Bson;
 using MongoDB.Bson.Serialization;
 using MongoDB.Bson.Serialization.Attributes;
@@ -152,5 +153,13 @@ public TDocument FullDocumentBeforeChange
         /// The update description.
         /// </value>
         public ChangeStreamUpdateDescription UpdateDescription => GetValue<ChangeStreamUpdateDescription>(nameof(UpdateDescription), null);
+
+        /// <summary>
+        /// Gets the wall time of the change stream event.
+        /// </summary>
+        /// <value>
+        /// The wall time.
+        /// </value>
+        public DateTime? WallTime => GetValue<DateTime?>(nameof(WallTime), null);
     }
 }

src/MongoDB.Driver.Core/ChangeStreamDocumentSerializer.cs

@@ -50,6 +50,7 @@ public ChangeStreamDocumentSerializer(
             RegisterMember("RenameTo", "to", ChangeStreamDocumentCollectionNamespaceSerializer.Instance);
             RegisterMember("ResumeToken", "_id", BsonDocumentSerializer.Instance);
             RegisterMember("UpdateDescription", "updateDescription", ChangeStreamUpdateDescriptionSerializer.Instance);
+            RegisterMember("WallTime", "wallTime", DateTimeSerializer.UtcInstance);
         }
 
         // public methods

tests/MongoDB.Driver.Core.Tests/ChangeStreamDocumentSerializerTests.cs

@@ -36,7 +36,7 @@ public void constructor_should_initialize_instance()
             var result = new ChangeStreamDocumentSerializer<BsonDocument>(documentSerializer);
 
             result._documentSerializer().Should().BeSameAs(documentSerializer);
-            result._memberSerializationInfo().Count.Should().Be(10);
+            result._memberSerializationInfo().Count.Should().Be(11);
             AssertRegisteredMember(result, "ClusterTime", "clusterTime", BsonTimestampSerializer.Instance);
             AssertRegisteredMember(result, "CollectionNamespace", "ns", ChangeStreamDocumentCollectionNamespaceSerializer.Instance);
             AssertRegisteredMember(result, "DatabaseNamespace", "ns", ChangeStreamDocumentDatabaseNamespaceSerializer.Instance);
@@ -47,6 +47,7 @@ public void constructor_should_initialize_instance()
             AssertRegisteredMember(result, "RenameTo", "to", ChangeStreamDocumentCollectionNamespaceSerializer.Instance);
             AssertRegisteredMember(result, "ResumeToken", "_id", BsonDocumentSerializer.Instance);
             AssertRegisteredMember(result, "UpdateDescription", "updateDescription", ChangeStreamUpdateDescriptionSerializer.Instance);
+            AssertRegisteredMember(result, "WallTime", "wallTime", DateTimeSerializer.UtcInstance);
         }
 
         [Fact]

tests/MongoDB.Driver.Core.Tests/ChangeStreamDocumentTests.cs

@@ -403,6 +403,29 @@ public void UpdateDescription_should_return_null_when_not_present()
             result.Should().BeNull();
         }
 
+        [Fact]
+        public void WallTime_should_return_expected_result()
+        {
+            var value = new DateTime(2022, 5, 25, 12, 35, 42, 144, DateTimeKind.Utc);
+            var backingDocument = new BsonDocument { { "other", 1 }, { "wallTime", value } };
+            var subject = CreateSubject(backingDocument: backingDocument);
+
+            var result = subject.WallTime;
+
+            result.Should().Be(value);
+        }
+
+        [Fact]
+        public void WallTime_should_return_null_when_not_present()
+        {
+            var backingDocument = new BsonDocument { { "other", 1 } };
+            var subject = CreateSubject(backingDocument: backingDocument);
+
+            var result = subject.WallTime;
+
+            result.Should().NotHaveValue();
+        }
+
         // private methods
         private ChangeStreamDocument<BsonDocument> CreateSubject(
             BsonDocument backingDocument = null,

tests/MongoDB.Driver.Tests/Specifications/change-streams/tests/unified/change-streams.json

@@ -1748,6 +1748,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
+            }
+          }
+        }
+      ]
     }
   ]
 }

tests/MongoDB.Driver.Tests/Specifications/change-streams/tests/unified/change-streams.yml

@@ -904,3 +904,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 }