Read preference option per query (#3865)

src/Adapters/Storage/Mongo/MongoCollection.js

@@ -13,13 +13,13 @@ export default class MongoCollection {
   // none, then build the geoindex.
   // This could be improved a lot but it's not clear if that's a good
   // idea. Or even if this behavior is a good idea.
-  find(query, { skip, limit, sort, keys, maxTimeMS } = {}) {
+  find(query, { skip, limit, sort, keys, maxTimeMS, readPreference } = {}) {
     // Support for Full Text Search - $text
     if(keys && keys.$score) {
       delete keys.$score;
       keys.score = {$meta: 'textScore'};
     }
-    return this._rawFind(query, { skip, limit, sort, keys, maxTimeMS })
+    return this._rawFind(query, { skip, limit, sort, keys, maxTimeMS, readPreference })
       .catch(error => {
         // Check for "no geoindex" error
         if (error.code != 17007 && !error.message.match(/unable to find index for .geoNear/)) {
@@ -35,13 +35,13 @@ export default class MongoCollection {
         index[key] = '2d';
         return this._mongoCollection.createIndex(index)
           // Retry, but just once.
-          .then(() => this._rawFind(query, { skip, limit, sort, keys, maxTimeMS }));
+          .then(() => this._rawFind(query, { skip, limit, sort, keys, maxTimeMS, readPreference }));
       });
   }
 
-  _rawFind(query, { skip, limit, sort, keys, maxTimeMS } = {}) {
+  _rawFind(query, { skip, limit, sort, keys, maxTimeMS, readPreference } = {}) {
     let findOperation = this._mongoCollection
-      .find(query, { skip, limit, sort })
+      .find(query, { skip, limit, sort, readPreference })
 
     if (keys) {
       findOperation = findOperation.project(keys);
@@ -54,8 +54,8 @@ export default class MongoCollection {
     return findOperation.toArray();
   }
 
-  count(query, { skip, limit, sort, maxTimeMS } = {}) {
-    const countOperation = this._mongoCollection.count(query, { skip, limit, sort, maxTimeMS });
+  count(query, { skip, limit, sort, maxTimeMS, readPreference } = {}) {
+    const countOperation = this._mongoCollection.count(query, { skip, limit, sort, maxTimeMS, readPreference });
 
     return countOperation;
   }

src/Adapters/Storage/Mongo/MongoStorageAdapter.js

@@ -17,6 +17,7 @@ import defaults              from '../../../defaults';
 
 const mongodb = require('mongodb');
 const MongoClient = mongodb.MongoClient;
+const ReadPreference = mongodb.ReadPreference;
 
 const MongoSchemaCollectionName = '_SCHEMA';
 
@@ -332,7 +333,7 @@ export class MongoStorageAdapter {
   }
 
   // Executes a find. Accepts: className, query in Parse format, and { skip, limit, sort }.
-  find(className, schema, query, { skip, limit, sort, keys }) {
+  find(className, schema, query, { skip, limit, sort, keys, readPreference }) {
     schema = convertParseSchemaToMongoSchema(schema);
     const mongoWhere = transformWhere(className, query, schema);
     const mongoSort = _.mapKeys(sort, (value, fieldName) => transformKey(className, fieldName, schema));
@@ -340,6 +341,7 @@ export class MongoStorageAdapter {
       memo[transformKey(className, key, schema)] = 1;
       return memo;
     }, {});
+    readPreference = this._parseReadPreference(readPreference);
     return this._adaptiveCollection(className)
       .then(collection => collection.find(mongoWhere, {
         skip,
@@ -347,6 +349,7 @@ export class MongoStorageAdapter {
         sort: mongoSort,
         keys: mongoKeys,
         maxTimeMS: this._maxTimeMS,
+        readPreference,
       }))
       .then(objects => objects.map(object => mongoObjectToParseObject(className, object, schema)))
   }
@@ -382,14 +385,41 @@ export class MongoStorageAdapter {
   }
 
   // Executes a count.
-  count(className, schema, query) {
+  count(className, schema, query, readPreference) {
     schema = convertParseSchemaToMongoSchema(schema);
+    readPreference = this._parseReadPreference(readPreference);
     return this._adaptiveCollection(className)
       .then(collection => collection.count(transformWhere(className, query, schema), {
         maxTimeMS: this._maxTimeMS,
+        readPreference,
       }));
   }
 
+  _parseReadPreference(readPreference) {
+    if (readPreference) {
+      switch (readPreference) {
+      case 'PRIMARY':
+        readPreference = ReadPreference.PRIMARY;
+        break;
+      case 'PRIMARY_PREFERRED':
+        readPreference = ReadPreference.PRIMARY_PREFERRED;
+        break;
+      case 'SECONDARY':
+        readPreference = ReadPreference.SECONDARY;
+        break;
+      case 'SECONDARY_PREFERRED':
+        readPreference = ReadPreference.SECONDARY_PREFERRED;
+        break;
+      case 'NEAREST':
+        readPreference = ReadPreference.NEAREST;
+        break;
+      default:
+        throw new Parse.Error(Parse.Error.INVALID_QUERY, 'Not supported read preference.');
+      }
+    }
+    return readPreference;
+  }
+
   performInitialization() {
     return Promise.resolve();
   }