JAVA-2041: Respect ReadConcern in DBCollection and DBCursor

Author: jyemin

driver/src/main/com/mongodb/DB.java

@@ -72,6 +72,7 @@ public class DB {
     private final Codec<DBObject> commandCodec;
     private volatile ReadPreference readPreference;
     private volatile WriteConcern writeConcern;
+    private volatile ReadConcern readConcern;
 
     DB(final Mongo mongo, final String name, final OperationExecutor executor) {
         if (!isValidName(name)) {
@@ -146,6 +147,28 @@ public WriteConcern getWriteConcern() {
         return writeConcern != null ? writeConcern : mongo.getWriteConcern();
     }
 
+    /**
+     * Sets the read concern for this database.
+     *
+     * @param readConcern the read concern to use for this collection
+     * @since 3.2
+     * @mongodb.server.release 3.2
+     */
+    void setReadConcern(final ReadConcern readConcern) {
+        this.readConcern = readConcern;
+    }
+
+    /**
+     * Get the read concern for this database.
+     *
+     * @return the {@link com.mongodb.ReadConcern}
+     * @since 3.2
+     * @mongodb.server.release 3.2
+     */
+    ReadConcern getReadConcern() {
+        return readConcern != null ? readConcern : mongo.getReadConcern();
+    }
+
     /**
      * <p>Gets a collection with a given name. If the collection does not exist, a new collection is created.</p>
      *

driver/src/main/com/mongodb/DBCollection.java

@@ -121,6 +121,7 @@ public class DBCollection {
     private final Bytes.OptionHolder optionHolder;
     private volatile ReadPreference readPreference;
     private volatile WriteConcern writeConcern;
+    private volatile ReadConcern readConcern;
     private List<DBObject> hintFields;
     private DBEncoderFactory encoderFactory;
     private DBDecoderFactory decoderFactory;
@@ -743,7 +744,7 @@ public DBObject findOne(final DBObject query, final DBObject projection, final R
      */
     public DBObject findOne(final DBObject query, final DBObject projection, final DBObject sort,
                             final ReadPreference readPreference) {
-        return findOne(query, projection, sort, readPreference, 0, MILLISECONDS);
+        return findOne(query, projection, sort, readPreference, getReadConcern(), 0, MILLISECONDS);
     }
 
     /**
@@ -753,16 +754,19 @@ public DBObject findOne(final DBObject query, final DBObject projection, final D
      * @param projection     specifies which projection MongoDB will return from the documents in the result set.
      * @param sort           A document whose fields specify the attributes on which to sort the result set.
      * @param readPreference {@code ReadPreference} to be used for this operation
+     * @param readConcern    {@code ReadConcern} to be used for this operation
      * @param maxTime        the maximum time that the server will allow this operation to execute before killing it
      * @param maxTimeUnit    the unit that maxTime is specified in
      * @return A document that satisfies the query specified as the argument to this method.
      * @mongodb.driver.manual tutorial/query-documents/ Querying
      * @since 2.12.0
      */
     DBObject findOne(final DBObject query, final DBObject projection, final DBObject sort,
-                     final ReadPreference readPreference, final long maxTime, final TimeUnit maxTimeUnit) {
+                     final ReadPreference readPreference, final ReadConcern readConcern,
+                     final long maxTime, final TimeUnit maxTimeUnit) {
         FindOperation<DBObject> operation = new FindOperation<DBObject>(getNamespace(),
                                                                             objectCodec)
+                                                    .readConcern(readConcern)
                                                     .projection(wrapAllowNull(projection))
                                                     .sort(wrapAllowNull(sort))
                                                     .limit(-1)
@@ -924,16 +928,18 @@ public long getCount(final DBObject query, final DBObject projection, final long
      */
     public long getCount(final DBObject query, final DBObject projection, final long limit, final long skip,
                          final ReadPreference readPreference) {
-        return getCount(query, projection, limit, skip, readPreference, 0, MILLISECONDS);
+        return getCount(query, limit, skip, readPreference, getReadConcern(), 0, MILLISECONDS);
     }
 
-    long getCount(final DBObject query, final DBObject projection, final long limit, final long skip,
-                  final ReadPreference readPreference, final long maxTime, final TimeUnit maxTimeUnit) {
-        return getCount(query, projection, limit, skip, readPreference, maxTime, maxTimeUnit, null);
+    long getCount(final DBObject query, final long limit, final long skip,
+                  final ReadPreference readPreference, final ReadConcern readConcern,
+                  final long maxTime, final TimeUnit maxTimeUnit) {
+        return getCount(query, limit, skip, readPreference, readConcern, maxTime, maxTimeUnit, null);
     }
 
-    long getCount(final DBObject query, final DBObject projection, final long limit, final long skip,
-                  final ReadPreference readPreference, final long maxTime, final TimeUnit maxTimeUnit,
+    long getCount(final DBObject query, final long limit, final long skip,
+                  final ReadPreference readPreference, final ReadConcern readConcern,
+                  final long maxTime, final TimeUnit maxTimeUnit,
                   final BsonValue hint) {
 
         if (limit > Integer.MAX_VALUE) {
@@ -945,6 +951,7 @@ long getCount(final DBObject query, final DBObject projection, final long limit,
         }
 
         CountOperation operation = new CountOperation(getNamespace())
+                                       .readConcern(readConcern)
                                        .hint(hint)
                                        .skip(skip)
                                        .limit(limit)
@@ -1105,7 +1112,9 @@ public List distinct(final String fieldName, final DBObject query) {
     @SuppressWarnings("unchecked")
     public List distinct(final String fieldName, final DBObject query, final ReadPreference readPreference) {
         return new OperationIterable<BsonValue>(new DistinctOperation<BsonValue>(getNamespace(), fieldName,
-                                                new BsonValueCodec()).filter(wrap(query)),
+                                                new BsonValueCodec())
+                                                .readConcern(getReadConcern())
+                                                .filter(wrap(query)),
                                                 readPreference, executor).map(new Function<BsonValue, Object>() {
             @Override
             public Object apply(final BsonValue bsonValue) {
@@ -1186,7 +1195,7 @@ public MapReduceOutput mapReduce(final MapReduceCommand command) {
                                                                   new BsonJavaScript(command.getMap()),
                                                                   new BsonJavaScript(command.getReduce()),
                                                                   getDefaultDBObjectCodec());
-
+            operation.readConcern(getReadConcern());
             operation.filter(wrapAllowNull(command.getQuery()));
             operation.limit(command.getLimit());
             operation.maxTime(command.getMaxTime(MILLISECONDS), MILLISECONDS);
@@ -1360,6 +1369,7 @@ private Cursor aggregate(final List<? extends DBObject> pipeline, final Aggregat
             }
         } else {
             AggregateOperation<DBObject> operation = new AggregateOperation<DBObject>(getNamespace(), stages, objectCodec)
+                                                         .readConcern(getReadConcern())
                                                          .maxTime(options.getMaxTime(MILLISECONDS), MILLISECONDS)
                                                          .allowDiskUse(options.getAllowDiskUse())
                                                          .batchSize(options.getBatchSize())
@@ -1416,6 +1426,7 @@ public List<Cursor> parallelScan(final ParallelScanOptions options) {
         ParallelCollectionScanOperation<DBObject> operation = new ParallelCollectionScanOperation<DBObject>(getNamespace(),
                                                                                                             options.getNumCursors(),
                                                                                                             objectCodec)
+                                                                  .readConcern(getReadConcern())
                                                                   .batchSize(options.getBatchSize());
         List<BatchCursor<DBObject>> mongoCursors = executor.execute(operation,
                                                                     options.getReadPreference() != null ? options.getReadPreference()
@@ -1871,6 +1882,32 @@ public void setReadPreference(final ReadPreference preference) {
         this.readPreference = preference;
     }
 
+    /**
+     * Sets the read concern for this collection.
+     *
+     * @param readConcern the read concern to use for this collection
+     * @since 3.2
+     * @mongodb.server.release 3.2
+     */
+    void setReadConcern(final ReadConcern readConcern) {
+        this.readConcern = readConcern;
+    }
+
+    /**
+     * Get the read concern for this collection.
+     *
+     * @return the {@link com.mongodb.ReadConcern}
+     * @since 3.2
+     * @mongodb.server.release 3.2
+     */
+    ReadConcern getReadConcern() {
+        if (readConcern != null) {
+            return readConcern;
+        }
+        return database.getReadConcern();
+    }
+
+
     /**
      * Makes this query ok to run on a slave node
      *

driver/src/main/com/mongodb/DBCursor.java

@@ -68,6 +68,7 @@ public class DBCursor implements Cursor, Iterable<DBObject> {
     private final FindOptions findOptions;
     private int options;
     private ReadPreference readPreference;
+    private ReadConcern readConcern;
     private Decoder<DBObject> resultDecoder;
     private DBDecoderFactory decoderFactory;
     private IteratorOrArray iteratorOrArray;
@@ -489,6 +490,7 @@ public DBObject explain() {
 
     private FindOperation<DBObject> getQueryOperation(final FindOptions options, final Decoder<DBObject> decoder) {
         FindOperation<DBObject> operation = new FindOperation<DBObject>(collection.getNamespace(), decoder)
+                                                .readConcern(getReadConcern())
                                                 .filter(collection.wrapAllowNull(filter))
                                                 .batchSize(options.getBatchSize())
                                                 .skip(options.getSkip())
@@ -665,7 +667,7 @@ public List<DBObject> toArray(final int max) {
      * @see DBCursor#size
      */
     public int count() {
-        return (int) collection.getCount(getQuery(), getKeysWanted(), 0, 0, getReadPreferenceForCursor(),
+        return (int) collection.getCount(getQuery(), 0, 0, getReadPreferenceForCursor(), getReadConcern(),
                                          findOptions.getMaxTime(MILLISECONDS), MILLISECONDS,
                                          collection.wrap(modifiers).get("$hint"));
     }
@@ -678,7 +680,7 @@ public int count() {
      */
     public DBObject one() {
         return collection.findOne(getQuery(), getKeysWanted(), sort,
-                                  getReadPreferenceForCursor(), findOptions.getMaxTime(MILLISECONDS), MILLISECONDS);
+                                  getReadPreferenceForCursor(), getReadConcern(), findOptions.getMaxTime(MILLISECONDS), MILLISECONDS);
     }
 
     /**
@@ -720,8 +722,8 @@ public int itcount() {
      * @see #count()
      */
     public int size() {
-        return (int) collection.getCount(getQuery(), getKeysWanted(), findOptions.getLimit(),
-                                         findOptions.getSkip(), getReadPreference(),
+        return (int) collection.getCount(getQuery(), findOptions.getLimit(),
+                                         findOptions.getSkip(), getReadPreference(), getReadConcern(),
                                          findOptions.getMaxTime(MILLISECONDS), MILLISECONDS);
     }
 
@@ -781,6 +783,33 @@ public ReadPreference getReadPreference() {
         return readPreference;
     }
 
+
+    /**
+     * Sets the read concern for this collection.
+     *
+     * @param readConcern the read concern to use for this collection
+     * @since 3.2
+     * @mongodb.server.release 3.2
+     */
+    DBCursor setReadConcern(final ReadConcern readConcern) {
+        this.readConcern = readConcern;
+        return this;
+    }
+
+    /**
+     * Get the read concern for this collection.
+     *
+     * @return the {@link com.mongodb.ReadConcern}
+     * @since 3.2
+     * @mongodb.server.release 3.2
+     */
+    ReadConcern getReadConcern() {
+        if (readConcern != null) {
+            return readConcern;
+        }
+        return collection.getReadConcern();
+    }
+
     /**
      * Sets the factory that will be used create a {@code DBDecoder} that will be used to decode BSON documents into DBObject instances.
      *