Added MongoClient.watch and MongoDatabase.watch

JAVA-2873

Author: rozza

driver-async/src/main/com/mongodb/async/client/ChangeStreamIterable.java

@@ -21,6 +21,7 @@
 import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonDocument;
+import org.bson.BsonTimestamp;
 
 import java.util.concurrent.TimeUnit;
 
@@ -89,4 +90,17 @@ 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.
+     *
+     * <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>
+     *
+     * @param startAtOperationTime the start at operation time
+     * @since 3.8
+     * @return this
+     * @mongodb.server.release 4.0
+     * @mongodb.driver.manual reference/method/db.runCommand/
+     */
+    ChangeStreamIterable<TResult> startAtOperationTime(BsonTimestamp startAtOperationTime);
 }

driver-async/src/main/com/mongodb/async/client/ChangeStreamIterableImpl.java

@@ -22,11 +22,13 @@
 import com.mongodb.async.AsyncBatchCursor;
 import com.mongodb.client.model.Collation;
 import com.mongodb.client.model.changestream.ChangeStreamDocument;
+import com.mongodb.client.model.changestream.ChangeStreamLevel;
 import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.lang.Nullable;
 import com.mongodb.operation.AsyncReadOperation;
 import com.mongodb.operation.ChangeStreamOperation;
 import org.bson.BsonDocument;
+import org.bson.BsonTimestamp;
 import org.bson.codecs.Codec;
 import org.bson.codecs.configuration.CodecRegistry;
 import org.bson.conversions.Bson;
@@ -44,21 +46,33 @@ final class ChangeStreamIterableImpl<TResult> extends MongoIterableImpl<ChangeSt
     private final CodecRegistry codecRegistry;
     private final List<? extends Bson> pipeline;
     private final Codec<ChangeStreamDocument<TResult>> codec;
+    private final ChangeStreamLevel changeStreamLevel;
 
     private FullDocument fullDocument = FullDocument.DEFAULT;
     private BsonDocument resumeToken;
     private long maxAwaitTimeMS;
     private Collation collation;
+    private BsonTimestamp startAtOperationTime;
 
 
+    ChangeStreamIterableImpl(@Nullable final ClientSession clientSession, final String databaseName, final CodecRegistry codecRegistry,
+                             final ReadPreference readPreference, final ReadConcern readConcern, final OperationExecutor executor,
+                             final List<? extends Bson> pipeline, final Class<TResult> resultClass,
+                             final ChangeStreamLevel changeStreamLevel) {
+        this(clientSession, new MongoNamespace(databaseName, "ignored"), codecRegistry, readPreference, readConcern, executor, pipeline,
+                resultClass, changeStreamLevel);
+    }
+
     ChangeStreamIterableImpl(@Nullable final ClientSession clientSession, final MongoNamespace namespace, final CodecRegistry codecRegistry,
                              final ReadPreference readPreference, final ReadConcern readConcern, final OperationExecutor executor,
-                             final List<? extends Bson> pipeline, final Class<TResult> resultClass) {
+                             final List<? extends Bson> pipeline, final Class<TResult> resultClass,
+                             final ChangeStreamLevel changeStreamLevel) {
         super(clientSession, executor, readConcern, readPreference);
         this.namespace = notNull("namespace", namespace);
         this.codecRegistry = notNull("codecRegistry", codecRegistry);
         this.pipeline = notNull("pipeline", pipeline);
         this.codec = ChangeStreamDocument.createCodec(notNull("resultClass", resultClass), codecRegistry);
+        this.changeStreamLevel = notNull("changeStreamLevel", changeStreamLevel);
     }
 
     @Override
@@ -104,24 +118,24 @@ AsyncReadOperation<AsyncBatchCursor<TDocument>> asAsyncReadOperation() {
         };
     }
 
+    @Override
+    public ChangeStreamIterable<TResult> startAtOperationTime(final BsonTimestamp startAtOperationTime) {
+        this.startAtOperationTime = notNull("startAtOperationTime", startAtOperationTime);
+        return this;
+    }
+
     @Override
     AsyncReadOperation<AsyncBatchCursor<ChangeStreamDocument<TResult>>> asAsyncReadOperation() {
         return createChangeStreamOperation(codec);
     }
 
     private <S> AsyncReadOperation<AsyncBatchCursor<S>> createChangeStreamOperation(final Codec<S> codec) {
-        List<BsonDocument> aggregateList = createBsonDocumentList(pipeline);
-
-        ChangeStreamOperation<S> changeStreamOperation = new ChangeStreamOperation<S>(namespace, fullDocument, aggregateList, codec)
-                .maxAwaitTime(maxAwaitTimeMS, MILLISECONDS)
+        return new ChangeStreamOperation<S>(namespace, fullDocument,  createBsonDocumentList(pipeline), codec, changeStreamLevel)
                 .batchSize(getBatchSize())
-                .collation(collation);
-
-        if (resumeToken != null) {
-            changeStreamOperation.resumeAfter(resumeToken);
-        }
-
-        return changeStreamOperation;
+                .collation(collation)
+                .maxAwaitTime(maxAwaitTimeMS, MILLISECONDS)
+                .resumeAfter(resumeToken)
+                .startAtOperationTime(startAtOperationTime);
     }
 
     private List<BsonDocument> createBsonDocumentList(final List<? extends Bson> pipeline) {

driver-async/src/main/com/mongodb/async/client/ClientSessionBinding.java

@@ -25,6 +25,7 @@
 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;
 
@@ -65,6 +66,11 @@ 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>() {

driver-async/src/main/com/mongodb/async/client/MongoClient.java

@@ -20,8 +20,10 @@
 import com.mongodb.annotations.Immutable;
 import com.mongodb.async.SingleResultCallback;
 import org.bson.Document;
+import org.bson.conversions.Bson;
 
 import java.io.Closeable;
+import java.util.List;
 
 /**
  * A client-side representation of a MongoDB cluster.  Instances can represent either a standalone MongoDB instance, a replica set,
@@ -146,4 +148,100 @@ public interface MongoClient extends Closeable {
      */
     <TResult> ListDatabasesIterable<TResult> listDatabases(ClientSession clientSession, Class<TResult> resultClass);
 
+    /**
+     * Creates a change stream for this client.
+     *
+     * @return the change stream iterable
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     */
+    ChangeStreamIterable<Document> watch();
+
+    /**
+     * Creates a change stream for this client.
+     *
+     * @param resultClass the class to decode each document into
+     * @param <TResult>   the target document type of the iterable.
+     * @return the change stream iterable
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     */
+    <TResult> ChangeStreamIterable<TResult> watch(Class<TResult> resultClass);
+
+    /**
+     * Creates a change stream for this client.
+     *
+     * @param pipeline the aggregation pipeline to apply to the change stream.
+     * @return the change stream iterable
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     */
+    ChangeStreamIterable<Document> watch(List<? extends Bson> pipeline);
+
+    /**
+     * Creates a change stream for this client.
+     *
+     * @param pipeline    the aggregation pipeline to apply to the change stream
+     * @param resultClass the class to decode each document into
+     * @param <TResult>   the target document type of the iterable.
+     * @return the change stream iterable
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     */
+    <TResult> ChangeStreamIterable<TResult> watch(List<? extends Bson> pipeline, Class<TResult> resultClass);
+
+    /**
+     * Creates a change stream for this client.
+     *
+     * @param clientSession the client session with which to associate this operation
+     * @return the change stream iterable
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     */
+    ChangeStreamIterable<Document> watch(ClientSession clientSession);
+
+    /**
+     * Creates a change stream for this client.
+     *
+     * @param clientSession the client session with which to associate this operation
+     * @param resultClass the class to decode each document into
+     * @param <TResult>   the target document type of the iterable.
+     * @return the change stream iterable
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     */
+    <TResult> ChangeStreamIterable<TResult> watch(ClientSession clientSession, Class<TResult> resultClass);
+
+    /**
+     * Creates a change stream for this client.
+     *
+     * @param clientSession the client session with which to associate this operation
+     * @param pipeline the aggregation pipeline to apply to the change stream.
+     * @return the change stream iterable
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     */
+    ChangeStreamIterable<Document> watch(ClientSession clientSession, List<? extends Bson> pipeline);
+
+    /**
+     * Creates a change stream for this client.
+     *
+     * @param clientSession the client session with which to associate this operation
+     * @param pipeline    the aggregation pipeline to apply to the change stream
+     * @param resultClass the class to decode each document into
+     * @param <TResult>   the target document type of the iterable.
+     * @return the change stream iterable
+     * @since 3.8
+     * @mongodb.server.release 4.0
+     * @mongodb.driver.dochub core/changestreams Change Streams
+     */
+    <TResult> ChangeStreamIterable<TResult> watch(ClientSession clientSession, List<? extends Bson> pipeline, Class<TResult> resultClass);
+
 }

driver-async/src/main/com/mongodb/async/client/MongoClientImpl.java

@@ -21,16 +21,20 @@
 import com.mongodb.MongoClientException;
 import com.mongodb.ReadPreference;
 import com.mongodb.async.SingleResultCallback;
+import com.mongodb.client.model.changestream.ChangeStreamLevel;
 import com.mongodb.connection.Cluster;
 import com.mongodb.diagnostics.logging.Logger;
 import com.mongodb.diagnostics.logging.Loggers;
 import com.mongodb.internal.session.ServerSessionPool;
 import com.mongodb.lang.Nullable;
 import org.bson.BsonDocument;
 import org.bson.Document;
+import org.bson.conversions.Bson;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
 
 import static com.mongodb.assertions.Assertions.notNull;
 import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
@@ -157,6 +161,56 @@ public <TResult> ListDatabasesIterable<TResult> listDatabases(final ClientSessio
         return createListDatabasesIterable(clientSession, resultClass);
     }
 
+
+    @Override
+    public ChangeStreamIterable<Document> watch() {
+        return watch(Collections.<Bson>emptyList());
+    }
+
+    @Override
+    public <TResult> ChangeStreamIterable<TResult> watch(final Class<TResult> resultClass) {
+        return watch(Collections.<Bson>emptyList(), resultClass);
+    }
+
+    @Override
+    public ChangeStreamIterable<Document> watch(final List<? extends Bson> pipeline) {
+        return watch(pipeline, Document.class);
+    }
+
+    @Override
+    public <TResult> ChangeStreamIterable<TResult> watch(final List<? extends Bson> pipeline, final Class<TResult> resultClass) {
+        return createChangeStreamIterable(null, pipeline, resultClass);
+    }
+
+    @Override
+    public ChangeStreamIterable<Document> watch(final ClientSession clientSession) {
+        return watch(clientSession, Collections.<Bson>emptyList(), Document.class);
+    }
+
+    @Override
+    public <TResult> ChangeStreamIterable<TResult> watch(final ClientSession clientSession, final Class<TResult> resultClass) {
+        return watch(clientSession, Collections.<Bson>emptyList(), resultClass);
+    }
+
+    @Override
+    public ChangeStreamIterable<Document> watch(final ClientSession clientSession, final List<? extends Bson> pipeline) {
+        return watch(clientSession, pipeline, Document.class);
+    }
+
+    @Override
+    public <TResult> ChangeStreamIterable<TResult> watch(final ClientSession clientSession, final List<? extends Bson> pipeline,
+                                                         final Class<TResult> resultClass) {
+        notNull("clientSession", clientSession);
+        return createChangeStreamIterable(clientSession, pipeline, resultClass);
+    }
+
+    private <TResult> ChangeStreamIterable<TResult> createChangeStreamIterable(@Nullable final ClientSession clientSession,
+                                                                               final List<? extends Bson> pipeline,
+                                                                               final Class<TResult> resultClass) {
+        return new ChangeStreamIterableImpl<TResult>(clientSession, "admin", settings.getCodecRegistry(),
+                settings.getReadPreference(), settings.getReadConcern(), executor, pipeline, resultClass, ChangeStreamLevel.CLIENT);
+    }
+
     private <T> ListDatabasesIterable<T> createListDatabasesIterable(@Nullable final ClientSession clientSession, final Class<T> clazz) {
         return new ListDatabasesIterableImpl<T>(clientSession, clazz, settings.getCodecRegistry(),
                 ReadPreference.primary(), executor);

driver-async/src/main/com/mongodb/async/client/MongoCollectionImpl.java

@@ -45,6 +45,7 @@
 import com.mongodb.client.model.ReplaceOptions;
 import com.mongodb.client.model.UpdateOptions;
 import com.mongodb.client.model.WriteModel;
+import com.mongodb.client.model.changestream.ChangeStreamLevel;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
 import com.mongodb.internal.operation.AsyncOperations;
@@ -343,7 +344,7 @@ private <TResult> ChangeStreamIterable<TResult> createChangeStreamIterable(@Null
                                                                                final List<? extends Bson> pipeline,
                                                                                final Class<TResult> resultClass) {
         return new ChangeStreamIterableImpl<TResult>(clientSession, namespace, codecRegistry, readPreference, readConcern, executor,
-                pipeline, resultClass);
+                pipeline, resultClass, ChangeStreamLevel.COLLECTION);
     }
 
     @Override