refactor(list-collections): support retryable reads

Author: mbroadst

lib/db.js

@@ -28,7 +28,6 @@ const AggregationCursor = require('./aggregation_cursor');
 const createListener = require('./operations/db_ops').createListener;
 const ensureIndex = require('./operations/db_ops').ensureIndex;
 const evaluate = require('./operations/db_ops').evaluate;
-const listCollectionsTransforms = require('./operations/db_ops').listCollectionsTransforms;
 const profilingInfo = require('./operations/db_ops').profilingInfo;
 const validateDatabaseName = require('./operations/db_ops').validateDatabaseName;
 
@@ -42,6 +41,7 @@ const DropCollectionOperation = require('./operations/drop').DropCollectionOpera
 const DropDatabaseOperation = require('./operations/drop').DropDatabaseOperation;
 const ExecuteDbAdminCommandOperation = require('./operations/execute_db_admin_command');
 const IndexInformationOperation = require('./operations/index_information');
+const ListCollectionsOperation = require('./operations/list_collections');
 const ProfilingLevelOperation = require('./operations/profiling_level');
 const RemoveUserOperation = require('./operations/remove_user');
 const RenameOperation = require('./operations/rename');
@@ -572,72 +572,11 @@ Db.prototype.listCollections = function(filter, options) {
   filter = filter || {};
   options = options || {};
 
-  // Shallow clone the object
-  options = Object.assign({}, options);
-  // Set the promise library
-  options.promiseLibrary = this.s.promiseLibrary;
-
-  // Ensure valid readPreference
-  options.readPreference = resolveReadPreference(this, options);
-
-  // Cursor options
-  let cursor = options.batchSize ? { batchSize: options.batchSize } : {};
-
-  // We have a list collections command
-  if (this.serverConfig.capabilities().hasListCollectionsCommand) {
-    const nameOnly = typeof options.nameOnly === 'boolean' ? options.nameOnly : false;
-    // Build the command
-    const command = { listCollections: 1, filter, cursor, nameOnly };
-    // Set the AggregationCursor constructor
-    options.cursorFactory = CommandCursor;
-    // Create the cursor
-    cursor = this.s.topology.cursor(
-      this.s.namespace.withCollection('$cmd').toString(),
-      command,
-      options
-    );
-    // Do we have a readPreference, apply it
-    if (options.readPreference) {
-      cursor.setReadPreference(options.readPreference);
-    }
-    // Return the cursor
-    return cursor;
-  }
-
-  // We cannot use the listCollectionsCommand
-  if (!this.serverConfig.capabilities().hasListCollectionsCommand) {
-    // If we have legacy mode and have not provided a full db name filter it
-    if (
-      typeof filter.name === 'string' &&
-      !new RegExp('^' + this.databaseName + '\\.').test(filter.name)
-    ) {
-      filter = Object.assign({}, filter);
-      filter.name = this.s.namespace.withCollection(filter.name).toString();
-    }
-  }
-
-  // No filter, filter by current database
-  if (filter == null) {
-    filter.name = `/${this.databaseName}/`;
-  }
-
-  // Rewrite the filter to use $and to filter out indexes
-  if (filter.name) {
-    filter = { $and: [{ name: filter.name }, { name: /^((?!\$).)*$/ }] };
-  } else {
-    filter = { name: /^((?!\$).)*$/ };
-  }
-
-  // Return options
-  const _options = { transforms: listCollectionsTransforms(this.databaseName) };
-  // Get the cursor
-  cursor = this.collection(CONSTANTS.SYSTEM_NAMESPACE_COLLECTION).find(filter, _options);
-  // Do we have a readPreference, apply it
-  if (options.readPreference) cursor.setReadPreference(options.readPreference);
-  // Set the passed in batch size if one was provided
-  if (options.batchSize) cursor = cursor.batchSize(options.batchSize);
-  // We have a fallback mode using legacy systems collections
-  return cursor;
+  return new CommandCursor(
+    this.s.topology,
+    new ListCollectionsOperation(this, filter, options),
+    options
+  );
 };
 
 /**

lib/operations/list_collections.js

@@ -0,0 +1,92 @@
+'use strict';
+
+const CommandOperationV2 = require('./command_v2');
+const Aspect = require('./operation').Aspect;
+const defineAspects = require('./operation').defineAspects;
+const maxWireVersion = require('../core/utils').maxWireVersion;
+const listCollectionsTransforms = require('./db_ops').listCollectionsTransforms;
+const CONSTANTS = require('../constants');
+
+const LIST_COLLECTIONS_WIRE_VERSION = 3;
+
+class ListCollectionsOperation extends CommandOperationV2 {
+  constructor(db, filter, options) {
+    super(db, options);
+    this.options.full = true;
+
+    this.db = db;
+    this.filter = filter;
+    this.nameOnly = !!this.options.nameOnly;
+
+    if (typeof this.options.batchSize === 'number') {
+      this.batchSize = this.options.batchSize;
+    }
+  }
+
+  execute(server, callback) {
+    if (maxWireVersion(server) < LIST_COLLECTIONS_WIRE_VERSION) {
+      let filter = this.filter;
+      const databaseName = this.db.s.namespace.db;
+
+      // If we have legacy mode and have not provided a full db name filter it
+      if (
+        typeof filter.name === 'string' &&
+        !new RegExp('^' + databaseName + '\\.').test(filter.name)
+      ) {
+        filter = Object.assign({}, filter);
+        filter.name = this.db.s.namespace.withCollection(filter.name).toString();
+      }
+
+      // No filter, filter by current database
+      if (filter == null) {
+        filter.name = `/${databaseName}/`;
+      }
+
+      // Rewrite the filter to use $and to filter out indexes
+      if (filter.name) {
+        filter = { $and: [{ name: filter.name }, { name: /^((?!\$).)*$/ }] };
+      } else {
+        filter = { name: /^((?!\$).)*$/ };
+      }
+
+      const transforms = listCollectionsTransforms(databaseName);
+      server.query(
+        `${databaseName}.${CONSTANTS.SYSTEM_NAMESPACE_COLLECTION}`,
+        { query: filter },
+        { batchSize: this.batchSize || 1000 },
+        {},
+        (err, result) => {
+          if (
+            result &&
+            result.message &&
+            result.message.documents &&
+            Array.isArray(result.message.documents)
+          ) {
+            result.message.documents = result.message.documents.map(transforms.doc);
+          }
+
+          callback(err, result);
+        }
+      );
+
+      return;
+    }
+
+    const command = {
+      listCollections: 1,
+      filter: this.filter,
+      cursor: this.batchSize ? { batchSize: this.batchSize } : {},
+      nameOnly: this.nameOnly
+    };
+
+    return super.executeCommand(server, command, callback);
+  }
+}
+
+defineAspects(ListCollectionsOperation, [
+  Aspect.READ_OPERATION,
+  Aspect.RETRYABLE,
+  Aspect.EXECUTE_WITH_SELECTION
+]);
+
+module.exports = ListCollectionsOperation;

test/functional/retryable_reads_tests.js

@@ -21,7 +21,9 @@ describe('Retryable Reads', function() {
       spec.description.match(/countDocuments/i) ||
       spec.description.match(/listIndexes/i) ||
       spec.description.match(/listDatabases/i) ||
-      spec.description.match(/listDatabaseNames/i)
+      spec.description.match(/listDatabaseNames/i) ||
+      spec.description.match(/listCollections/i) ||
+      spec.description.match(/listCollectionNames/i)
     );
   });
 });

test/functional/runner/index.js

@@ -415,6 +415,7 @@ function extractBulkRequests(requests) {
 function translateOperationName(operationName) {
   if (operationName === 'runCommand') return 'command';
   if (operationName === 'listDatabaseNames') return 'listDatabases';
+  if (operationName === 'listCollectionNames') return 'listCollections';
   return operationName;
 }
 
@@ -464,7 +465,7 @@ function resolveOperationArgs(operationName, operationArgs, context) {
   return result;
 }
 
-const CURSOR_COMMANDS = new Set(['find', 'aggregate', 'listIndexes']);
+const CURSOR_COMMANDS = new Set(['find', 'aggregate', 'listIndexes', 'listCollections']);
 const ADMIN_COMMANDS = new Set(['listDatabases']);
 
 /**
@@ -557,7 +558,7 @@ function testOperation(operation, obj, context, options) {
     obj = obj.db().admin();
   }
 
-  if (operation.name === 'listDatabaseNames') {
+  if (operation.name === 'listDatabaseNames' || operation.name === 'listCollectionNames') {
     opOptions.nameOnly = true;
   }