cursor.js 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769
  1. 'use strict';
  2. const Logger = require('./connection/logger');
  3. const retrieveBSON = require('./connection/utils').retrieveBSON;
  4. const MongoError = require('./error').MongoError;
  5. const MongoNetworkError = require('./error').MongoNetworkError;
  6. const mongoErrorContextSymbol = require('./error').mongoErrorContextSymbol;
  7. const f = require('util').format;
  8. const collationNotSupported = require('./utils').collationNotSupported;
  9. const wireProtocol = require('./wireprotocol');
  10. const BSON = retrieveBSON();
  11. const Long = BSON.Long;
  12. /**
  13. * This is a cursor results callback
  14. *
  15. * @callback resultCallback
  16. * @param {error} error An error object. Set to null if no error present
  17. * @param {object} document
  18. */
  19. /**
  20. * @fileOverview The **Cursor** class is an internal class that embodies a cursor on MongoDB
  21. * allowing for iteration over the results returned from the underlying query.
  22. *
  23. * **CURSORS Cannot directly be instantiated**
  24. * @example
  25. * var Server = require('mongodb-core').Server
  26. * , ReadPreference = require('mongodb-core').ReadPreference
  27. * , assert = require('assert');
  28. *
  29. * var server = new Server({host: 'localhost', port: 27017});
  30. * // Wait for the connection event
  31. * server.on('connect', function(server) {
  32. * assert.equal(null, err);
  33. *
  34. * // Execute the write
  35. * var cursor = _server.cursor('integration_tests.inserts_example4', {
  36. * find: 'integration_tests.example4'
  37. * , query: {a:1}
  38. * }, {
  39. * readPreference: new ReadPreference('secondary');
  40. * });
  41. *
  42. * // Get the first document
  43. * cursor.next(function(err, doc) {
  44. * assert.equal(null, err);
  45. * server.destroy();
  46. * });
  47. * });
  48. *
  49. * // Start connecting
  50. * server.connect();
  51. */
  52. /**
  53. * Creates a new Cursor, not to be used directly
  54. * @class
  55. * @param {object} bson An instance of the BSON parser
  56. * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
  57. * @param {{object}|Long} cmd The selector (can be a command or a cursorId)
  58. * @param {object} [options=null] Optional settings.
  59. * @param {object} [options.batchSize=1000] Batchsize for the operation
  60. * @param {array} [options.documents=[]] Initial documents list for cursor
  61. * @param {object} [options.transforms=null] Transform methods for the cursor results
  62. * @param {function} [options.transforms.query] Transform the value returned from the initial query
  63. * @param {function} [options.transforms.doc] Transform each document returned from Cursor.prototype.next
  64. * @param {object} topology The server topology instance.
  65. * @param {object} topologyOptions The server topology options.
  66. * @return {Cursor} A cursor instance
  67. * @property {number} cursorBatchSize The current cursorBatchSize for the cursor
  68. * @property {number} cursorLimit The current cursorLimit for the cursor
  69. * @property {number} cursorSkip The current cursorSkip for the cursor
  70. */
  71. var Cursor = function(bson, ns, cmd, options, topology, topologyOptions) {
  72. options = options || {};
  73. // Cursor pool
  74. this.pool = null;
  75. // Cursor server
  76. this.server = null;
  77. // Do we have a not connected handler
  78. this.disconnectHandler = options.disconnectHandler;
  79. // Set local values
  80. this.bson = bson;
  81. this.ns = ns;
  82. this.cmd = cmd;
  83. this.options = options;
  84. this.topology = topology;
  85. // All internal state
  86. this.cursorState = {
  87. cursorId: null,
  88. cmd: cmd,
  89. documents: options.documents || [],
  90. cursorIndex: 0,
  91. dead: false,
  92. killed: false,
  93. init: false,
  94. notified: false,
  95. limit: options.limit || cmd.limit || 0,
  96. skip: options.skip || cmd.skip || 0,
  97. batchSize: options.batchSize || cmd.batchSize || 1000,
  98. currentLimit: 0,
  99. // Result field name if not a cursor (contains the array of results)
  100. transforms: options.transforms,
  101. raw: options.raw || (cmd && cmd.raw)
  102. };
  103. if (typeof options.session === 'object') {
  104. this.cursorState.session = options.session;
  105. }
  106. // Add promoteLong to cursor state
  107. if (typeof topologyOptions.promoteLongs === 'boolean') {
  108. this.cursorState.promoteLongs = topologyOptions.promoteLongs;
  109. } else if (typeof options.promoteLongs === 'boolean') {
  110. this.cursorState.promoteLongs = options.promoteLongs;
  111. }
  112. // Add promoteValues to cursor state
  113. if (typeof topologyOptions.promoteValues === 'boolean') {
  114. this.cursorState.promoteValues = topologyOptions.promoteValues;
  115. } else if (typeof options.promoteValues === 'boolean') {
  116. this.cursorState.promoteValues = options.promoteValues;
  117. }
  118. // Add promoteBuffers to cursor state
  119. if (typeof topologyOptions.promoteBuffers === 'boolean') {
  120. this.cursorState.promoteBuffers = topologyOptions.promoteBuffers;
  121. } else if (typeof options.promoteBuffers === 'boolean') {
  122. this.cursorState.promoteBuffers = options.promoteBuffers;
  123. }
  124. if (topologyOptions.reconnect) {
  125. this.cursorState.reconnect = topologyOptions.reconnect;
  126. }
  127. // Logger
  128. this.logger = Logger('Cursor', topologyOptions);
  129. //
  130. // Did we pass in a cursor id
  131. if (typeof cmd === 'number') {
  132. this.cursorState.cursorId = Long.fromNumber(cmd);
  133. this.cursorState.lastCursorId = this.cursorState.cursorId;
  134. } else if (cmd instanceof Long) {
  135. this.cursorState.cursorId = cmd;
  136. this.cursorState.lastCursorId = cmd;
  137. }
  138. };
  139. Cursor.prototype.setCursorBatchSize = function(value) {
  140. this.cursorState.batchSize = value;
  141. };
  142. Cursor.prototype.cursorBatchSize = function() {
  143. return this.cursorState.batchSize;
  144. };
  145. Cursor.prototype.setCursorLimit = function(value) {
  146. this.cursorState.limit = value;
  147. };
  148. Cursor.prototype.cursorLimit = function() {
  149. return this.cursorState.limit;
  150. };
  151. Cursor.prototype.setCursorSkip = function(value) {
  152. this.cursorState.skip = value;
  153. };
  154. Cursor.prototype.cursorSkip = function() {
  155. return this.cursorState.skip;
  156. };
  157. Cursor.prototype._endSession = function(options, callback) {
  158. if (typeof options === 'function') {
  159. callback = options;
  160. options = {};
  161. }
  162. options = options || {};
  163. const session = this.cursorState.session;
  164. if (session && (options.force || session.owner === this)) {
  165. this.cursorState.session = undefined;
  166. session.endSession(callback);
  167. return true;
  168. }
  169. if (callback) {
  170. callback();
  171. }
  172. return false;
  173. };
  174. //
  175. // Handle callback (including any exceptions thrown)
  176. var handleCallback = function(callback, err, result) {
  177. try {
  178. callback(err, result);
  179. } catch (err) {
  180. process.nextTick(function() {
  181. throw err;
  182. });
  183. }
  184. };
  185. // Internal methods
  186. Cursor.prototype._getmore = function(callback) {
  187. if (this.logger.isDebug())
  188. this.logger.debug(f('schedule getMore call for query [%s]', JSON.stringify(this.query)));
  189. // Set the current batchSize
  190. var batchSize = this.cursorState.batchSize;
  191. if (
  192. this.cursorState.limit > 0 &&
  193. this.cursorState.currentLimit + batchSize > this.cursorState.limit
  194. ) {
  195. batchSize = this.cursorState.limit - this.cursorState.currentLimit;
  196. }
  197. wireProtocol.getMore(this.server, this.ns, this.cursorState, batchSize, this.options, callback);
  198. };
  199. /**
  200. * Clone the cursor
  201. * @method
  202. * @return {Cursor}
  203. */
  204. Cursor.prototype.clone = function() {
  205. return this.topology.cursor(this.ns, this.cmd, this.options);
  206. };
  207. /**
  208. * Checks if the cursor is dead
  209. * @method
  210. * @return {boolean} A boolean signifying if the cursor is dead or not
  211. */
  212. Cursor.prototype.isDead = function() {
  213. return this.cursorState.dead === true;
  214. };
  215. /**
  216. * Checks if the cursor was killed by the application
  217. * @method
  218. * @return {boolean} A boolean signifying if the cursor was killed by the application
  219. */
  220. Cursor.prototype.isKilled = function() {
  221. return this.cursorState.killed === true;
  222. };
  223. /**
  224. * Checks if the cursor notified it's caller about it's death
  225. * @method
  226. * @return {boolean} A boolean signifying if the cursor notified the callback
  227. */
  228. Cursor.prototype.isNotified = function() {
  229. return this.cursorState.notified === true;
  230. };
  231. /**
  232. * Returns current buffered documents length
  233. * @method
  234. * @return {number} The number of items in the buffered documents
  235. */
  236. Cursor.prototype.bufferedCount = function() {
  237. return this.cursorState.documents.length - this.cursorState.cursorIndex;
  238. };
  239. /**
  240. * Returns current buffered documents
  241. * @method
  242. * @return {Array} An array of buffered documents
  243. */
  244. Cursor.prototype.readBufferedDocuments = function(number) {
  245. var unreadDocumentsLength = this.cursorState.documents.length - this.cursorState.cursorIndex;
  246. var length = number < unreadDocumentsLength ? number : unreadDocumentsLength;
  247. var elements = this.cursorState.documents.slice(
  248. this.cursorState.cursorIndex,
  249. this.cursorState.cursorIndex + length
  250. );
  251. // Transform the doc with passed in transformation method if provided
  252. if (this.cursorState.transforms && typeof this.cursorState.transforms.doc === 'function') {
  253. // Transform all the elements
  254. for (var i = 0; i < elements.length; i++) {
  255. elements[i] = this.cursorState.transforms.doc(elements[i]);
  256. }
  257. }
  258. // Ensure we do not return any more documents than the limit imposed
  259. // Just return the number of elements up to the limit
  260. if (
  261. this.cursorState.limit > 0 &&
  262. this.cursorState.currentLimit + elements.length > this.cursorState.limit
  263. ) {
  264. elements = elements.slice(0, this.cursorState.limit - this.cursorState.currentLimit);
  265. this.kill();
  266. }
  267. // Adjust current limit
  268. this.cursorState.currentLimit = this.cursorState.currentLimit + elements.length;
  269. this.cursorState.cursorIndex = this.cursorState.cursorIndex + elements.length;
  270. // Return elements
  271. return elements;
  272. };
  273. /**
  274. * Kill the cursor
  275. * @method
  276. * @param {resultCallback} callback A callback function
  277. */
  278. Cursor.prototype.kill = function(callback) {
  279. // Set cursor to dead
  280. this.cursorState.dead = true;
  281. this.cursorState.killed = true;
  282. // Remove documents
  283. this.cursorState.documents = [];
  284. // If no cursor id just return
  285. if (
  286. this.cursorState.cursorId == null ||
  287. this.cursorState.cursorId.isZero() ||
  288. this.cursorState.init === false
  289. ) {
  290. if (callback) callback(null, null);
  291. return;
  292. }
  293. wireProtocol.killCursors(this.server, this.ns, this.cursorState, callback);
  294. };
  295. /**
  296. * Resets the cursor
  297. * @method
  298. * @return {null}
  299. */
  300. Cursor.prototype.rewind = function() {
  301. if (this.cursorState.init) {
  302. if (!this.cursorState.dead) {
  303. this.kill();
  304. }
  305. this.cursorState.currentLimit = 0;
  306. this.cursorState.init = false;
  307. this.cursorState.dead = false;
  308. this.cursorState.killed = false;
  309. this.cursorState.notified = false;
  310. this.cursorState.documents = [];
  311. this.cursorState.cursorId = null;
  312. this.cursorState.cursorIndex = 0;
  313. }
  314. };
  315. /**
  316. * Validate if the pool is dead and return error
  317. */
  318. var isConnectionDead = function(self, callback) {
  319. if (self.pool && self.pool.isDestroyed()) {
  320. self.cursorState.killed = true;
  321. const err = new MongoNetworkError(
  322. f('connection to host %s:%s was destroyed', self.pool.host, self.pool.port)
  323. );
  324. _setCursorNotifiedImpl(self, () => callback(err));
  325. return true;
  326. }
  327. return false;
  328. };
  329. /**
  330. * Validate if the cursor is dead but was not explicitly killed by user
  331. */
  332. var isCursorDeadButNotkilled = function(self, callback) {
  333. // Cursor is dead but not marked killed, return null
  334. if (self.cursorState.dead && !self.cursorState.killed) {
  335. self.cursorState.killed = true;
  336. setCursorNotified(self, callback);
  337. return true;
  338. }
  339. return false;
  340. };
  341. /**
  342. * Validate if the cursor is dead and was killed by user
  343. */
  344. var isCursorDeadAndKilled = function(self, callback) {
  345. if (self.cursorState.dead && self.cursorState.killed) {
  346. handleCallback(callback, new MongoError('cursor is dead'));
  347. return true;
  348. }
  349. return false;
  350. };
  351. /**
  352. * Validate if the cursor was killed by the user
  353. */
  354. var isCursorKilled = function(self, callback) {
  355. if (self.cursorState.killed) {
  356. setCursorNotified(self, callback);
  357. return true;
  358. }
  359. return false;
  360. };
  361. /**
  362. * Mark cursor as being dead and notified
  363. */
  364. var setCursorDeadAndNotified = function(self, callback) {
  365. self.cursorState.dead = true;
  366. setCursorNotified(self, callback);
  367. };
  368. /**
  369. * Mark cursor as being notified
  370. */
  371. var setCursorNotified = function(self, callback) {
  372. _setCursorNotifiedImpl(self, () => handleCallback(callback, null, null));
  373. };
  374. var _setCursorNotifiedImpl = function(self, callback) {
  375. self.cursorState.notified = true;
  376. self.cursorState.documents = [];
  377. self.cursorState.cursorIndex = 0;
  378. if (self._endSession) {
  379. return self._endSession(undefined, () => callback());
  380. }
  381. return callback();
  382. };
  383. var nextFunction = function(self, callback) {
  384. // We have notified about it
  385. if (self.cursorState.notified) {
  386. return callback(new Error('cursor is exhausted'));
  387. }
  388. // Cursor is killed return null
  389. if (isCursorKilled(self, callback)) return;
  390. // Cursor is dead but not marked killed, return null
  391. if (isCursorDeadButNotkilled(self, callback)) return;
  392. // We have a dead and killed cursor, attempting to call next should error
  393. if (isCursorDeadAndKilled(self, callback)) return;
  394. // We have just started the cursor
  395. if (!self.cursorState.init) {
  396. return initializeCursor(self, callback);
  397. }
  398. if (self.cursorState.limit > 0 && self.cursorState.currentLimit >= self.cursorState.limit) {
  399. // Ensure we kill the cursor on the server
  400. self.kill();
  401. // Set cursor in dead and notified state
  402. return setCursorDeadAndNotified(self, callback);
  403. } else if (
  404. self.cursorState.cursorIndex === self.cursorState.documents.length &&
  405. !Long.ZERO.equals(self.cursorState.cursorId)
  406. ) {
  407. // Ensure an empty cursor state
  408. self.cursorState.documents = [];
  409. self.cursorState.cursorIndex = 0;
  410. // Check if topology is destroyed
  411. if (self.topology.isDestroyed())
  412. return callback(
  413. new MongoNetworkError('connection destroyed, not possible to instantiate cursor')
  414. );
  415. // Check if connection is dead and return if not possible to
  416. // execute a getmore on this connection
  417. if (isConnectionDead(self, callback)) return;
  418. // Execute the next get more
  419. self._getmore(function(err, doc, connection) {
  420. if (err) {
  421. if (err instanceof MongoError) {
  422. err[mongoErrorContextSymbol].isGetMore = true;
  423. }
  424. return handleCallback(callback, err);
  425. }
  426. if (self.cursorState.cursorId && self.cursorState.cursorId.isZero() && self._endSession) {
  427. self._endSession();
  428. }
  429. // Save the returned connection to ensure all getMore's fire over the same connection
  430. self.connection = connection;
  431. // Tailable cursor getMore result, notify owner about it
  432. // No attempt is made here to retry, this is left to the user of the
  433. // core module to handle to keep core simple
  434. if (
  435. self.cursorState.documents.length === 0 &&
  436. self.cmd.tailable &&
  437. Long.ZERO.equals(self.cursorState.cursorId)
  438. ) {
  439. // No more documents in the tailed cursor
  440. return handleCallback(
  441. callback,
  442. new MongoError({
  443. message: 'No more documents in tailed cursor',
  444. tailable: self.cmd.tailable,
  445. awaitData: self.cmd.awaitData
  446. })
  447. );
  448. } else if (
  449. self.cursorState.documents.length === 0 &&
  450. self.cmd.tailable &&
  451. !Long.ZERO.equals(self.cursorState.cursorId)
  452. ) {
  453. return nextFunction(self, callback);
  454. }
  455. if (self.cursorState.limit > 0 && self.cursorState.currentLimit >= self.cursorState.limit) {
  456. return setCursorDeadAndNotified(self, callback);
  457. }
  458. nextFunction(self, callback);
  459. });
  460. } else if (
  461. self.cursorState.documents.length === self.cursorState.cursorIndex &&
  462. self.cmd.tailable &&
  463. Long.ZERO.equals(self.cursorState.cursorId)
  464. ) {
  465. return handleCallback(
  466. callback,
  467. new MongoError({
  468. message: 'No more documents in tailed cursor',
  469. tailable: self.cmd.tailable,
  470. awaitData: self.cmd.awaitData
  471. })
  472. );
  473. } else if (
  474. self.cursorState.documents.length === self.cursorState.cursorIndex &&
  475. Long.ZERO.equals(self.cursorState.cursorId)
  476. ) {
  477. setCursorDeadAndNotified(self, callback);
  478. } else {
  479. if (self.cursorState.limit > 0 && self.cursorState.currentLimit >= self.cursorState.limit) {
  480. // Ensure we kill the cursor on the server
  481. self.kill();
  482. // Set cursor in dead and notified state
  483. return setCursorDeadAndNotified(self, callback);
  484. }
  485. // Increment the current cursor limit
  486. self.cursorState.currentLimit += 1;
  487. // Get the document
  488. var doc = self.cursorState.documents[self.cursorState.cursorIndex++];
  489. // Doc overflow
  490. if (!doc || doc.$err) {
  491. // Ensure we kill the cursor on the server
  492. self.kill();
  493. // Set cursor in dead and notified state
  494. return setCursorDeadAndNotified(self, function() {
  495. handleCallback(callback, new MongoError(doc ? doc.$err : undefined));
  496. });
  497. }
  498. // Transform the doc with passed in transformation method if provided
  499. if (self.cursorState.transforms && typeof self.cursorState.transforms.doc === 'function') {
  500. doc = self.cursorState.transforms.doc(doc);
  501. }
  502. // Return the document
  503. handleCallback(callback, null, doc);
  504. }
  505. };
  506. function initializeCursor(cursor, callback) {
  507. // Topology is not connected, save the call in the provided store to be
  508. // Executed at some point when the handler deems it's reconnected
  509. if (!cursor.topology.isConnected(cursor.options)) {
  510. // Only need this for single server, because repl sets and mongos
  511. // will always continue trying to reconnect
  512. if (cursor.topology._type === 'server' && !cursor.topology.s.options.reconnect) {
  513. // Reconnect is disabled, so we'll never reconnect
  514. return callback(new MongoError('no connection available'));
  515. }
  516. if (cursor.disconnectHandler != null) {
  517. if (cursor.topology.isDestroyed()) {
  518. // Topology was destroyed, so don't try to wait for it to reconnect
  519. return callback(new MongoError('Topology was destroyed'));
  520. }
  521. return cursor.disconnectHandler.addObjectAndMethod(
  522. 'cursor',
  523. cursor,
  524. 'next',
  525. [callback],
  526. callback
  527. );
  528. }
  529. }
  530. // Very explicitly choose what is passed to selectServer
  531. const serverSelectOptions = {};
  532. if (cursor.cursorState.session) {
  533. serverSelectOptions.session = cursor.cursorState.session;
  534. }
  535. if (cursor.options.readPreference) {
  536. serverSelectOptions.readPreference = cursor.options.readPreference;
  537. }
  538. return cursor.topology.selectServer(serverSelectOptions, (err, server) => {
  539. if (err) {
  540. const disconnectHandler = cursor.disconnectHandler;
  541. if (disconnectHandler != null) {
  542. return disconnectHandler.addObjectAndMethod('cursor', cursor, 'next', [callback], callback);
  543. }
  544. return callback(err);
  545. }
  546. cursor.server = server;
  547. cursor.cursorState.init = true;
  548. if (collationNotSupported(cursor.server, cursor.cmd)) {
  549. return callback(new MongoError(`server ${cursor.server.name} does not support collation`));
  550. }
  551. function done() {
  552. if (
  553. cursor.cursorState.cursorId &&
  554. cursor.cursorState.cursorId.isZero() &&
  555. cursor._endSession
  556. ) {
  557. cursor._endSession();
  558. }
  559. if (
  560. cursor.cursorState.documents.length === 0 &&
  561. cursor.cursorState.cursorId &&
  562. cursor.cursorState.cursorId.isZero() &&
  563. !cursor.cmd.tailable &&
  564. !cursor.cmd.awaitData
  565. ) {
  566. return setCursorNotified(cursor, callback);
  567. }
  568. nextFunction(cursor, callback);
  569. }
  570. // NOTE: this is a special internal method for cloning a cursor, consider removing
  571. if (cursor.cursorState.cursorId != null) {
  572. return done();
  573. }
  574. const queryCallback = (err, r) => {
  575. if (err) return callback(err);
  576. const result = r.message;
  577. if (result.queryFailure) {
  578. return callback(new MongoError(result.documents[0]), null);
  579. }
  580. // Check if we have a command cursor
  581. if (
  582. Array.isArray(result.documents) &&
  583. result.documents.length === 1 &&
  584. (!cursor.cmd.find || (cursor.cmd.find && cursor.cmd.virtual === false)) &&
  585. (typeof result.documents[0].cursor !== 'string' ||
  586. result.documents[0]['$err'] ||
  587. result.documents[0]['errmsg'] ||
  588. Array.isArray(result.documents[0].result))
  589. ) {
  590. // We have an error document, return the error
  591. if (result.documents[0]['$err'] || result.documents[0]['errmsg']) {
  592. return callback(new MongoError(result.documents[0]), null);
  593. }
  594. // We have a cursor document
  595. if (result.documents[0].cursor != null && typeof result.documents[0].cursor !== 'string') {
  596. var id = result.documents[0].cursor.id;
  597. // If we have a namespace change set the new namespace for getmores
  598. if (result.documents[0].cursor.ns) {
  599. cursor.ns = result.documents[0].cursor.ns;
  600. }
  601. // Promote id to long if needed
  602. cursor.cursorState.cursorId = typeof id === 'number' ? Long.fromNumber(id) : id;
  603. cursor.cursorState.lastCursorId = cursor.cursorState.cursorId;
  604. cursor.cursorState.operationTime = result.documents[0].operationTime;
  605. // If we have a firstBatch set it
  606. if (Array.isArray(result.documents[0].cursor.firstBatch)) {
  607. cursor.cursorState.documents = result.documents[0].cursor.firstBatch; //.reverse();
  608. }
  609. // Return after processing command cursor
  610. return done(result);
  611. }
  612. if (Array.isArray(result.documents[0].result)) {
  613. cursor.cursorState.documents = result.documents[0].result;
  614. cursor.cursorState.cursorId = Long.ZERO;
  615. return done(result);
  616. }
  617. }
  618. // Otherwise fall back to regular find path
  619. const cursorId = result.cursorId || 0;
  620. cursor.cursorState.cursorId = Long.fromNumber(cursorId);
  621. cursor.cursorState.documents = result.documents;
  622. cursor.cursorState.lastCursorId = result.cursorId;
  623. // Transform the results with passed in transformation method if provided
  624. if (
  625. cursor.cursorState.transforms &&
  626. typeof cursor.cursorState.transforms.query === 'function'
  627. ) {
  628. cursor.cursorState.documents = cursor.cursorState.transforms.query(result);
  629. }
  630. // Return callback
  631. done(result);
  632. };
  633. if (cursor.logger.isDebug()) {
  634. cursor.logger.debug(
  635. `issue initial query [${JSON.stringify(cursor.cmd)}] with flags [${JSON.stringify(
  636. cursor.query
  637. )}]`
  638. );
  639. }
  640. if (cursor.cmd.find != null) {
  641. wireProtocol.query(
  642. cursor.server,
  643. cursor.ns,
  644. cursor.cmd,
  645. cursor.cursorState,
  646. cursor.options,
  647. queryCallback
  648. );
  649. return;
  650. }
  651. cursor.query = wireProtocol.command(
  652. cursor.server,
  653. cursor.ns,
  654. cursor.cmd,
  655. cursor.options,
  656. queryCallback
  657. );
  658. });
  659. }
  660. /**
  661. * Retrieve the next document from the cursor
  662. * @method
  663. * @param {resultCallback} callback A callback function
  664. */
  665. Cursor.prototype.next = function(callback) {
  666. nextFunction(this, callback);
  667. };
  668. module.exports = Cursor;