<commit_before>var pull = require('pull-stream')

exports = module.exports = 
pull.pipeableSource(function (getStream, length, min, max) {
  if(min == null)
    return getStream(0)
      .pipe(range(min, max))
  //getStream(n) returns an iterator (pull-stream style)
  //that starts in the nth block.
  //length is the number of blocks.
  //compare is a sort function like Array#sort(comparator)
  
  var defer = pull.defer()

  ;(function recurse (left, right, lStream, rStream, lPeek, rPeek) {
    if(left + 1 === right) {
      return defer.resolve(lStream || rStream)
    }
    var middle = Math.round((left + right) / 2)
    middle = Math.min(Math.max(0, middle), length)

    var mStream =
      getStream(middle)
      .pipe(peek(function (end, mPeek) {
        if(mPeek == min)
          defer.resolve(mStream)
        else if(mPeek > min)
          recurse(left, middle, lStream, mStream, lPeek, mPeek)
        else
          recurse(middle, right, mStream, rStream, mPeek, rPeek)
      }))

  })(-1, length)

  return defer.pipe(range(min, max))
})

var range = exports.range = function (min, max) {
  return pull.filter(function (data) {
    return min == null || min <= data
  }).pipe(pull.take(function (data) {
    return max == null || max >= data
  }))
}


//get the first item from a stream,
//and return a new stream that 

var peek = exports.peek = pull.pipeable(function (read, cb) {
  var ended, _cb, _data, ready = false

  read(null, function (end, data) {
    ended = ended || end
    _data = data
    //incase read has responded syncly,
    //delay the callback so some one can recieve the returned read!)
    process.nextTick(function () {
      cb(ended, _data)
      ready = true
      if(_cb) _cb(ended, _data)
    })
  })

  return function (end, cb) {
    //end immediately, if ending...
    if(end) return read(end, cb)
    //only allow one peek!
    if(_cb && !ready)
      throw new Error('read must not be called until after callback')
    if(!ready)    _cb = cb
    else if(!_cb) _cb = true, cb(ended, _data)
    else          read(end, cb)

  }
})


<commit_msg>you have to pass compare!<commit_after>var pull = require('pull-stream')

exports = module.exports = 
pull.pipeableSource(function (getStream, compare, length, min, max) {

  if(min == null)
    return getStream(0)
      .pipe(range(min, max))
  //getStream(n) returns an iterator (pull-stream style)
  //that starts in the nth block.
  //length is the number of blocks.
  //compare is a sort function like Array#sort(comparator)
  
  var defer = pull.defer()

  ;(function recurse (left, right, lStream, rStream, lPeek, rPeek) {
    if(left + 1 === right) {
      return defer.resolve(lStream || rStream)
    }

    var middle = Math.round((left + right) / 2)
    middle = Math.min(Math.max(0, middle), length)

    var mStream =
      getStream(middle)
      .pipe(peek(function (end, mPeek) {
        console.log('COMPARE', mPeek, min)
        if(mPeek == min)
          defer.resolve(mStream)
        else if(mPeek > min)
          recurse(left, middle, lStream, mStream, lPeek, mPeek)
        else
          recurse(middle, right, mStream, rStream, mPeek, rPeek)
      }))

  })(-1, length)

  return defer.pipe(range(min, max))
})

var range = exports.range = function (min, max) {
  return pull.filter(function (data) {
    return min == null || min <= data
  }).pipe(pull.take(function (data) {
    return max == null || max >= data
  }))
}


//get the first item from a stream,
//and return a new stream that 

var peek = exports.peek = pull.pipeable(function (read, cb) {
  var ended, _cb, _data, ready = false

  read(null, function (end, data) {
    ended = ended || end
    _data = data
    //incase read has responded syncly,
    //delay the callback so some one can recieve the returned read!)
    process.nextTick(function () {
      cb(ended, _data)
      ready = true
      if(_cb) _cb(ended, _data)
    })
  })

  return function (end, cb) {
    //end immediately, if ending...
    if(end) return read(end, cb)
    //only allow one peek!
    if(_cb && !ready)
      throw new Error('read must not be called until after callback')
    if(!ready)    _cb = cb
    else if(!_cb) _cb = true, cb(ended, _data)
    else          read(end, cb)

  }
})



{'fraction_non_alphanumeric': 0.13344277673545965, 'fraction_numerical': 0.0023452157598499064, 'mean_word_length': 2.651541095890411, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6824213', 'n_tokens_mistral': 1449, 'n_tokens_neox': 1409, 'n_words': 440}