|
/**
* Module dependecies.
*/
var Stream = require('stream').Stream;
/**
* CursorStream
*
* Returns a stream interface for the **cursor**.
*
* Events
* - **data** {function(item) {}} the data event triggers when a document is ready.
* - **error** {function(err) {}} the error event triggers if an error happens.
* - **close** {function() {}} the end event triggers when there is no more documents available.
*
* @class Represents a CursorStream.
* @param {Cursor} cursor a cursor object that the stream wraps.
* @return {Stream}
*/
function CursorStream(cursor) {
if(!(this instanceof CursorStream)) return new CursorStream(cursor);
Stream.call(this);
this.readable = true;
this.paused = false;
this._cursor = cursor;
this._destroyed = null;
// give time to hook up events
var self = this;
process.nextTick(function () {
self._init();
});
}
/**
* Inherit from Stream
* @ignore
* @api private
*/
CursorStream.prototype.__proto__ = Stream.prototype;
/**
* Flag stating whether or not this stream is readable.
*/
CursorStream.prototype.readable;
/**
* Flag stating whether or not this stream is paused.
*/
CursorStream.prototype.paused;
/**
* Initialize the cursor.
* @ignore
* @api private
*/
CursorStream.prototype._init = function () {
if (this._destroyed) return;
this._next();
}
/**
* Pull the next document from the cursor.
* @ignore
* @api private
*/
CursorStream.prototype._next = function () {
if (this.paused || this._destroyed) return;
var self = this;
// nextTick is necessary to avoid stack overflows when
// dealing with large result sets.
process.nextTick(function () {
self._cursor.nextObject(function (err, doc) {
self._onNextObject(err, doc);
});
});
}
/**
* Handle each document as its returned from the cursor.
* @ignore
* @api private
*/
CursorStream.prototype._onNextObject = function (err, doc) {
if (err) return this.destroy(err);
// when doc is null we hit the end of the cursor
if (!doc) {
this.emit('end')
return this.destroy();
}
this.emit('data', doc);
this._next();
}
/**
* Pauses the stream.
*
* @api public
*/
CursorStream.prototype.pause = function () {
this.paused = true;
}
/**
* Resumes the stream.
*
* @api public
*/
CursorStream.prototype.resume = function () {
var self = this;
process.nextTick(function() {
self.paused = false;
self._next();
})
}
/**
* Destroys the stream, closing the underlying
* cursor. No more events will be emitted.
*
* @api public
*/
CursorStream.prototype.destroy = function (err) {
if (this._destroyed) return;
this._destroyed = true;
this.readable = false;
this._cursor.close();
if (err) {
this.emit('error', err);
}
this.emit('close');
}
// TODO - maybe implement the raw option to pass binary?
//CursorStream.prototype.setEncoding = function () {
//}
module.exports = exports = CursorStream;
|
By viewing downloads associated with this article you agree to the Terms of Service and the article's licence.
If a file you wish to view isn't highlighted, and is a text file (not binary), please
let us know and we'll add colourisation support for it.
I currently hold the following qualifications (amongst others, I also studied Music Technology and Electronics, for my sins)
- MSc (Passed with distinctions), in Information Technology for E-Commerce
- BSc Hons (1st class) in Computer Science & Artificial Intelligence
Both of these at Sussex University UK.
Award(s)
I am lucky enough to have won a few awards for Zany Crazy code articles over the years
- Microsoft C# MVP 2016
- Codeproject MVP 2016
- Microsoft C# MVP 2015
- Codeproject MVP 2015
- Microsoft C# MVP 2014
- Codeproject MVP 2014
- Microsoft C# MVP 2013
- Codeproject MVP 2013
- Microsoft C# MVP 2012
- Codeproject MVP 2012
- Microsoft C# MVP 2011
- Codeproject MVP 2011
- Microsoft C# MVP 2010
- Codeproject MVP 2010
- Microsoft C# MVP 2009
- Codeproject MVP 2009
- Microsoft C# MVP 2008
- Codeproject MVP 2008
- And numerous codeproject awards which you can see over at my blog