<commit_before>var duplexify = require('duplexify')
var eos = require('end-of-stream')
var streams = require('./streams')

var noop = function () {}

var pbs = function (down, encode, decode) {
  var iterators = []

  decode.puts(function (req, cb) {
    down.put(req.key, req.value, function (err) {
      encode.callbacks({id: req.id, error: err && err.message}, cb)
    })
  })

  decode.gets(function (req, cb) {
    down.get(req.key, {valueEncoding: 'binary'}, function (err, value) {
      encode.callbacks({id: req.id, error: err && err.message, value: value}, cb)
    })
  })

  decode.deletes(function (req, cb) {
    down.del(req.key, function (err) {
      encode.callbacks({id: req.id, error: err && err.message}, cb)
    })
  })

  decode.batches(function (req, cb) {
    down.batch(req.ops, function (err) {
      encode.callbacks({id: req.id, error: err && err.message}, cb)
    })
  })

  decode.iterators(function (req, cb) {
    while (iterators.length < req.id) iterators.push(null)

    var prev = iterators[req.id]

    if (prev) {
      if (!req.batch) {
        prev.end()
      } else {
        prev.batch = req.batch
        prev.next()
      }
      return cb()
    }

    var first = true
    var nexting = false
    var ended = false
    var iterator = down.iterator(req.options)

    var end = function () {
      ended = true
      iterators[req.id] = null
      while (iterators.length && iterators[iterators.length - 1]) iterators.pop()
      iterator.end(noop)
    }

    var next = function () {
      if (nexting || ended) return
      if (!first && (!data.batch || data.error || (!data.key && !data.value))) return
      first = false
      nexting = true
      iterator.next(function (err, key, value) {
        nexting = false<commit_msg>[fix] add correct condition in iterator end on server found by @mafintosh<commit_after>var duplexify = require('duplexify')
var eos = require('end-of-stream')
var streams = require('./streams')

var noop = function () {}

var pbs = function (down, encode, decode) {
  var iterators = []

  decode.puts(function (req, cb) {
    down.put(req.key, req.value, function (err) {
      encode.callbacks({id: req.id, error: err && err.message}, cb)
    })
  })

  decode.gets(function (req, cb) {
    down.get(req.key, {valueEncoding: 'binary'}, function (err, value) {
      encode.callbacks({id: req.id, error: err && err.message, value: value}, cb)
    })
  })

  decode.deletes(function (req, cb) {
    down.del(req.key, function (err) {
      encode.callbacks({id: req.id, error: err && err.message}, cb)
    })
  })

  decode.batches(function (req, cb) {
    down.batch(req.ops, function (err) {
      encode.callbacks({id: req.id, error: err && err.message}, cb)
    })
  })

  decode.iterators(function (req, cb) {
    while (iterators.length < req.id) iterators.push(null)

    var prev = iterators[req.id]

    if (prev) {
      if (!req.batch) {
        prev.end()
      } else {
        prev.batch = req.batch
        prev.next()
      }
      return cb()
    }

    var first = true
    var nexting = false
    var ended = false
    var iterator = down.iterator(req.options)

    var end = function () {
      ended = true
      iterators[req.id] = null
      while (iterators.length && !iterators[iterators.length - 1]) iterators.pop()
      iterator.end(noop)
    }

    var next = function () {
      if (nexting || ended) return
      if (!first && (!data.batch || data.error || (!data.key && !data.value))) return
      first = false
      nexting = true
      iterator.next(function (err, key, value) {
        nexting = false
{'fraction_non_alphanumeric': 0.15665651812897868, 'fraction_numerical': 0.0005535566011624688, 'mean_word_length': 2.5641025641025643, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, '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': '4243247', 'n_tokens_mistral': 1217, 'n_tokens_neox': 1187, 'n_words': 298}