Change streams now save the startAtOperationTime

rozza · rozza · commit 31d4887562b4 · 2018-06-19T18:03:03.000+01:00
JAVA-2889
diff --git a/config/codenarc/codenarc.xml b/config/codenarc/codenarc.xml
@@ -110,6 +110,9 @@
         <rule-config name='ParameterCount'>
             <property name='doNotApplyToFilesMatching' value='.*Specification.groovy'/>
         </rule-config>
+        <rule-config name='NestedBlockDepth'>
+            <property name='doNotApplyToFilesMatching' value='.*Specification.groovy'/>
+        </rule-config>
         <exclude name='CrapMetric'/>
         <exclude name='AbcMetric'/>
         <exclude name='MethodSize'/>
diff --git a/driver-async/src/main/com/mongodb/async/client/ChangeStreamIterable.java b/driver-async/src/main/com/mongodb/async/client/ChangeStreamIterable.java
@@ -91,7 +91,7 @@ public interface ChangeStreamIterable<TResult> extends MongoIterable<ChangeStrea
     <TDocument> MongoIterable<TDocument> withDocumentClass(Class<TDocument> clazz);
 
     /**
-     * The change stream will only provides changes that occurred after the specified timestamp.
+     * The change stream will only provide changes that occurred at or after the specified timestamp.
      *
      * <p>Any command run against the server will return an operation time that can be used here.</p>
      * <p>The default value is an operation time obtained from the server before the change stream was created.</p>
diff --git a/driver-async/src/main/com/mongodb/async/client/ClientSessionBinding.java b/driver-async/src/main/com/mongodb/async/client/ClientSessionBinding.java
@@ -25,7 +25,6 @@
 import com.mongodb.connection.ServerDescription;
 import com.mongodb.internal.session.ClientSessionContext;
 import com.mongodb.session.SessionContext;
-import org.bson.BsonTimestamp;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -66,11 +65,6 @@ public SessionContext getSessionContext() {
         return sessionContext;
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        return wrapped.getClusterTime();
-    }
-
     @Override
     public void getReadConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
         wrapped.getReadConnectionSource(new SingleResultCallback<AsyncConnectionSource>() {
diff --git a/driver-async/src/test/functional/com/mongodb/async/client/ChangeStreamsTest.java b/driver-async/src/test/functional/com/mongodb/async/client/ChangeStreamsTest.java
@@ -27,7 +27,6 @@
 import com.mongodb.client.test.CollectionHelper;
 import com.mongodb.connection.ServerVersion;
 import com.mongodb.event.CommandEvent;
-import com.mongodb.event.CommandStartedEvent;
 import com.mongodb.internal.connection.TestCommandListener;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonArray;
@@ -216,21 +215,6 @@ private void checkExpectations() {
             for (int i = 0; i < expectedEvents.size(); i++) {
                 CommandEvent expectedEvent = expectedEvents.get(i);
                 CommandEvent event = events.get(i);
-                if (expectedEvent instanceof CommandStartedEvent && event instanceof CommandStartedEvent) {
-                    BsonDocument eventCommand = ((CommandStartedEvent) event).getCommand();
-
-                    BsonDocument eventChangeStream = eventCommand.getArray("pipeline").get(0).asDocument().getDocument("$changeStream");
-
-                    BsonDocument expectedChangeStream = ((CommandStartedEvent) expectedEvent).getCommand().getArray("pipeline").get(0)
-                            .asDocument().getDocument("$changeStream");
-
-                    if (expectedChangeStream.containsKey("startAtOperationTime") && eventChangeStream.containsKey("startAtOperationTime")) {
-                        eventChangeStream.put("startAtOperationTime", expectedChangeStream.get("startAtOperationTime"));
-                    } else {
-                        eventChangeStream.remove("startAtOperationTime");
-                    }
-                }
-
                 CommandMonitoringTestHelper.assertEventsEquality(singletonList(expectedEvent), singletonList(event));
             }
         }
diff --git a/driver-core/src/main/com/mongodb/binding/AsyncClusterBinding.java b/driver-core/src/main/com/mongodb/binding/AsyncClusterBinding.java
@@ -28,7 +28,6 @@
 import com.mongodb.selector.ServerSelector;
 import com.mongodb.selector.WritableServerSelector;
 import com.mongodb.session.SessionContext;
-import org.bson.BsonTimestamp;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -85,11 +84,6 @@ public SessionContext getSessionContext() {
         return new ReadConcernAwareNoOpSessionContext(readConcern);
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        return cluster.getClusterTime();
-    }
-
     @Override
     public void getReadConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
         getAsyncClusterBindingConnectionSource(new ReadPreferenceServerSelector(readPreference), callback);
diff --git a/driver-core/src/main/com/mongodb/binding/AsyncReadBinding.java b/driver-core/src/main/com/mongodb/binding/AsyncReadBinding.java
@@ -19,7 +19,6 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.async.SingleResultCallback;
 import com.mongodb.session.SessionContext;
-import org.bson.BsonTimestamp;
 
 /**
  * An asynchronous factory of connection sources to servers that can be read from and that satisfy the specified read preference.
@@ -42,14 +41,6 @@ public interface AsyncReadBinding extends ReferenceCounted {
      */
     SessionContext getSessionContext();
 
-    /**
-     * Get the last seen cluster time
-     *
-     * @since 3.8
-     * @return the last seen cluster time
-     */
-    BsonTimestamp getClusterTime();
-
     /**
      * Returns a connection source to a server that satisfies the specified read preference.
      * @param callback the to be passed the connection source
diff --git a/driver-core/src/main/com/mongodb/binding/AsyncSingleConnectionReadBinding.java b/driver-core/src/main/com/mongodb/binding/AsyncSingleConnectionReadBinding.java
@@ -20,9 +20,8 @@
 import com.mongodb.async.SingleResultCallback;
 import com.mongodb.connection.AsyncConnection;
 import com.mongodb.connection.ServerDescription;
-import com.mongodb.session.SessionContext;
 import com.mongodb.internal.connection.NoOpSessionContext;
-import org.bson.BsonTimestamp;
+import com.mongodb.session.SessionContext;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -60,11 +59,6 @@ public SessionContext getSessionContext() {
         return NoOpSessionContext.INSTANCE;
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public void getReadConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
           callback.onResult(new AsyncSingleConnectionSource(), null);
diff --git a/driver-core/src/main/com/mongodb/binding/ClusterBinding.java b/driver-core/src/main/com/mongodb/binding/ClusterBinding.java
@@ -27,7 +27,6 @@
 import com.mongodb.selector.ServerSelector;
 import com.mongodb.selector.WritableServerSelector;
 import com.mongodb.session.SessionContext;
-import org.bson.BsonTimestamp;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -87,11 +86,6 @@ public SessionContext getSessionContext() {
         return new ReadConcernAwareNoOpSessionContext(readConcern);
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        return cluster.getClusterTime();
-    }
-
     @Override
     public ConnectionSource getWriteConnectionSource() {
         return new ClusterBindingConnectionSource(new WritableServerSelector());
diff --git a/driver-core/src/main/com/mongodb/binding/ReadBinding.java b/driver-core/src/main/com/mongodb/binding/ReadBinding.java
@@ -18,7 +18,6 @@
 
 import com.mongodb.ReadPreference;
 import com.mongodb.session.SessionContext;
-import org.bson.BsonTimestamp;
 
 /**
  * A factory of connection sources to servers that can be read from and that satisfy the specified read preference.
@@ -47,14 +46,6 @@ public interface ReadBinding extends ReferenceCounted {
      */
     SessionContext getSessionContext();
 
-    /**
-     * Get the last seen cluster time
-     *
-     * @since 3.8
-     * @return the last seen cluster time
-     */
-    BsonTimestamp getClusterTime();
-
     @Override
     ReadBinding retain();
 }
diff --git a/driver-core/src/main/com/mongodb/binding/SingleConnectionReadBinding.java b/driver-core/src/main/com/mongodb/binding/SingleConnectionReadBinding.java
@@ -19,9 +19,8 @@
 import com.mongodb.ReadPreference;
 import com.mongodb.connection.Connection;
 import com.mongodb.connection.ServerDescription;
-import com.mongodb.session.SessionContext;
 import com.mongodb.internal.connection.NoOpSessionContext;
-import org.bson.BsonTimestamp;
+import com.mongodb.session.SessionContext;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -65,11 +64,6 @@ public SessionContext getSessionContext() {
         return NoOpSessionContext.INSTANCE;
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public ReadBinding retain() {
         super.retain();
diff --git a/driver-core/src/main/com/mongodb/binding/SingleServerBinding.java b/driver-core/src/main/com/mongodb/binding/SingleServerBinding.java
@@ -22,10 +22,9 @@
 import com.mongodb.connection.Connection;
 import com.mongodb.connection.Server;
 import com.mongodb.connection.ServerDescription;
-import com.mongodb.session.SessionContext;
 import com.mongodb.internal.connection.NoOpSessionContext;
 import com.mongodb.selector.ServerAddressSelector;
-import org.bson.BsonTimestamp;
+import com.mongodb.session.SessionContext;
 
 import static com.mongodb.assertions.Assertions.notNull;
 
@@ -80,11 +79,6 @@ public SessionContext getSessionContext() {
         return NoOpSessionContext.INSTANCE;
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        return cluster.getClusterTime();
-    }
-
     @Override
     public SingleServerBinding retain() {
         super.retain();
diff --git a/driver-core/src/main/com/mongodb/operation/AggregateOperationImpl.java b/driver-core/src/main/com/mongodb/operation/AggregateOperationImpl.java
@@ -37,7 +37,6 @@
 import org.bson.BsonInt32;
 import org.bson.BsonInt64;
 import org.bson.BsonString;
-import org.bson.BsonTimestamp;
 import org.bson.BsonValue;
 import org.bson.codecs.Decoder;
 
@@ -192,7 +191,7 @@ public BatchCursor<T> execute(final ReadBinding binding) {
             public BatchCursor<T> call(final ConnectionSource source, final Connection connection) {
                 validateReadConcernAndCollation(connection, binding.getSessionContext().getReadConcern(), collation);
                 return executeWrappedCommandProtocol(binding, namespace.getDatabaseName(),
-                        getCommand(connection.getDescription(), binding.getSessionContext(), binding.getClusterTime()),
+                        getCommand(connection.getDescription(), binding.getSessionContext()),
                         CommandResultDocumentCodec.create(decoder, FIELD_NAMES_WITH_RESULT),
                         connection, transformer(source, connection));
             }
@@ -218,8 +217,7 @@ public void call(final AsyncConnectionSource source, final AsyncConnection conne
                                         wrappedCallback.onResult(null, t);
                                     } else {
                                         executeWrappedCommandProtocolAsync(binding, namespace.getDatabaseName(),
-                                                getCommand(connection.getDescription(), binding.getSessionContext(),
-                                                        binding.getClusterTime()),
+                                                getCommand(connection.getDescription(), binding.getSessionContext()),
                                                 CommandResultDocumentCodec.create(decoder, FIELD_NAMES_WITH_RESULT),
                                                 connection, asyncTransformer(source, connection), wrappedCallback);
                                     }
@@ -234,12 +232,11 @@ private boolean isInline(final ConnectionDescription description) {
         return !serverIsAtLeastVersionThreeDotSix(description) && ((useCursor != null && !useCursor));
     }
 
-    private BsonDocument getCommand(final ConnectionDescription description, final SessionContext sessionContext,
-                                    final BsonTimestamp clusterTime) {
+    private BsonDocument getCommand(final ConnectionDescription description, final SessionContext sessionContext) {
         BsonDocument commandDocument = new BsonDocument("aggregate", aggregateTarget.create());
 
         appendReadConcernToCommand(sessionContext, commandDocument);
-        commandDocument.put("pipeline", pipelineCreator.create(description, sessionContext, clusterTime));
+        commandDocument.put("pipeline", pipelineCreator.create(description, sessionContext));
         if (maxTimeMS > 0) {
             commandDocument.put("maxTimeMS", new BsonInt64(maxTimeMS));
         }
@@ -303,7 +300,7 @@ interface AggregateTarget {
     }
 
     interface PipelineCreator {
-        BsonArray create(ConnectionDescription connectionDescription, SessionContext sessionContext, BsonTimestamp clusterTime);
+        BsonArray create(ConnectionDescription connectionDescription, SessionContext sessionContext);
     }
 
     private static AggregateTarget defaultAggregateTarget(final String collectionName) {
@@ -318,8 +315,7 @@ public BsonValue create() {
     private static PipelineCreator defaultPipelineCreator(final List<BsonDocument> pipeline) {
         return new PipelineCreator() {
             @Override
-            public BsonArray create(final ConnectionDescription connectionDescription, final SessionContext sessionContext,
-                                    final BsonTimestamp clusterTime) {
+            public BsonArray create(final ConnectionDescription connectionDescription, final SessionContext sessionContext) {
                 return new BsonArray(pipeline);
             }
         };
diff --git a/driver-core/src/main/com/mongodb/operation/AsyncChangeStreamBatchCursor.java b/driver-core/src/main/com/mongodb/operation/AsyncChangeStreamBatchCursor.java
@@ -40,6 +40,9 @@ final class AsyncChangeStreamBatchCursor<T> implements AsyncBatchCursor<T> {
     AsyncChangeStreamBatchCursor(final ChangeStreamOperation<T> changeStreamOperation,
                                  final AsyncBatchCursor<RawBsonDocument> wrapped,
                                  final AsyncReadBinding binding) {
+        if (changeStreamOperation.getResumeToken() == null && changeStreamOperation.getStartAtOperationTime() == null) {
+            changeStreamOperation.startAtOperationTime(binding.getSessionContext().getOperationTime());
+        }
         this.changeStreamOperation = changeStreamOperation;
         this.resumeToken = changeStreamOperation.getResumeToken();
         this.wrapped = wrapped;
diff --git a/driver-core/src/main/com/mongodb/operation/ChangeStreamBatchCursor.java b/driver-core/src/main/com/mongodb/operation/ChangeStreamBatchCursor.java
@@ -40,6 +40,9 @@ final class ChangeStreamBatchCursor<T> implements BatchCursor<T> {
     ChangeStreamBatchCursor(final ChangeStreamOperation<T> changeStreamOperation,
                             final BatchCursor<RawBsonDocument> wrapped,
                             final ReadBinding binding) {
+        if (changeStreamOperation.getResumeToken() == null && changeStreamOperation.getStartAtOperationTime() == null) {
+            changeStreamOperation.startAtOperationTime(binding.getSessionContext().getOperationTime());
+        }
         this.changeStreamOperation = changeStreamOperation;
         this.resumeToken = changeStreamOperation.getResumeToken();
         this.wrapped = wrapped;
diff --git a/driver-core/src/main/com/mongodb/operation/ChangeStreamOperation.java b/driver-core/src/main/com/mongodb/operation/ChangeStreamOperation.java
@@ -42,7 +42,6 @@
 import java.util.concurrent.TimeUnit;
 
 import static com.mongodb.assertions.Assertions.notNull;
-import static com.mongodb.internal.operation.ServerVersionHelper.serverIsAtLeastVersionFourDotZero;
 
 /**
  * An operation that executes an {@code $changeStream} aggregation.
@@ -284,8 +283,7 @@ public BsonValue create() {
     private AggregateOperationImpl.PipelineCreator getPipelineCreator() {
         return new AggregateOperationImpl.PipelineCreator() {
             @Override
-            public BsonArray create(final ConnectionDescription description, final SessionContext sessionContext,
-                                    final BsonTimestamp clusterTime) {
+            public BsonArray create(final ConnectionDescription description, final SessionContext sessionContext) {
                 List<BsonDocument> changeStreamPipeline = new ArrayList<BsonDocument>();
                 BsonDocument changeStream = new BsonDocument("fullDocument", new BsonString(fullDocument.getValue()));
 
@@ -297,13 +295,8 @@ public BsonArray create(final ConnectionDescription description, final SessionCo
                     changeStream.append("resumeAfter", resumeToken);
                 }
 
-                BsonTimestamp startAtTime = startAtOperationTime;
-                if (startAtTime == null && resumeToken == null && serverIsAtLeastVersionFourDotZero(description)) {
-                    startAtTime = clusterTime;
-                }
-
-                if (startAtTime != null) {
-                    changeStream.append("startAtOperationTime", startAtTime);
+                if (startAtOperationTime != null) {
+                    changeStream.append("startAtOperationTime", startAtOperationTime);
                 }
 
                 changeStreamPipeline.add(new BsonDocument("$changeStream", changeStream));
diff --git a/driver-core/src/test/functional/com/mongodb/binding/AsyncSessionBinding.java b/driver-core/src/test/functional/com/mongodb/binding/AsyncSessionBinding.java
@@ -21,7 +21,6 @@
 import com.mongodb.connection.AsyncConnection;
 import com.mongodb.connection.ServerDescription;
 import com.mongodb.session.SessionContext;
-import org.bson.BsonTimestamp;
 
 import static org.bson.assertions.Assertions.notNull;
 
@@ -59,11 +58,6 @@ public SessionContext getSessionContext() {
         return sessionContext;
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        return wrapped.getClusterTime();
-    }
-
     @Override
     public void getReadConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
         wrapped.getReadConnectionSource(new SingleResultCallback<AsyncConnectionSource>() {
diff --git a/driver-core/src/test/functional/com/mongodb/binding/AsyncSingleConnectionBinding.java b/driver-core/src/test/functional/com/mongodb/binding/AsyncSingleConnectionBinding.java
@@ -24,11 +24,10 @@
 import com.mongodb.connection.Cluster;
 import com.mongodb.connection.Server;
 import com.mongodb.connection.ServerDescription;
-import com.mongodb.session.SessionContext;
 import com.mongodb.internal.connection.NoOpSessionContext;
 import com.mongodb.selector.ReadPreferenceServerSelector;
 import com.mongodb.selector.WritableServerSelector;
-import org.bson.BsonTimestamp;
+import com.mongodb.session.SessionContext;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -156,11 +155,6 @@ public SessionContext getSessionContext() {
         return NoOpSessionContext.INSTANCE;
     }
 
-    @Override
-    public BsonTimestamp getClusterTime() {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public void getReadConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
         isTrue("open", getCount() > 0);
diff --git a/driver-core/src/test/functional/com/mongodb/binding/SessionBinding.java b/driver-core/src/test/functional/com/mongodb/binding/SessionBinding.java
diff --git a/driver-core/src/test/functional/com/mongodb/binding/SingleConnectionBinding.java b/driver-core/src/test/functional/com/mongodb/binding/SingleConnectionBinding.java
diff --git a/driver-core/src/test/functional/com/mongodb/operation/ChangeStreamOperationSpecification.groovy b/driver-core/src/test/functional/com/mongodb/operation/ChangeStreamOperationSpecification.groovy
diff --git a/driver-core/src/test/resources/change-streams/change-streams.json b/driver-core/src/test/resources/change-streams/change-streams.json
diff --git a/driver-sync/src/main/com/mongodb/client/ChangeStreamIterable.java b/driver-sync/src/main/com/mongodb/client/ChangeStreamIterable.java
diff --git a/driver-sync/src/main/com/mongodb/client/internal/ClientSessionBinding.java b/driver-sync/src/main/com/mongodb/client/internal/ClientSessionBinding.java
diff --git a/driver-sync/src/test/functional/com/mongodb/client/ChangeStreamsTest.java b/driver-sync/src/test/functional/com/mongodb/client/ChangeStreamsTest.java

Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,7 @@ public interface ChangeStreamIterable<TResult> extends MongoIterable<ChangeStrea`
`91`	`91`	`<TDocument> MongoIterable<TDocument> withDocumentClass(Class<TDocument> clazz);`
`92`	`92`
`93`	`93`	`/**`
`94`		`- * The change stream will only provides changes that occurred after the specified timestamp.`
	`94`	`+ * The change stream will only provide changes that occurred at or after the specified timestamp.`
`95`	`95`	`*`
`96`	`96`	`* <p>Any command run against the server will return an operation time that can be used here.</p>`
`97`	`97`	`* <p>The default value is an operation time obtained from the server before the change stream was created.</p>`