nl_tokens: Stream constructor
pl_tokens: 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 ( <mask> ) . 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 ) ; }