node_modules/mongodb-core/lib/sdam/cursor.js

'use strict';

const Logger = require('../connection/logger');
const BSON = require('../connection/utils').retrieveBSON();
const MongoError = require('../error').MongoError;
const MongoNetworkError = require('../error').MongoNetworkError;
const mongoErrorContextSymbol = require('../error').mongoErrorContextSymbol;
const Long = BSON.Long;
const deprecate = require('util').deprecate;
const readPreferenceServerSelector = require('./server_selectors').readPreferenceServerSelector;
const ReadPreference = require('../topologies/read_preference');

/**
 * Handle callback (including any exceptions thrown)
 */
function handleCallback(callback, err, result) {
  try {
    callback(err, result);
  } catch (err) {
    process.nextTick(function() {
      throw err;
    });
  }
}

/**
 * This is a cursor results callback
 *
 * @callback resultCallback
 * @param {error} error An error object. Set to null if no error present
 * @param {object} document
 */

/**
 * An internal class that embodies a cursor on MongoDB, allowing for iteration over the
 * results returned from a query.
 *
 * @property {number} cursorBatchSize The current cursorBatchSize for the cursor
 * @property {number} cursorLimit The current cursorLimit for the cursor
 * @property {number} cursorSkip The current cursorSkip for the cursor
 */
class Cursor {
  /**
   * Create a cursor
   *
   * @param {object} bson An instance of the BSON parser
   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
   * @param {{object}|Long} cmd The selector (can be a command or a cursorId)
   * @param {object} [options=null] Optional settings.
   * @param {object} [options.batchSize=1000] Batchsize for the operation
   * @param {array} [options.documents=[]] Initial documents list for cursor
   * @param {object} [options.transforms=null] Transform methods for the cursor results
   * @param {function} [options.transforms.query] Transform the value returned from the initial query
   * @param {function} [options.transforms.doc] Transform each document returned from Cursor.prototype.next
   * @param {object} topology The server topology instance.
   * @param {object} topologyOptions The server topology options.
   */
  constructor(bson, ns, cmd, options, topology, topologyOptions) {
    options = options || {};

    // Cursor pool
    this.pool = null;
    // Cursor server
    this.server = null;

    // Do we have a not connected handler
    this.disconnectHandler = options.disconnectHandler;

    // Set local values
    this.bson = bson;
    this.ns = ns;
    this.cmd = cmd;
    this.options = options;
    this.topology = topology;

    // All internal state
    this.s = {
      cursorId: null,
      cmd: cmd,
      documents: options.documents || [],
      cursorIndex: 0,
      dead: false,
      killed: false,
      init: false,
      notified: false,
      limit: options.limit || cmd.limit || 0,
      skip: options.skip || cmd.skip || 0,
      batchSize: options.batchSize || cmd.batchSize || 1000,
      currentLimit: 0,
      // Result field name if not a cursor (contains the array of results)
      transforms: options.transforms
    };

    if (typeof options.session === 'object') {
      this.s.session = options.session;
    }

    // Add promoteLong to cursor state
    if (typeof topologyOptions.promoteLongs === 'boolean') {
      this.s.promoteLongs = topologyOptions.promoteLongs;
    } else if (typeof options.promoteLongs === 'boolean') {
      this.s.promoteLongs = options.promoteLongs;
    }

    // Add promoteValues to cursor state
    if (typeof topologyOptions.promoteValues === 'boolean') {
      this.s.promoteValues = topologyOptions.promoteValues;
    } else if (typeof options.promoteValues === 'boolean') {
      this.s.promoteValues = options.promoteValues;
    }

    // Add promoteBuffers to cursor state
    if (typeof topologyOptions.promoteBuffers === 'boolean') {
      this.s.promoteBuffers = topologyOptions.promoteBuffers;
    } else if (typeof options.promoteBuffers === 'boolean') {
      this.s.promoteBuffers = options.promoteBuffers;
    }

    if (topologyOptions.reconnect) {
      this.s.reconnect = topologyOptions.reconnect;
    }

    // Logger
    this.logger = Logger('Cursor', topologyOptions);

    //
    // Did we pass in a cursor id
    if (typeof cmd === 'number') {
      this.s.cursorId = Long.fromNumber(cmd);
      this.s.lastCursorId = this.s.cursorId;
    } else if (cmd instanceof Long) {
      this.s.cursorId = cmd;
      this.s.lastCursorId = cmd;
    }
  }

  setCursorBatchSize(value) {
    this.s.batchSize = value;
  }

  cursorBatchSize() {
    return this.s.batchSize;
  }

  setCursorLimit(value) {
    this.s.limit = value;
  }

  cursorLimit() {
    return this.s.limit;
  }

  setCursorSkip(value) {
    this.s.skip = value;
  }

  cursorSkip() {
    return this.s.skip;
  }

  _endSession(options, callback) {
    if (typeof options === 'function') {
      callback = options;
      options = {};
    }
    options = options || {};

    const session = this.s.session;
    if (session && (options.force || session.owner === this)) {
      this.s.session = undefined;
      session.endSession(callback);
      return true;
    }

    if (callback) {
      callback();
    }

    return false;
  }

  /**
   * Clone the cursor
   * @method
   * @return {Cursor}
   */
  clone() {
    return this.topology.cursor(this.ns, this.cmd, this.options);
  }

  /**
   * Checks if the cursor is dead
   * @method
   * @return {boolean} A boolean signifying if the cursor is dead or not
   */
  isDead() {
    return this.s.dead === true;
  }

  /**
   * Checks if the cursor was killed by the application
   * @method
   * @return {boolean} A boolean signifying if the cursor was killed by the application
   */
  isKilled() {
    return this.s.killed === true;
  }

  /**
   * Checks if the cursor notified it's caller about it's death
   * @method
   * @return {boolean} A boolean signifying if the cursor notified the callback
   */
  isNotified() {
    return this.s.notified === true;
  }

  /**
   * Returns current buffered documents length
   * @method
   * @return {number} The number of items in the buffered documents
   */
  bufferedCount() {
    return this.s.documents.length - this.s.cursorIndex;
  }

  /**
   * Kill the cursor
   *
   * @param {resultCallback} callback A callback function
   */
  kill(callback) {
    // Set cursor to dead
    this.s.dead = true;
    this.s.killed = true;
    // Remove documents
    this.s.documents = [];

    // If no cursor id just return
    if (this.s.cursorId == null || this.s.cursorId.isZero() || this.s.init === false) {
      if (callback) callback(null, null);
      return;
    }

    // Default pool
    const pool = this.s.server.s.pool;

    // Execute command
    this.s.server.s.wireProtocolHandler.killCursor(this.bson, this.ns, this.s, pool, callback);
  }

  /**
   * Resets the cursor
   */
  rewind() {
    if (this.s.init) {
      if (!this.s.dead) {
        this.kill();
      }

      this.s.currentLimit = 0;
      this.s.init = false;
      this.s.dead = false;
      this.s.killed = false;
      this.s.notified = false;
      this.s.documents = [];
      this.s.cursorId = null;
      this.s.cursorIndex = 0;
    }
  }

  /**
   * Returns current buffered documents
   * @method
   * @return {Array} An array of buffered documents
   */
  readBufferedDocuments(number) {
    const unreadDocumentsLength = this.s.documents.length - this.s.cursorIndex;
    const length = number < unreadDocumentsLength ? number : unreadDocumentsLength;
    let elements = this.s.documents.slice(this.s.cursorIndex, this.s.cursorIndex + length);

    // Transform the doc with passed in transformation method if provided
    if (this.s.transforms && typeof this.s.transforms.doc === 'function') {
      // Transform all the elements
      for (let i = 0; i < elements.length; i++) {
        elements[i] = this.s.transforms.doc(elements[i]);
      }
    }

    // Ensure we do not return any more documents than the limit imposed
    // Just return the number of elements up to the limit
    if (this.s.limit > 0 && this.s.currentLimit + elements.length > this.s.limit) {
      elements = elements.slice(0, this.s.limit - this.s.currentLimit);
      this.kill();
    }

    // Adjust current limit
    this.s.currentLimit = this.s.currentLimit + elements.length;
    this.s.cursorIndex = this.s.cursorIndex + elements.length;

    // Return elements
    return elements;
  }

  /**
   * Retrieve the next document from the cursor
   *
   * @param {resultCallback} callback A callback function
   */
  next(callback) {
    nextFunction(this, callback);
  }
}

Cursor.prototype._find = deprecate(
  callback => _find(this, callback),
  '_find() is deprecated, please stop using it'
);

Cursor.prototype._getmore = deprecate(
  callback => _getmore(this, callback),
  '_getmore() is deprecated, please stop using it'
);

function _getmore(cursor, callback) {
  if (cursor.logger.isDebug()) {
    cursor.logger.debug(`schedule getMore call for query [${JSON.stringify(cursor.query)}]`);
  }

  // Determine if it's a raw query
  const raw = cursor.options.raw || cursor.cmd.raw;

  // Set the current batchSize
  let batchSize = cursor.s.batchSize;
  if (cursor.s.limit > 0 && cursor.s.currentLimit + batchSize > cursor.s.limit) {
    batchSize = cursor.s.limit - cursor.s.currentLimit;
  }

  // Default pool
  const pool = cursor.s.server.s.pool;

  // We have a wire protocol handler
  cursor.s.server.s.wireProtocolHandler.getMore(
    cursor.bson,
    cursor.ns,
    cursor.s,
    batchSize,
    raw,
    pool,
    cursor.options,
    callback
  );
}

function _find(cursor, callback) {
  if (cursor.logger.isDebug()) {
    cursor.logger.debug(
      `issue initial query [${JSON.stringify(cursor.cmd)}] with flags [${JSON.stringify(
        cursor.query
      )}]`
    );
  }

  const queryCallback = (err, r) => {
    if (err) return callback(err);

    // Get the raw message
    const result = r.message;

    // Query failure bit set
    if (result.queryFailure) {
      return callback(new MongoError(result.documents[0]), null);
    }

    // Check if we have a command cursor
    if (
      Array.isArray(result.documents) &&
      result.documents.length === 1 &&
      (!cursor.cmd.find || (cursor.cmd.find && cursor.cmd.virtual === false)) &&
      (result.documents[0].cursor !== 'string' ||
        result.documents[0]['$err'] ||
        result.documents[0]['errmsg'] ||
        Array.isArray(result.documents[0].result))
    ) {
      // We have a an error document return the error
      if (result.documents[0]['$err'] || result.documents[0]['errmsg']) {
        return callback(new MongoError(result.documents[0]), null);
      }

      // We have a cursor document
      if (result.documents[0].cursor != null && typeof result.documents[0].cursor !== 'string') {
        const id = result.documents[0].cursor.id;
        // If we have a namespace change set the new namespace for getmores
        if (result.documents[0].cursor.ns) {
          cursor.ns = result.documents[0].cursor.ns;
        }
        // Promote id to long if needed
        cursor.s.cursorId = typeof id === 'number' ? Long.fromNumber(id) : id;
        cursor.s.lastCursorId = cursor.s.cursorId;
        // If we have a firstBatch set it
        if (Array.isArray(result.documents[0].cursor.firstBatch)) {
          cursor.s.documents = result.documents[0].cursor.firstBatch;
        }

        // Return after processing command cursor
        return callback(null, result);
      }

      if (Array.isArray(result.documents[0].result)) {
        cursor.s.documents = result.documents[0].result;
        cursor.s.cursorId = Long.ZERO;
        return callback(null, result);
      }
    }

    // Otherwise fall back to regular find path
    cursor.s.cursorId = result.cursorId;
    cursor.s.documents = result.documents;
    cursor.s.lastCursorId = result.cursorId;

    // Transform the results with passed in transformation method if provided
    if (cursor.s.transforms && typeof cursor.s.transforms.query === 'function') {
      cursor.s.documents = cursor.s.transforms.query(result);
    }

    // Return callback
    callback(null, result);
  };

  // Options passed to the pool
  const queryOptions = {};

  // If we have a raw query decorate the function
  if (cursor.options.raw || cursor.cmd.raw) {
    queryOptions.raw = cursor.options.raw || cursor.cmd.raw;
  }

  // Do we have documentsReturnedIn set on the query
  if (typeof cursor.query.documentsReturnedIn === 'string') {
    queryOptions.documentsReturnedIn = cursor.query.documentsReturnedIn;
  }

  // Add promote Long value if defined
  if (typeof cursor.s.promoteLongs === 'boolean') {
    queryOptions.promoteLongs = cursor.s.promoteLongs;
  }

  // Add promote values if defined
  if (typeof cursor.s.promoteValues === 'boolean') {
    queryOptions.promoteValues = cursor.s.promoteValues;
  }

  // Add promote values if defined
  if (typeof cursor.s.promoteBuffers === 'boolean') {
    queryOptions.promoteBuffers = cursor.s.promoteBuffers;
  }

  if (typeof cursor.s.session === 'object') {
    queryOptions.session = cursor.s.session;
  }

  // Write the initial command out
  cursor.s.server.s.pool.write(cursor.query, queryOptions, queryCallback);
}

/**
 * Validate if the pool is dead and return error
 */
function isConnectionDead(cursor, callback) {
  if (cursor.pool && cursor.pool.isDestroyed()) {
    cursor.s.killed = true;
    const err = new MongoNetworkError(
      `connection to host ${cursor.pool.host}:${cursor.pool.port} was destroyed`
    );
    _setCursorNotifiedImpl(cursor, () => callback(err));
    return true;
  }

  return false;
}

/**
 * Validate if the cursor is dead but was not explicitly killed by user
 */
function isCursorDeadButNotkilled(cursor, callback) {
  // Cursor is dead but not marked killed, return null
  if (cursor.s.dead && !cursor.s.killed) {
    cursor.s.killed = true;
    setCursorNotified(cursor, callback);
    return true;
  }

  return false;
}

/**
 * Validate if the cursor is dead and was killed by user
 */
function isCursorDeadAndKilled(cursor, callback) {
  if (cursor.s.dead && cursor.s.killed) {
    handleCallback(callback, new MongoError('cursor is dead'));
    return true;
  }

  return false;
}

/**
 * Validate if the cursor was killed by the user
 */
function isCursorKilled(cursor, callback) {
  if (cursor.s.killed) {
    setCursorNotified(cursor, callback);
    return true;
  }

  return false;
}

/**
 * Mark cursor as being dead and notified
 */
function setCursorDeadAndNotified(cursor, callback) {
  cursor.s.dead = true;
  setCursorNotified(cursor, callback);
}

/**
 * Mark cursor as being notified
 */
function setCursorNotified(cursor, callback) {
  _setCursorNotifiedImpl(cursor, () => handleCallback(callback, null, null));
}

function _setCursorNotifiedImpl(cursor, callback) {
  cursor.s.notified = true;
  cursor.s.documents = [];
  cursor.s.cursorIndex = 0;
  if (cursor._endSession) {
    return cursor._endSession(undefined, () => callback());
  }
  return callback();
}

function initializeCursorAndRetryNext(cursor, callback) {
  cursor.topology.selectServer(
    readPreferenceServerSelector(cursor.options.readPreference || ReadPreference.primary),
    (err, server) => {
      if (err) {
        callback(err, null);
        return;
      }

      cursor.s.server = server;
      cursor.s.init = true;

      // check if server supports collation
      // NOTE: this should be a part of the selection predicate!
      if (cursor.cmd && cursor.cmd.collation && cursor.server.description.maxWireVersion < 5) {
        callback(new MongoError(`server ${cursor.server.name} does not support collation`));
        return;
      }

      try {
        cursor.query = cursor.s.server.s.wireProtocolHandler.command(
          cursor.bson,
          cursor.ns,
          cursor.cmd,
          cursor.s,
          cursor.topology,
          cursor.options
        );

        nextFunction(cursor, callback);
      } catch (err) {
        callback(err);
        return;
      }
    }
  );
}

function nextFunction(cursor, callback) {
  // We have notified about it
  if (cursor.s.notified) {
    return callback(new Error('cursor is exhausted'));
  }

  // Cursor is killed return null
  if (isCursorKilled(cursor, callback)) return;

  // Cursor is dead but not marked killed, return null
  if (isCursorDeadButNotkilled(cursor, callback)) return;

  // We have a dead and killed cursor, attempting to call next should error
  if (isCursorDeadAndKilled(cursor, callback)) return;

  // We have just started the cursor
  if (!cursor.s.init) {
    return initializeCursorAndRetryNext(cursor, callback);
  }

  // If we don't have a cursorId execute the first query
  if (cursor.s.cursorId == null) {
    // Check if pool is dead and return if not possible to
    // execute the query against the db
    if (isConnectionDead(cursor, callback)) return;

    // query, cmd, options, s, callback
    return _find(cursor, function(err) {
      if (err) return handleCallback(callback, err, null);

      if (cursor.s.cursorId && cursor.s.cursorId.isZero() && cursor._endSession) {
        cursor._endSession();
      }

      if (
        cursor.s.documents.length === 0 &&
        cursor.s.cursorId &&
        cursor.s.cursorId.isZero() &&
        !cursor.cmd.tailable &&
        !cursor.cmd.awaitData
      ) {
        return setCursorNotified(cursor, callback);
      }

      nextFunction(cursor, callback);
    });
  }

  if (cursor.s.documents.length === cursor.s.cursorIndex && Long.ZERO.equals(cursor.s.cursorId)) {
    setCursorDeadAndNotified(cursor, callback);
    return;
  }

  if (cursor.s.limit > 0 && cursor.s.currentLimit >= cursor.s.limit) {
    // Ensure we kill the cursor on the server
    cursor.kill();
    // Set cursor in dead and notified state
    setCursorDeadAndNotified(cursor, callback);
    return;
  }

  if (
    cursor.s.documents.length === cursor.s.cursorIndex &&
    cursor.cmd.tailable &&
    Long.ZERO.equals(cursor.s.cursorId)
  ) {
    return handleCallback(
      callback,
      new MongoError({
        message: 'No more documents in tailed cursor',
        tailable: cursor.cmd.tailable,
        awaitData: cursor.cmd.awaitData
      })
    );
  }

  if (cursor.s.cursorIndex === cursor.s.documents.length && !Long.ZERO.equals(cursor.s.cursorId)) {
    // Ensure an empty cursor state
    cursor.s.documents = [];
    cursor.s.cursorIndex = 0;

    // Check if connection is dead and return if not possible to
    if (isConnectionDead(cursor, callback)) return;

    // Execute the next get more
    return _getmore(cursor, function(err, doc, connection) {
      if (err) {
        if (err instanceof MongoError) {
          err[mongoErrorContextSymbol].isGetMore = true;
        }

        return handleCallback(callback, err);
      }

      if (cursor.s.cursorId && cursor.s.cursorId.isZero() && cursor._endSession) {
        cursor._endSession();
      }

      // Save the returned connection to ensure all getMore's fire over the same connection
      cursor.connection = connection;

      // Tailable cursor getMore result, notify owner about it
      // No attempt is made here to retry, this is left to the user of the
      // core module to handle to keep core simple
      if (
        cursor.s.documents.length === 0 &&
        cursor.cmd.tailable &&
        Long.ZERO.equals(cursor.s.cursorId)
      ) {
        // No more documents in the tailed cursor
        return handleCallback(
          callback,
          new MongoError({
            message: 'No more documents in tailed cursor',
            tailable: cursor.cmd.tailable,
            awaitData: cursor.cmd.awaitData
          })
        );
      } else if (
        cursor.s.documents.length === 0 &&
        cursor.cmd.tailable &&
        !Long.ZERO.equals(cursor.s.cursorId)
      ) {
        return nextFunction(cursor, callback);
      }

      if (cursor.s.limit > 0 && cursor.s.currentLimit >= cursor.s.limit) {
        return setCursorDeadAndNotified(cursor, callback);
      }

      nextFunction(cursor, callback);
    });
  }

  if (cursor.s.limit > 0 && cursor.s.currentLimit >= cursor.s.limit) {
    // Ensure we kill the cursor on the server
    cursor.kill();
    // Set cursor in dead and notified state
    return setCursorDeadAndNotified(cursor, callback);
  }

  // Increment the current cursor limit
  cursor.s.currentLimit += 1;

  // Get the document
  let doc = cursor.s.documents[cursor.s.cursorIndex++];

  // Doc overflow
  if (!doc || doc.$err) {
    // Ensure we kill the cursor on the server
    cursor.kill();
    // Set cursor in dead and notified state
    return setCursorDeadAndNotified(cursor, function() {
      handleCallback(callback, new MongoError(doc ? doc.$err : undefined));
    });
  }

  // Transform the doc with passed in transformation method if provided
  if (cursor.s.transforms && typeof cursor.s.transforms.doc === 'function') {
    doc = cursor.s.transforms.doc(doc);
  }

  // Return the document
  handleCallback(callback, null, doc);
}

module.exports = Cursor;