1/*!
  2 * Module dependencies.
  3 */
  4
  5var PromiseProvider = require('./promise_provider');
  6var Readable = require('stream').Readable;
  7var helpers = require('./queryhelpers');
  8var util = require('util');
  9
 10/**
 11 * A QueryCursor is a concurrency primitive for processing query results
 12 * one document at a time. A QueryCursor fulfills the [Node.js streams3 API](https://strongloop.com/strongblog/whats-new-io-js-beta-streams3/),
 13 * in addition to several other mechanisms for loading documents from MongoDB
 14 * one at a time.
 15 *
 16 * Unless you're an advanced user, do **not** instantiate this class directly.
 17 * Use [`Query#cursor()`](/api.html#query_Query-cursor) instead.
 18 *
 19 * @param {Query} query
 20 * @param {Object} options query options passed to `.find()`
 21 * @inherits Readable
 22 * @event `cursor`: Emitted when the cursor is created
 23 * @event `error`: Emitted when an error occurred
 24 * @event `data`: Emitted when the stream is flowing and the next doc is ready
 25 * @event `end`: Emitted when the stream is exhausted
 26 * @api public
 27 */
 28
 29function QueryCursor(query, options) {
 30  Readable.call(this, { objectMode: true });
 31
 32  this.cursor = null;
 33  this.query = query;
 34  var _this = this;
 35  var model = query.model;
 36  model.collection.find(query._conditions, options, function(err, cursor) {
 37    if (_this._error) {
 38      cursor.close(function() {});
 39      _this.listeners('error').length > 0 && _this.emit('error', _this._error);
 40    }
 41    if (err) {
 42      return _this.emit('error', err);
 43    }
 44    _this.cursor = cursor;
 45    _this.emit('cursor', cursor);
 46  });
 47}
 48
 49util.inherits(QueryCursor, Readable);
 50
 51/*!
 52 * Necessary to satisfy the Readable API
 53 */
 54
 55QueryCursor.prototype._read = function() {
 56  var _this = this;
 57  _next(this, function(error, doc) {
 58    if (error) {
 59      return _this.emit('error', error);
 60    }
 61    if (!doc) {
 62      _this.push(null);
 63      return _this.cursor.close(function(error) {
 64        if (error) {
 65          return _this.emit('error', error);
 66        }
 67        _this.emit('close');
 68      });
 69    }
 70    _this.push(doc);
 71  });
 72};
 73
 74/*!
 75 * Marks this cursor as errored
 76 */
 77
 78QueryCursor.prototype._markError = function(error) {
 79  this._error = error;
 80  return this;
 81};
 82
 83/**
 84 * Marks this cursor as closed. Will stop streaming and subsequent calls to
 85 * `next()` will error.
 86 *
 87 * @param {Function} callback
 88 * @return {Promise}
 89 * @api public
 90 * @method close
 91 * @emits close
 92 * @see MongoDB driver cursor#close http://mongodb.github.io/node-mongodb-native/2.1/api/Cursor.html#close
 93 */
 94
 95QueryCursor.prototype.close = function(callback) {
 96  var Promise = PromiseProvider.get();
 97  var _this = this;
 98  return new Promise.ES6(function(resolve, reject) {
 99    _this.cursor.close(function(error) {
100      if (error) {
101        callback && callback(error);
102        reject(error);
103        return _this.listeners('error').length > 0 &&
104          _this.emit('error', error);
105      }
106      _this.emit('close');
107      resolve();
108      callback && callback();
109    });
110  });
111};
112
113/**
114 * Get the next document from this cursor. Will return `null` when there are
115 * no documents left.
116 *
117 * @param {Function} callback
118 * @return {Promise}
119 * @api public
120 * @method next
121 */
122
123QueryCursor.prototype.next = function(callback) {
124  var Promise = PromiseProvider.get();
125  var _this = this;
126  return new Promise.ES6(function(resolve, reject) {
127    _next(_this, function(error, doc) {
128      if (error) {
129        callback && callback(error);
130        return reject(error);
131      }
132      callback && callback(null, doc);
133      resolve(doc);
134    });
135  });
136};
137
138/**
139 * Execute `fn` for every document in the cursor. If `fn` returns a promise,
140 * will wait for the promise to resolve before iterating on to the next one.
141 * Returns a promise that resolves when done.
142 *
143 * @param {Function} fn
144 * @param {Function} [callback] executed when all docs have been processed
145 * @return {Promise}
146 * @api public
147 * @method eachAsync
148 */
149
150QueryCursor.prototype.eachAsync = function(fn, callback) {
151  var Promise = PromiseProvider.get();
152  var _this = this;
153
154  var handleNextResult = function(doc, callback) {
155    var promise = fn(doc);
156    if (promise && typeof promise.then === 'function') {
157      promise.then(
158        function() { callback(null); },
159        function(error) { callback(error); });
160    } else {
161      callback(null);
162    }
163  };
164
165  var iterate = function(callback) {
166    return _next(_this, function(error, doc) {
167      if (error) {
168        return callback(error);
169      }
170      if (!doc) {
171        return callback(null);
172      }
173      handleNextResult(doc, function(error) {
174        if (error) {
175          return callback(error);
176        }
177        iterate(callback);
178      });
179    });
180  };
181
182  return new Promise.ES6(function(resolve, reject) {
183    iterate(function(error) {
184      if (error) {
185        callback && callback(error);
186        return reject(error);
187      }
188      callback && callback(null);
189      return resolve();
190    });
191  });
192};
193
194/*!
195 * Get the next doc from the underlying cursor and mongooseify it
196 * (populate, etc.)
197 */
198
199function _next(ctx, callback) {
200  if (ctx._error) {
201    return process.nextTick(function() {
202      callback(ctx._error);
203    });
204  }
205
206  if (ctx.cursor) {
207    ctx.cursor.next(function(error, doc) {
208      if (error) {
209        return callback(error);
210      }
211      if (!doc) {
212        return callback(null, null);
213      }
214
215      var opts = ctx.query._mongooseOptions;
216      if (!opts.populate) {
217        return opts.lean === true ?
218                callback(null, doc) :
219                _create(ctx, doc, null, callback);
220      }
221
222      var pop = helpers.preparePopulationOptionsMQ(ctx.query,
223        ctx.query._mongooseOptions);
224      pop.forEach(function(option) {
225        delete option.model;
226      });
227      pop.__noPromise = true;
228      ctx.query.model.populate(doc, pop, function(err, doc) {
229        if (err) {
230          return callback(err);
231        }
232        return opts.lean === true ?
233          callback(null, doc) :
234          _create(ctx, doc, pop, callback);
235      });
236    });
237  } else {
238    ctx.once('cursor', function() {
239      _next(ctx, callback);
240    });
241  }
242}
243
244/*!
245 * Convert a raw doc into a full mongoose doc.
246 */
247
248function _create(ctx, doc, populatedIds, cb) {
249  var instance = helpers.createModel(ctx.query.model, doc, ctx.query._fields);
250  var opts = populatedIds ?
251    { populated: populatedIds } :
252    undefined;
253
254  instance.init(doc, opts, function(err) {
255    if (err) {
256      return cb(err);
257    }
258    cb(null, instance);
259  });
260}
261
262module.exports = QueryCursor;