index.js
function (dag, opts) {
  var since = opts.since || 0
  var limit = opts.limit || -1
  var wait = null

  var read = function (size, cb) {
    if (dag.changes <= since) {
      wait = cb
      return
    }

    if (!limit) return cb(null, null)

    dag.db.get(CHANGES + lexint.pack(since + 1, 'hex'), function (err, hash) {
      if (err) return cb(err)
      dag.get(hash, opts, function (err, node) {
        if (err) return cb(err)
        since = node.change
        if (limit !== -1) limit--
        cb(null, node)
      })
    })
  }

  var kick = function () {
    if (!wait) return
    var cb = wait
    wait = null
    read(0, cb)
  }

  dag.on('add', kick)
  dag.ready(kick)

  var rs = from.obj(read)

  rs.once('close', function () {
    dag.removeListener('add', kick)
  })

  return rs
}
function (dag, opts) {
  var since = opts.since || 0
  var limit = opts.limit || -1
  var wait = null

  var read = function (size, cb) {
    if (dag.changes <= since) {
      wait = cb
      return
    }

    if (!limit) return cb(null, null)

    dag.db.get(CHANGES + lexint.pack(since + 1, 'hex'), function (err, hash) {
      if (err) return cb(err)
      dag.get(hash, opts, function (err, node) {
        if (err) return cb(err)
        since = node.change
        if (limit !== -1) limit--
        cb(null, node)
      })
    })
  }

  var kick = function () {
    if (!wait) return
    var cb = wait
    wait = null
    read(0, cb)
  }

  dag.on('add', kick)
  dag.ready(kick)

  var rs = from.obj(read)

  rs.once('close', function () {
    dag.removeListener('add', kick)
  })

  return rs
}
function ( dag , opts ) { var since = opts . since || 0 var limit = opts . limit || - 1 var wait = null var read = function ( size , cb ) { if ( dag . changes <= since ) { wait = cb return } if ( ! limit ) return cb ( null , null ) dag . db . get ( CHANGES + lexint . pack ( since + 1 , 'hex' ) , function ( err , hash ) { if ( err ) return cb ( err ) dag . get ( hash , opts , function ( err , node ) { if ( err ) return cb ( err ) since = node . change if ( limit !== - 1 ) limit -- cb ( null , node ) } ) } ) } var kick = function ( ) { if ( ! wait ) return var cb = wait wait = null read ( 0 , cb ) } dag . on ( 'add' , kick ) dag . ready ( kick ) var rs = from . obj ( read ) rs . once ( 'close' , function ( ) { dag . removeListener ( 'add' , kick ) } ) return rs }
Produce a readable stream of all nodes added from this point onward, in topographic order.
Produce a readable stream of all nodes added from this point onward in topographic order .