command_cursor.js 9.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311
  1. "use strict";
  2. var inherits = require('util').inherits
  3. , ReadPreference = require('./read_preference')
  4. , MongoError = require('mongodb-core').MongoError
  5. , Readable = require('stream').Readable || require('readable-stream').Readable
  6. , Define = require('./metadata')
  7. , CoreCursor = require('./cursor')
  8. , CoreReadPreference = require('mongodb-core').ReadPreference;
  9. /**
  10. * @fileOverview The **CommandCursor** class is an internal class that embodies a
  11. * generalized cursor based on a MongoDB command allowing for iteration over the
  12. * results returned. It supports one by one document iteration, conversion to an
  13. * array or can be iterated as a Node 0.10.X or higher stream
  14. *
  15. * **CommandCursor Cannot directly be instantiated**
  16. * @example
  17. * var MongoClient = require('mongodb').MongoClient,
  18. * test = require('assert');
  19. * // Connection url
  20. * var url = 'mongodb://localhost:27017/test';
  21. * // Connect using MongoClient
  22. * MongoClient.connect(url, function(err, db) {
  23. * // Create a collection we want to drop later
  24. * var col = db.collection('listCollectionsExample1');
  25. * // Insert a bunch of documents
  26. * col.insert([{a:1, b:1}
  27. * , {a:2, b:2}, {a:3, b:3}
  28. * , {a:4, b:4}], {w:1}, function(err, result) {
  29. * test.equal(null, err);
  30. *
  31. * // List the database collections available
  32. * db.listCollections().toArray(function(err, items) {
  33. * test.equal(null, err);
  34. * db.close();
  35. * });
  36. * });
  37. * });
  38. */
  39. /**
  40. * Namespace provided by the browser.
  41. * @external Readable
  42. */
  43. /**
  44. * Creates a new Command Cursor instance (INTERNAL TYPE, do not instantiate directly)
  45. * @class CommandCursor
  46. * @extends external:Readable
  47. * @fires CommandCursor#data
  48. * @fires CommandCursor#end
  49. * @fires CommandCursor#close
  50. * @fires CommandCursor#readable
  51. * @return {CommandCursor} an CommandCursor instance.
  52. */
  53. var CommandCursor = function(bson, ns, cmd, options, topology, topologyOptions) {
  54. CoreCursor.apply(this, Array.prototype.slice.call(arguments, 0));
  55. var state = CommandCursor.INIT;
  56. var streamOptions = {};
  57. // MaxTimeMS
  58. var maxTimeMS = null;
  59. // Get the promiseLibrary
  60. var promiseLibrary = options.promiseLibrary;
  61. // No promise library selected fall back
  62. if(!promiseLibrary) {
  63. promiseLibrary = typeof global.Promise == 'function' ?
  64. global.Promise : require('es6-promise').Promise;
  65. }
  66. // Set up
  67. Readable.call(this, {objectMode: true});
  68. // Internal state
  69. this.s = {
  70. // MaxTimeMS
  71. maxTimeMS: maxTimeMS
  72. // State
  73. , state: state
  74. // Stream options
  75. , streamOptions: streamOptions
  76. // BSON
  77. , bson: bson
  78. // Namespae
  79. , ns: ns
  80. // Command
  81. , cmd: cmd
  82. // Options
  83. , options: options
  84. // Topology
  85. , topology: topology
  86. // Topology Options
  87. , topologyOptions: topologyOptions
  88. // Promise library
  89. , promiseLibrary: promiseLibrary
  90. }
  91. }
  92. /**
  93. * CommandCursor stream data event, fired for each document in the cursor.
  94. *
  95. * @event CommandCursor#data
  96. * @type {object}
  97. */
  98. /**
  99. * CommandCursor stream end event
  100. *
  101. * @event CommandCursor#end
  102. * @type {null}
  103. */
  104. /**
  105. * CommandCursor stream close event
  106. *
  107. * @event CommandCursor#close
  108. * @type {null}
  109. */
  110. /**
  111. * CommandCursor stream readable event
  112. *
  113. * @event CommandCursor#readable
  114. * @type {null}
  115. */
  116. // Inherit from Readable
  117. inherits(CommandCursor, Readable);
  118. // Set the methods to inherit from prototype
  119. var methodsToInherit = ['_next', 'next', 'each', 'forEach', 'toArray'
  120. , 'rewind', 'bufferedCount', 'readBufferedDocuments', 'close', 'isClosed', 'kill'
  121. , '_find', '_getmore', '_killcursor', 'isDead', 'explain', 'isNotified', 'isKilled'];
  122. // Only inherit the types we need
  123. for(var i = 0; i < methodsToInherit.length; i++) {
  124. CommandCursor.prototype[methodsToInherit[i]] = CoreCursor.prototype[methodsToInherit[i]];
  125. }
  126. var define = CommandCursor.define = new Define('CommandCursor', CommandCursor, true);
  127. /**
  128. * Set the ReadPreference for the cursor.
  129. * @method
  130. * @param {(string|ReadPreference)} readPreference The new read preference for the cursor.
  131. * @throws {MongoError}
  132. * @return {Cursor}
  133. */
  134. CommandCursor.prototype.setReadPreference = function(r) {
  135. if(this.s.state == CommandCursor.CLOSED || this.isDead()) throw MongoError.create({message: "Cursor is closed", driver:true});
  136. if(this.s.state != CommandCursor.INIT) throw MongoError.create({message: 'cannot change cursor readPreference after cursor has been accessed', driver:true});
  137. if(r instanceof ReadPreference) {
  138. this.s.options.readPreference = new CoreReadPreference(r.mode, r.tags, {maxStalenessSeconds: r.maxStalenessSeconds});
  139. } else if(typeof r == 'string') {
  140. this.s.options.readPreference = new CoreReadPreference(r);
  141. } else if(r instanceof CoreReadPreference) {
  142. this.s.options.readPreference = r;
  143. }
  144. return this;
  145. }
  146. define.classMethod('setReadPreference', {callback: false, promise:false, returns: [CommandCursor]});
  147. /**
  148. * Set the batch size for the cursor.
  149. * @method
  150. * @param {number} value The batchSize for the cursor.
  151. * @throws {MongoError}
  152. * @return {CommandCursor}
  153. */
  154. CommandCursor.prototype.batchSize = function(value) {
  155. if(this.s.state == CommandCursor.CLOSED || this.isDead()) throw MongoError.create({message: "Cursor is closed", driver:true});
  156. if(typeof value != 'number') throw MongoError.create({message: "batchSize requires an integer", driver:true});
  157. if(this.s.cmd.cursor) this.s.cmd.cursor.batchSize = value;
  158. this.setCursorBatchSize(value);
  159. return this;
  160. }
  161. define.classMethod('batchSize', {callback: false, promise:false, returns: [CommandCursor]});
  162. /**
  163. * Add a maxTimeMS stage to the aggregation pipeline
  164. * @method
  165. * @param {number} value The state maxTimeMS value.
  166. * @return {CommandCursor}
  167. */
  168. CommandCursor.prototype.maxTimeMS = function(value) {
  169. if(this.s.topology.lastIsMaster().minWireVersion > 2) {
  170. this.s.cmd.maxTimeMS = value;
  171. }
  172. return this;
  173. }
  174. define.classMethod('maxTimeMS', {callback: false, promise:false, returns: [CommandCursor]});
  175. CommandCursor.prototype.get = CommandCursor.prototype.toArray;
  176. define.classMethod('get', {callback: true, promise:false});
  177. // Inherited methods
  178. define.classMethod('toArray', {callback: true, promise:true});
  179. define.classMethod('each', {callback: true, promise:false});
  180. define.classMethod('forEach', {callback: true, promise:false});
  181. define.classMethod('next', {callback: true, promise:true});
  182. define.classMethod('close', {callback: true, promise:true});
  183. define.classMethod('isClosed', {callback: false, promise:false, returns: [Boolean]});
  184. define.classMethod('rewind', {callback: false, promise:false});
  185. define.classMethod('bufferedCount', {callback: false, promise:false, returns: [Number]});
  186. define.classMethod('readBufferedDocuments', {callback: false, promise:false, returns: [Array]});
  187. /**
  188. * Get the next available document from the cursor, returns null if no more documents are available.
  189. * @function CommandCursor.prototype.next
  190. * @param {CommandCursor~resultCallback} [callback] The result callback.
  191. * @throws {MongoError}
  192. * @return {Promise} returns Promise if no callback passed
  193. */
  194. /**
  195. * The callback format for results
  196. * @callback CommandCursor~toArrayResultCallback
  197. * @param {MongoError} error An error instance representing the error during the execution.
  198. * @param {object[]} documents All the documents the satisfy the cursor.
  199. */
  200. /**
  201. * Returns an array of documents. The caller is responsible for making sure that there
  202. * is enough memory to store the results. Note that the array only contain partial
  203. * results when this cursor had been previouly accessed.
  204. * @method CommandCursor.prototype.toArray
  205. * @param {CommandCursor~toArrayResultCallback} [callback] The result callback.
  206. * @throws {MongoError}
  207. * @return {Promise} returns Promise if no callback passed
  208. */
  209. /**
  210. * The callback format for results
  211. * @callback CommandCursor~resultCallback
  212. * @param {MongoError} error An error instance representing the error during the execution.
  213. * @param {(object|null)} result The result object if the command was executed successfully.
  214. */
  215. /**
  216. * Iterates over all the documents for this cursor. As with **{cursor.toArray}**,
  217. * not all of the elements will be iterated if this cursor had been previouly accessed.
  218. * In that case, **{cursor.rewind}** can be used to reset the cursor. However, unlike
  219. * **{cursor.toArray}**, the cursor will only hold a maximum of batch size elements
  220. * at any given time if batch size is specified. Otherwise, the caller is responsible
  221. * for making sure that the entire result can fit the memory.
  222. * @method CommandCursor.prototype.each
  223. * @param {CommandCursor~resultCallback} callback The result callback.
  224. * @throws {MongoError}
  225. * @return {null}
  226. */
  227. /**
  228. * Close the cursor, sending a KillCursor command and emitting close.
  229. * @method CommandCursor.prototype.close
  230. * @param {CommandCursor~resultCallback} [callback] The result callback.
  231. * @return {Promise} returns Promise if no callback passed
  232. */
  233. /**
  234. * Is the cursor closed
  235. * @method CommandCursor.prototype.isClosed
  236. * @return {boolean}
  237. */
  238. /**
  239. * Clone the cursor
  240. * @function CommandCursor.prototype.clone
  241. * @return {CommandCursor}
  242. */
  243. /**
  244. * Resets the cursor
  245. * @function CommandCursor.prototype.rewind
  246. * @return {CommandCursor}
  247. */
  248. /**
  249. * The callback format for the forEach iterator method
  250. * @callback CommandCursor~iteratorCallback
  251. * @param {Object} doc An emitted document for the iterator
  252. */
  253. /**
  254. * The callback error format for the forEach iterator method
  255. * @callback CommandCursor~endCallback
  256. * @param {MongoError} error An error instance representing the error during the execution.
  257. */
  258. /*
  259. * Iterates over all the documents for this cursor using the iterator, callback pattern.
  260. * @method CommandCursor.prototype.forEach
  261. * @param {CommandCursor~iteratorCallback} iterator The iteration callback.
  262. * @param {CommandCursor~endCallback} callback The end callback.
  263. * @throws {MongoError}
  264. * @return {null}
  265. */
  266. CommandCursor.INIT = 0;
  267. CommandCursor.OPEN = 1;
  268. CommandCursor.CLOSED = 2;
  269. module.exports = CommandCursor;