Lazy asynchronous iterator

LazyIterator.js

/* jshint browser: true */

(function (root) {
  'use strict';

  /**
   * Simple lazy asynchronous iterator.
   * 
   * Lazily iterate over a collection served from a standard REST endpoint.
   * Elements are loaded by batch (of size `highWaterMark`), and
   * pre-emptively fetched (once the iterator has less than `lowWaterMark`
   * elements cached.)
   * 
   * Sample usage:
   * 
   *   var iterator = new LazyIterator(
   *     'http://localhost:8080/things',
   *     {lowWaterMark: 2, highWaterMark: 10}
   *   );
   * 
   *   iterator.next(function (err, elem) {
   *     // Do things (`elem` will be `null` if
   *     // the collection is exhausted).
   *   });
   * 
   */
  function LazyIterator(url, opts) {

    opts = opts || {};
    var lowWaterMark = opts.lowWaterMark || 3;
    var highWaterMark = opts.highWaterMark || 5;
    var params = JSON.parse(JSON.stringify(opts.params || {}));
    var transform = opts.transform || JSON.parse;
    var limitParamName = opts.limitParamName || 'limit';
    var offsetParamName = opts.offsetParamName || 'offset';

    params[offsetParamName] = 0;
    params[limitParamName] = highWaterMark;

    var exhausted = false;
    var fetching = false;
    var fetchCb = null;
    var data = [];

    this.next = function (cb) {

      if (!exhausted && !data.length) {
        fetch(cb);
      } else {
        if (!exhausted && data.length <= lowWaterMark) {
          fetch();
        }
        setTimeout(function () { cb(null, data.shift()); }, 0);
      }

    };

    function fetch(cb) {

      if (fetching) {
        if (fetchCb && cb) {
          cb(new Error('Already fetching.'));
        } else {
          fetchCb = cb;
        }
        return;
      }

      fetching = true;
      fetchCb = cb;

      var req = new XMLHttpRequest();
      req.open('GET', url + '?' + serialize(params), true);
      req.setRequestHeader('Accept', 'application/json');

      req.onreadystatechange = function () {

        if (req.readyState !== 4 || !fetchCb) {
          return;
        }

        fetching = false;
        if (req.status === 200) {
          data = data.concat(transform(req.response));
          if (data.length) {
            params[offsetParamName] += data.length;
          } else {
            exhausted = true;
          }
          fetchCb(null, data.shift());
        } else {
          var err = new Error(req.statusText);
          err.req = req;
          fetchCb(err);
        }

      };

      req.send();

      function serialize(obj, prefix) {

        var arr = [];
        var k;
        for (k in obj) {
          if (obj.hasOwnProperty(k)) {
            var p = prefix ? prefix + '[' + k + ']' : k;
            var v = obj[p];
            arr.push(
              typeof v == 'object' ?
                serialize(v, p) :
                encodeURIComponent(p) + '=' + encodeURIComponent(v)
            );
          }
        }
        return arr.join('&');

      }

    }

  }

  root.LazyIterator = LazyIterator;

})(this);