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