<commit_before>import {Readable} from "stream"

export default tuple => {
  let sink = Readable({read, objectMode: true})
  let names = Object.keys(tuple)
  let sources = names.map(name => tuple[name])

  let isStarted = false
  let isPushable = false

  let numLeft = sources.length
  let numReadable = 0

  function start() {
    isStarted = true

    for (let source of sources) {
      source.on("readable", onreadable)
      source.on("end", onend)
    }
  }

  function onreadable() {
    numReadable++
    flush()
  }

  function onend() {
    numLeft--
    flush()
  }

  function read() {
    if (!isStarted) start()

    isPushable = true
    flush()
  }

  function flush() {
    while (isPushable && numReadable === numLeft) {
      let kvs = sources.map(source => source.read())
      let key = kvs.map(kv => kv && kv.key).sort()[0]

      if (!key) return sink.push(null)

      let value = new tuple.constructor

      kvs.forEach((kv, i) => {
        if (!kv) numReadable--

        else if (kv.key !== key) sources[i].unshift(kv)

        else value[names[i]] = kv.value
      })

      sink.push({key, value})
    }
  }

  return sink
}<commit_msg>try transforms instead of .read()/readable<commit_after>import {Readable, Transform} from "stream"

export default tuple => {
  let sources = Object.keys(tuple).map(name => {
    let source = {name}

    let transform = Transform({
      objectMode: true,

      transform(kv, enc, cb) {
        source.kv = kv
        source.cb = cb

        push()
      },

      flush(cb) {
        let index = sources.filter(x => x.name === name)[0]
        sources.splice(index, 1)

        push()
        cb()
      }
    })

    tuple[name].pipe(transform)

    return source
  })

  let pushable = false
  let read = () => {
    pushable = true
    push()
  }

  let push = () => {
    if (!sources.length) return tuples.push(null)

    let kvs = sources.map(source => source.kv)
    let ready = kvs.every(Boolean)

    if (!ready) return

    let value = new tuple.constructor
    let key = kvs.map(kv => kv.key).sort()[0]

    let cbs = sources
      .filter(source => source.kv && source.kv.key === key)
      .map(source => {
        let cb = source.cb

        value[source.name] = source.kv.value

        delete source.kv
        delete source.cb

        return cb
      })

    pushable = tuples.push({key, value})

    cbs.forEach(cb => cb())
  }

  let tuples = new Readable({read, objectMode: true})

  return tuples
}
{'fraction_non_alphanumeric': 0.14067524115755628, 'fraction_numerical': 0.0020096463022508037, 'mean_word_length': 2.2116129032258063, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 18, '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': '1974282', 'n_tokens_mistral': 842, 'n_tokens_neox': 796, 'n_words': 208}