JAVA-2833: Add clusterTime property to ChangeStreamDocument

jyemin · jyemin · commit 4b6a82cdabe5 · 2018-05-02T17:58:16.000-04:00
MongoDB 4.0 adds a clusterTime field to every change stream document,
so the driver now exposes it as a property to the statically typed
ChangeStreamDocument class
diff --git a/driver-core/src/main/com/mongodb/client/model/changestream/ChangeStreamDocument.java b/driver-core/src/main/com/mongodb/client/model/changestream/ChangeStreamDocument.java
@@ -17,7 +17,9 @@
 package com.mongodb.client.model.changestream;
 
 import com.mongodb.MongoNamespace;
+import com.mongodb.lang.Nullable;
 import org.bson.BsonDocument;
+import org.bson.BsonTimestamp;
 import org.bson.codecs.Codec;
 import org.bson.codecs.configuration.CodecRegistry;
 import org.bson.codecs.pojo.annotations.BsonCreator;
@@ -41,6 +43,7 @@ public final class ChangeStreamDocument<TDocument> {
     private final MongoNamespace namespace;
     private final TDocument fullDocument;
     private final BsonDocument documentKey;
+    private final BsonTimestamp clusterTime;
     private final OperationType operationType;
     private final UpdateDescription updateDescription;
 
@@ -53,18 +56,43 @@ public final class ChangeStreamDocument<TDocument> {
      * @param fullDocument the fullDocument
      * @param operationType the operation type
      * @param updateDescription the update description
+     * @deprecated Prefer {@link #ChangeStreamDocument(BsonDocument, MongoNamespace, Object, BsonDocument, BsonTimestamp, OperationType,
+     *                                                 UpdateDescription)}
+     */
+    @Deprecated
+    public ChangeStreamDocument(@BsonProperty("resumeToken") final BsonDocument resumeToken,
+                                @BsonProperty("namespace") final MongoNamespace namespace,
+                                @BsonProperty("fullDocument") final TDocument fullDocument,
+                                @BsonProperty("documentKey") final BsonDocument documentKey,
+                                @BsonProperty("operationType") final OperationType operationType,
+                                @BsonProperty("updateDescription") final UpdateDescription updateDescription) {
+        this(resumeToken, namespace, fullDocument, documentKey, null, operationType, updateDescription);
+    }
+
+    /**
+     * Creates a new instance
+     *
+     * @param resumeToken the resume token
+     * @param namespace the namespace
+     * @param documentKey a document containing the _id of the changed document
+     * @param clusterTime the cluster time at which the change occurred
+     * @param fullDocument the fullDocument
+     * @param operationType the operation type
+     * @param updateDescription the update description
      */
     @BsonCreator
     public ChangeStreamDocument(@BsonProperty("resumeToken") final BsonDocument resumeToken,
                                 @BsonProperty("namespace") final MongoNamespace namespace,
                                 @BsonProperty("fullDocument") final TDocument fullDocument,
                                 @BsonProperty("documentKey") final BsonDocument documentKey,
+                                @Nullable @BsonProperty("clusterTime") final BsonTimestamp clusterTime,
                                 @BsonProperty("operationType") final OperationType operationType,
                                 @BsonProperty("updateDescription") final UpdateDescription updateDescription) {
         this.resumeToken = resumeToken;
         this.namespace = namespace;
         this.documentKey = documentKey;
         this.fullDocument = fullDocument;
+        this.clusterTime = clusterTime;
         this.operationType = operationType;
         this.updateDescription = updateDescription;
     }
@@ -111,6 +139,18 @@ public BsonDocument getDocumentKey() {
         return documentKey;
     }
 
+    /**
+     * Gets the cluster time at which the change occurred.
+     *
+     * @return the cluster time at which the change occurred
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     */
+    @Nullable
+    public BsonTimestamp getClusterTime() {
+        return clusterTime;
+    }
+
     /**
      * Returns the operationType
      *
@@ -165,6 +205,9 @@ public boolean equals(final Object o) {
         if (documentKey != null ? !documentKey.equals(that.documentKey) : that.documentKey != null) {
             return false;
         }
+        if (clusterTime != null ? !clusterTime.equals(that.clusterTime) : that.clusterTime != null) {
+            return false;
+        }
         if (operationType != that.operationType) {
             return false;
         }
@@ -181,6 +224,7 @@ public int hashCode() {
         result = 31 * result + (namespace != null ? namespace.hashCode() : 0);
         result = 31 * result + (fullDocument != null ? fullDocument.hashCode() : 0);
         result = 31 * result + (documentKey != null ? documentKey.hashCode() : 0);
+        result = 31 * result + (clusterTime != null ? clusterTime.hashCode() : 0);
         result = 31 * result + (operationType != null ? operationType.hashCode() : 0);
         result = 31 * result + (updateDescription != null ? updateDescription.hashCode() : 0);
         return result;
@@ -193,6 +237,7 @@ public String toString() {
                 + ", namespace=" + namespace
                 + ", fullDocument=" + fullDocument
                 + ", documentKey=" + documentKey
+                + ", clusterTime=" + clusterTime
                 + ", operationType=" + operationType
                 + ", updateDescription=" + updateDescription
                 + "}";
diff --git a/driver-core/src/main/com/mongodb/client/model/changestream/ChangeStreamDocumentCodec.java b/driver-core/src/main/com/mongodb/client/model/changestream/ChangeStreamDocumentCodec.java
@@ -19,8 +19,10 @@
 import com.mongodb.MongoNamespace;
 import org.bson.BsonDocument;
 import org.bson.BsonReader;
+import org.bson.BsonTimestamp;
 import org.bson.BsonWriter;
 import org.bson.codecs.BsonDocumentCodec;
+import org.bson.codecs.BsonTimestampCodec;
 import org.bson.codecs.Codec;
 import org.bson.codecs.DecoderContext;
 import org.bson.codecs.EncoderContext;
@@ -44,6 +46,7 @@ final class ChangeStreamDocumentCodec<TResult> implements Codec<ChangeStreamDocu
     ChangeStreamDocumentCodec(final Class<TResult> fullDocumentClass, final CodecRegistry codecRegistry) {
 
         ClassModelBuilder<ChangeStreamDocument> classModelBuilder = ClassModel.builder(ChangeStreamDocument.class);
+        ((PropertyModelBuilder<BsonTimestamp>) classModelBuilder.getProperty("clusterTime")).codec(new BsonTimestampCodec());
         ((PropertyModelBuilder<BsonDocument>) classModelBuilder.getProperty("documentKey")).codec(BSON_DOCUMENT_CODEC);
         ((PropertyModelBuilder<TResult>) classModelBuilder.getProperty("fullDocument")).codec(codecRegistry.get(fullDocumentClass));
         ((PropertyModelBuilder<BsonDocument>) classModelBuilder.getProperty("resumeToken")).codec(BSON_DOCUMENT_CODEC);
diff --git a/driver-core/src/test/unit/com/mongodb/client/model/changestream/ChangeStreamDocumentCodecSpecification.groovy b/driver-core/src/test/unit/com/mongodb/client/model/changestream/ChangeStreamDocumentCodecSpecification.groovy
@@ -22,6 +22,7 @@ import org.bson.BsonDocumentReader
 import org.bson.BsonDocumentWriter
 import org.bson.BsonInt32
 import org.bson.BsonReader
+import org.bson.BsonTimestamp
 import org.bson.Document
 import org.bson.codecs.BsonValueCodecProvider
 import org.bson.codecs.DecoderContext
@@ -61,6 +62,7 @@ class ChangeStreamDocumentCodecSpecification extends Specification {
                         new MongoNamespace('databaseName.collectionName'),
                         Document.parse('{key: "value for fullDocument"}'),
                         new BsonDocument('_id', new BsonInt32(1)),
+                        new BsonTimestamp(1234, 2),
                         OperationType.INSERT,
                         null
                 ),
@@ -69,14 +71,15 @@ class ChangeStreamDocumentCodecSpecification extends Specification {
                         new MongoNamespace('databaseName.collectionName'),
                         BsonDocument.parse('{key: "value for fullDocument"}'),
                         new BsonDocument('_id', new BsonInt32(2)),
+                        null,
                         OperationType.UPDATE,
                         new UpdateDescription(['a', 'b'], BsonDocument.parse('{c: 1}'))
                 )
         ]
         clazz << [Document, BsonDocument]
         json << [
             '''{_id: {token: true}, ns: {db: "databaseName", coll: "collectionName"}, documentKey : {_id : 1},
-                fullDocument: {key: "value for fullDocument"},
+                fullDocument: {key: "value for fullDocument"}, clusterTime: { "$timestamp" : { "t" : 1234, "i" : 2 } }
                 operationType: "insert"}''',
             '''{_id: {token: true}, ns: {db: "databaseName", coll: "collectionName"}, documentKey : {_id : 2},
                 fullDocument: {key: "value for fullDocument"},
diff --git a/driver-core/src/test/unit/com/mongodb/client/model/changestream/ChangeStreamDocumentSpecification.groovy b/driver-core/src/test/unit/com/mongodb/client/model/changestream/ChangeStreamDocumentSpecification.groovy
@@ -18,6 +18,7 @@ package com.mongodb.client.model.changestream
 
 import com.mongodb.MongoNamespace
 import org.bson.BsonDocument
+import org.bson.BsonTimestamp
 import org.bson.RawBsonDocument
 import spock.lang.Specification
 
@@ -29,17 +30,19 @@ class ChangeStreamDocumentSpecification extends Specification {
         def namespace = new MongoNamespace('databaseName.collectionName')
         def fullDocument = BsonDocument.parse('{key: "value for fullDocument"}')
         def documentKey = BsonDocument.parse('{_id : 1}')
+        def clusterTime = new BsonTimestamp(1234, 2)
         def operationType = OperationType.UPDATE
         def updateDesc = new UpdateDescription(['a', 'b'], BsonDocument.parse('{c: 1}'))
 
         when:
-        def changeStreamDocument = new ChangeStreamDocument<BsonDocument>(resumeToken, namespace, fullDocument, documentKey,
+        def changeStreamDocument = new ChangeStreamDocument<BsonDocument>(resumeToken, namespace, fullDocument, documentKey, clusterTime,
                 operationType, updateDesc)
 
         then:
         changeStreamDocument.getResumeToken() == resumeToken
         changeStreamDocument.getFullDocument() == fullDocument
         changeStreamDocument.getDocumentKey() == documentKey
+        changeStreamDocument.getClusterTime() == clusterTime
         changeStreamDocument.getNamespace() == namespace
         changeStreamDocument.getOperationType() == operationType
         changeStreamDocument.getUpdateDescription() == updateDesc
