nosql-stream/lib/read-stream.js

(function() {
  var EncodingError, Errors, FILTER_EXCLUDED, FILTER_INCLUDED, FILTER_STOPPED, ReadStream, Readable, consts, defaultOptions, extend, inherits, isFunction, isObject, isString, minimatch, util;

  Readable = require('readable-stream').Readable;

  minimatch = require('minimatch');

  util = require('abstract-object/lib/util');

  Errors = require('./errors');

  consts = require('./consts');

  inherits = util.inherits;

  isFunction = util.isFunction;

  isObject = util.isObject;

  isString = util.isString;

  extend = util._extend;

  FILTER_INCLUDED = consts.FILTER_INCLUDED;

  FILTER_EXCLUDED = consts.FILTER_EXCLUDED;

  FILTER_STOPPED = consts.FILTER_STOPPED;

  EncodingError = Errors.EncodingError;

  defaultOptions = {
    highWaterMark: 1e5
  };


  /*
  readStream is used to search and read the [abstract-nosql](https://github.com/snowyu/abstract-nosql) database.
  
  you must implement the iterator.next() and iterator.end() to use. (see [abstract-nosql](https://github.com/snowyu/abstract-nosql))
  
  The resulting stream is a Node.js-style [Readable Stream](http://nodejs.org/docs/latest/api/stream.html#stream_readable_stream) 
  where `'data'` events emit objects with `'key'` and `'value'` pairs. 
  
  You can also use the `gt`, `lt` and `limit` options to control the 
  range of keys that are streamed.
   */

  module.exports = ReadStream = (function() {
    inherits(ReadStream, Readable);

    function ReadStream(db, aOptions, aMakeData) {
      if (!(this instanceof ReadStream)) {
        return new ReadStream(db, aOptions, aMakeData);
      }
      this._options = extend({}, defaultOptions);
      this._options = extend(this._options, aOptions || {});
      aOptions = this._options;
      Readable.call(this, {
        objectMode: true,
        highWaterMark: aOptions.highWaterMark
      });
      this._waiting = false;
      if (db == null) {
        db = aOptions.db;
      }
      if (aMakeData) {
        this._makeData = aMakeData;
      } else {
        this._makeData = aOptions.keys !== false && aOptions.values !== false ? function(key, value) {
          return {
            key: key,
            value: value
          };
        } : aOptions.values === false ? function(key) {
          return key;
        } : aOptions.keys === false ? function(_, value) {
          return value;
        } : function() {};
      }
      if (db) {
        if (!db.isOpen || db.isOpen()) {
          this.setIterator(db.iterator(aOptions));
        } else {
          db.once('ready', (function(_this) {
            return function() {
              return _this.setIterator(db.iterator(aOptions));
            };
          })(this));
        }
      }
    }

    ReadStream.prototype.setIterator = function(aIterator) {
      this._iterator = aIterator;
      if (this._destroyed) {
        return aIterator.end(function() {});
      }
      if (this._waiting) {
        this._waiting = false;
        return this._read();
      }
      return this;
    };

    ReadStream.prototype._read = function() {
      var self;
      if (this._destroyed) {
        return;
      }
      if (!this._iterator) {
        return this._waiting = true;
      }
      self = this;
      return this._iterator.next(function(err, key, value) {
        var e;
        if (err || (key === void 0 && value === void 0)) {
          if (!err && !self._destroyed) {
            self.push(null);
          }
          return self._cleanup(err);
        }
        try {
          value = self._makeData(key, value);
        } catch (_error) {
          e = _error;
          return self._cleanup(new EncodingError(e));
        }
        if (!self._destroyed) {
          return self.push(value);
        }
      });
    };

    ReadStream.prototype._cleanup = function(aError) {
      if (this._destroyed) {
        return;
      }
      this._destroyed = true;
      if (aError) {
        this.emit('error', err);
      }
      if (this._iterator) {
        this.emit('last', this._iterator.last);
        return this._iterator.end((function(_this) {
          return function() {
            _this._iterator = null;
            return _this.emit('close');
          };
        })(this));
      } else {
        return this.emit('close');
      }
    };

    ReadStream.prototype.destroy = function() {
      return this.cleanup();
    };

    ReadStream.prototype.toString = function() {
      return 'NoSQLReadStream';
    };

    return ReadStream;

  })();

}).call(this);

//# sourceMappingURL=read-stream.js.map