1/* eslint no-empty: 1 */
  2
  3/*!
  4 * Module dependencies.
  5 */
  6
  7var Stream = require('stream').Stream;
  8var utils = require('./utils');
  9var helpers = require('./queryhelpers');
 10var K = function(k) {
 11  return k;
 12};
 13
 14/**
 15 * Provides a Node.js 0.8 style [ReadStream](http://nodejs.org/docs/v0.8.21/api/stream.html#stream_readable_stream) interface for Queries.
 16 *
 17 *     var stream = Model.find().stream();
 18 *
 19 *     stream.on('data', function (doc) {
 20 *       // do something with the mongoose document
 21 *     }).on('error', function (err) {
 22 *       // handle the error
 23 *     }).on('close', function () {
 24 *       // the stream is closed
 25 *     });
 26 *
 27 *
 28 * The stream interface allows us to simply "plug-in" to other _Node.js 0.8_ style write streams.
 29 *
 30 *     Model.where('created').gte(twoWeeksAgo).stream().pipe(writeStream);
 31 *
 32 * ####Valid options
 33 *
 34 *   - `transform`: optional function which accepts a mongoose document. The return value of the function will be emitted on `data`.
 35 *
 36 * ####Example
 37 *
 38 *     // JSON.stringify all documents before emitting
 39 *     var stream = Thing.find().stream({ transform: JSON.stringify });
 40 *     stream.pipe(writeStream);
 41 *
 42 * _NOTE: plugging into an HTTP response will *not* work out of the box. Those streams expect only strings or buffers to be emitted, so first formatting our documents as strings/buffers is necessary._
 43 *
 44 * _NOTE: these streams are Node.js 0.8 style read streams which differ from Node.js 0.10 style. Node.js 0.10 streams are not well tested yet and are not guaranteed to work._
 45 *
 46 * @param {Query} query
 47 * @param {Object} [options]
 48 * @inherits NodeJS Stream http://nodejs.org/docs/v0.8.21/api/stream.html#stream_readable_stream
 49 * @event `data`: emits a single Mongoose document
 50 * @event `error`: emits when an error occurs during streaming. This will emit _before_ the `close` event.
 51 * @event `close`: emits when the stream reaches the end of the cursor or an error occurs, or the stream is manually `destroy`ed. After this event, no more events are emitted.
 52 * @api public
 53 */
 54
 55function QueryStream(query, options) {
 56  Stream.call(this);
 57
 58  this.query = query;
 59  this.readable = true;
 60  this.paused = false;
 61  this._cursor = null;
 62  this._destroyed = null;
 63  this._fields = null;
 64  this._buffer = null;
 65  this._inline = T_INIT;
 66  this._running = false;
 67  this._transform = options && typeof options.transform === 'function'
 68      ? options.transform
 69      : K;
 70
 71  // give time to hook up events
 72  var _this = this;
 73  process.nextTick(function() {
 74    _this._init();
 75  });
 76}
 77
 78/*!
 79 * Inherit from Stream
 80 */
 81
 82QueryStream.prototype.__proto__ = Stream.prototype;
 83
 84/**
 85 * Flag stating whether or not this stream is readable.
 86 *
 87 * @property readable
 88 * @api public
 89 */
 90
 91QueryStream.prototype.readable;
 92
 93/**
 94 * Flag stating whether or not this stream is paused.
 95 *
 96 * @property paused
 97 * @api public
 98 */
 99
100QueryStream.prototype.paused;
101
102// trampoline flags
103var T_INIT = 0;
104var T_IDLE = 1;
105var T_CONT = 2;
106
107/**
108 * Initializes the query.
109 *
110 * @api private
111 */
112
113QueryStream.prototype._init = function() {
114  if (this._destroyed) {
115    return;
116  }
117
118  var query = this.query,
119      model = query.model,
120      options = query._optionsForExec(model),
121      _this = this;
122
123  try {
124    query.cast(model);
125  } catch (err) {
126    return _this.destroy(err);
127  }
128
129  _this._fields = utils.clone(query._fields);
130  options.fields = query._castFields(_this._fields);
131
132  model.collection.find(query._conditions, options, function(err, cursor) {
133    if (err) {
134      return _this.destroy(err);
135    }
136    _this._cursor = cursor;
137    _this._next();
138  });
139};
140
141/**
142 * Trampoline for pulling the next doc from cursor.
143 *
144 * @see QueryStream#__next #querystream_QueryStream-__next
145 * @api private
146 */
147
148QueryStream.prototype._next = function _next() {
149  if (this.paused || this._destroyed) {
150    this._running = false;
151    return this._running;
152  }
153
154  this._running = true;
155
156  if (this._buffer && this._buffer.length) {
157    var arg;
158    while (!this.paused && !this._destroyed && (arg = this._buffer.shift())) { // eslint-disable-line no-cond-assign
159      this._onNextObject.apply(this, arg);
160    }
161  }
162
163  // avoid stack overflows with large result sets.
164  // trampoline instead of recursion.
165  while (this.__next()) {
166  }
167};
168
169/**
170 * Pulls the next doc from the cursor.
171 *
172 * @see QueryStream#_next #querystream_QueryStream-_next
173 * @api private
174 */
175
176QueryStream.prototype.__next = function() {
177  if (this.paused || this._destroyed) {
178    this._running = false;
179    return this._running;
180  }
181
182  var _this = this;
183  _this._inline = T_INIT;
184
185  _this._cursor.nextObject(function cursorcb(err, doc) {
186    _this._onNextObject(err, doc);
187  });
188
189  // if onNextObject() was already called in this tick
190  // return ourselves to the trampoline.
191  if (T_CONT === this._inline) {
192    return true;
193  }
194  // onNextObject() hasn't fired yet. tell onNextObject
195  // that its ok to call _next b/c we are not within
196  // the trampoline anymore.
197  this._inline = T_IDLE;
198};
199
200/**
201 * Transforms raw `doc`s returned from the cursor into a model instance.
202 *
203 * @param {Error|null} err
204 * @param {Object} doc
205 * @api private
206 */
207
208QueryStream.prototype._onNextObject = function _onNextObject(err, doc) {
209  if (this._destroyed) {
210    return;
211  }
212
213  if (this.paused) {
214    this._buffer || (this._buffer = []);
215    this._buffer.push([err, doc]);
216    this._running = false;
217    return this._running;
218  }
219
220  if (err) {
221    return this.destroy(err);
222  }
223
224  // when doc is null we hit the end of the cursor
225  if (!doc) {
226    this.emit('end');
227    return this.destroy();
228  }
229
230  var opts = this.query._mongooseOptions;
231
232  if (!opts.populate) {
233    return opts.lean === true ?
234        emit(this, doc) :
235        createAndEmit(this, null, doc);
236  }
237
238  var _this = this;
239  var pop = helpers.preparePopulationOptionsMQ(_this.query, _this.query._mongooseOptions);
240
241  // Hack to work around gh-3108
242  pop.forEach(function(option) {
243    delete option.model;
244  });
245
246  pop.__noPromise = true;
247  _this.query.model.populate(doc, pop, function(err, doc) {
248    if (err) {
249      return _this.destroy(err);
250    }
251    return opts.lean === true ?
252        emit(_this, doc) :
253        createAndEmit(_this, pop, doc);
254  });
255};
256
257function createAndEmit(self, populatedIds, doc) {
258  var instance = helpers.createModel(self.query.model, doc, self._fields);
259  var opts = populatedIds ?
260    {populated: populatedIds} :
261    undefined;
262
263  instance.init(doc, opts, function(err) {
264    if (err) {
265      return self.destroy(err);
266    }
267    emit(self, instance);
268  });
269}
270
271/*!
272 * Emit a data event and manage the trampoline state
273 */
274
275function emit(self, doc) {
276  self.emit('data', self._transform(doc));
277
278  // trampoline management
279  if (T_IDLE === self._inline) {
280    // no longer in trampoline. restart it.
281    self._next();
282  } else {
283    // in a trampoline. tell __next that its
284    // ok to continue jumping.
285    self._inline = T_CONT;
286  }
287}
288
289/**
290 * Pauses this stream.
291 *
292 * @api public
293 */
294
295QueryStream.prototype.pause = function() {
296  this.paused = true;
297};
298
299/**
300 * Resumes this stream.
301 *
302 * @api public
303 */
304
305QueryStream.prototype.resume = function() {
306  this.paused = false;
307
308  if (!this._cursor) {
309    // cannot start if not initialized
310    return;
311  }
312
313  // are we within the trampoline?
314  if (T_INIT === this._inline) {
315    return;
316  }
317
318  if (!this._running) {
319    // outside QueryStream control, need manual restart
320    return this._next();
321  }
322};
323
324/**
325 * Destroys the stream, closing the underlying cursor, which emits the close event. No more events will be emitted after the close event.
326 *
327 * @param {Error} [err]
328 * @api public
329 */
330
331QueryStream.prototype.destroy = function(err) {
332  if (this._destroyed) {
333    return;
334  }
335  this._destroyed = true;
336  this._running = false;
337  this.readable = false;
338
339  if (this._cursor) {
340    this._cursor.close();
341  }
342
343  if (err) {
344    this.emit('error', err);
345  }
346
347  this.emit('close');
348};
349
350/**
351 * Pipes this query stream into another stream. This method is inherited from NodeJS Streams.
352 *
353 * ####Example:
354 *
355 *     query.stream().pipe(writeStream [, options])
356 *
357 * @method pipe
358 * @memberOf QueryStream
359 * @see NodeJS http://nodejs.org/api/stream.html
360 * @api public
361 */
362
363/*!
364 * Module exports
365 */
366
367module.exports = exports = QueryStream;