'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;