Stream constructor

@param  {Firenext} ref
@param  {[Object]} opts
function Stream(ref, opts) {
  var data = {},
      emitter = this,
      mapFn, started, ended, readable, marker;

  if (!opts) opts = {};
  if (!opts.page) opts.page = 10;


  // default mapping function
  mapFn = function(snapshot) {
    return snapshot;
  };

  /**
   * convert internal data object to Array
   * @return {Array}
   */

  function toArray() {
    return Object
      .keys(data)
      .map(function(name) { return data[name]; })
      .sort(function(a, b) {
        return a.compareTo(b);
      });
  }

  /**
   * add a snapshot to data
   *
   * @param {DataSnapshot} snapshot
   * @return {Promise}
   */

  function add(snapshot) {
    return Promise
      .resolve(mapFn(snapshot))
      .then(function(val) {
        data[snapshot.key()] = val;
      });
  }

  /**
   * convert data and emit value event
   */

  function emitValue() {
    var arr = toArray();
    marker = arr[arr.length - 1];
    setTimeout(function() {
      emitter.emit('value', arr);
    }, 0);
  }

  /**
   * remove child snapshot fom data
   */

  function onChildRemove(oldData) {
    delete data[oldData.key()];
    emitValue();
  }

  /**
   * add child to data buffer, and emit value
   *
   * @return {Promise}
   */

  function onChildAdd(newData) {

    // ignore all events before first read
    if (!started) return;

    // ignore already added item
    if (data[newData.key()]) return;

    // ignore older items that are not loaded
    if ( !ended
      && !data[newData.key()]
      && marker && newData.compareTo(marker) > 0) return;

    return add(newData).then(emitValue);
  }

  /**
   * add child to data buffer, and emit value
   *
   * @return {Promise}
   */

  function onChildUpdate(newData) {
    return add(newData).then(emitValue);
  }

  /**
   * add children to data buffer, and emit value
   *
   * @return {Promise}
   */

  function addChildren(newData) {
    var promises = [];
    newData.forEach(function (child) {
      promises.push(add(child));
    });

    return Promise.all(promises).then(emitValue);
  }

  /**
   * read next batch of children
   *
   * @chainable
   */

  this.read = function read() {
    var priority = marker && marker.getPriority(),
        key = marker && marker.key(),
        page = marker
          ? opts.page + 1
          : opts.page;

    readable = false;

    debug('read %s from %s children starting from %s', page, ref.toString(), key || 'top');

    ref
      .orderByPriority()
      .startAt(priority, key)
      .limitToFirst(page)
      .exec()
      .then(function(data) {

        // mark as started
        if (!started) started = true;

        return addChildren(data).then(function() {

          // check if there is more items
          ended = !data.hasChildren()
            || marker
                && data.numChildren() === 1
                && data.child(marker.key()).val();

          readable = !ended;

          if (ended) {
            emitter.emit('end');
          } else {
            emitter.emit('readable');
          }
        });
      })
      .catch(function(error) {
        emitter.emit('error', error);
      });

    return this;
  };

  /**
   * is stream readable
   *
   * @return {Boolean}
   */

  this.isReadable = function() {
    return readable;
  };

  /**
   * get array
   *
   * @return {Array}
   */

  this.toArray = toArray;

  /**
   * is stream ended
   *
   * @return {Boolean}
   */

  this.isStarted = function() {
    return started;
  };

  /**
   * is stream ended
   *
   * @return {Boolean}
   */

  this.isEnded = function() {
    return ended;
  };

  /**
   * remove all listeners
   */

  this.destroy = function() {
    emitter.removeAllListeners();
    ref.off('child_added', onChildAdd);
    ref.off('child_moved', onChildUpdate);
    ref.off('child_changed', onChildUpdate);
    ref.off('child_removed', onChildRemove);
  };

  /**
   * set mapping function
   *
   * @chainable
   */

  this.map = function(fn) {
    mapFn = fn;
    return this;
  };

  // listen to child_added event on both extremities
  ref.startAt().limitToFirst(1).on('child_added', onChildAdd);
  ref.startAt().limitToLast(1).on('child_added', onChildAdd);

  ref.on('child_changed', onChildUpdate);
  // ref.on('child_moved', onChildUpdate);
  ref.on('child_removed', onChildRemove);
}