<commit_before> 'use strict';

//split one stream into two.
//if the stream ends, end both streams.
//if a reader stream aborts,
//continue to write to the other stream.

module.exports = function (sinks, select, create) {
  if('function' === typeof sinks) {
    create = select
    select = sinks
    sinks = []
  }

  var running = sinks.length
  var cbs     = new Array(running)
  var aborted = new Array(running)
  var open = false, ended, j, data, read, reading

  function init (key, reader) {
    reader(function (abort, cb) {
      if(abort) {
        aborted[key] = abort;
        --running
        //cbs[key] = cb
        //if all sinks have aborted, abort the sink.
        if(j === key) j = null
        pull(running ? null : abort)
        cb(abort)
        //continue, incase we have already read something
        //for this stream. might need to drop that.
        return
      }

      if(j === key) {
        var _j = j; j = null
        if(aborted[key]) return
        cb(null, data)
      }
      else if(ended) {
        return cb(ended)
      }
      else
        cbs[key] = cb

      pull()
    })
  }

  for(var k in sinks)
    init(k, sinks[k])

  function endAll () {
    for(var k in cbs) {
      if(cbs[k]) {
        var cb = cbs[k]
        cbs[k] = null
        cb (ended)
      }
    }
  }

  function pull (abort) {
    if(reading || !read || j != null) return
    reading = true
    read(abort, function (end, _data) {
      reading = false
      var _j

      if(end) {
        ended = end
        return endAll()
      }
      else
        _j = select(_data, sinks)

      if(aborted[_j])
        return pull()

      if(!sinks[_j]) {
        //edgecase: if select returns a new stream
        //          but user did not provide create?
        sinks[_j] = create(_j, sinks)
        //init will pass the sink read,
        //which will then grab data.
        //so we don't need to be in this callback anymore.
        j = _j
        data = _data
        return init(_j, sinks[_j])
      }

      if(cbs[_j]) {
        var cb = cbs[_j]
        cbs[_j] = null
        cb(null, _data)
      } else {
        j = _j
        data = _data
      }
    })
  }

  return function (_read) {
    read = _read
    pull()
  }

}<commit_msg>.add method<commit_after> 'use strict';

//split one stream into two.
//if the stream ends, end both streams.
//if a reader stream aborts,
//continue to write to the other stream.

module.exports = function (sinks, select, create) {
  if('function' === typeof sinks) {
    create = select
    select = sinks
    sinks = []
  }

  var running = 0
  var cbs     = new Array(sinks.length)
  var aborted = new Array(sinks.length)
  var open = false, ended, j, data, read, reading

  function init (key, reader) {
    running ++
    reader(function (abort, cb) {
      if(abort) {
        aborted[key] = abort;
        --running
        //cbs[key] = cb
        //if all sinks have aborted, abort the sink.
        if(j === key) j = null

        pull(running ? null : abort)
        if(cb) cb(abort)
        //continue, incase we have already read something
        //for this stream. might need to drop that.
        return
      }

      if(j === key) {
        var _j = j; j = null
        if(aborted[key]) return
        cb(null, data)
      }
      else if(ended) {
        return cb(ended)
      }
      else {
        cbs[key] = cb
      }
      pull()
    })
  }

  for(var k in sinks)
    init(k, sinks[k])

  function endAll () {
    for(var k in cbs) {
      if(cbs[k]) {
        var cb = cbs[k]
        cbs[k] = null
        cb (ended)
      }
    }
  }

  function pull (abort) {
    if(reading || !read || j != null) return
    reading = true
    read(abort, function (end, _data) {
      reading = false
      var _j
      if(end) {
        ended = end
        return endAll()
      }
      else
        _j = select(_data, sinks)

      if(aborted[_j])
        return pull()

      if(!sinks[_j]) {
        //edgecase: if select returns a new stream
        //          but user did not provide create?
        sinks[_j] = create(_j, sinks)
        //init will pass the sink read,
        //which will then grab data.
        //so we don't need to be in this callback anymore.
        j = _j
        data = _data
        return init(_j, sinks[_j])
      }

      if(cbs[_j]) {
        var cb = cbs[_j]
        cbs[_j] = null
        cb(null, _data)
      } else {
        j = _j; data = _data
      }
    })
  }

  function reader (_read) {
    read = _read; pull()
  }

  reader.add = function (key, sink) {
    sinks[key] = sink
    init(key, sink)
  }

  return reader
}
{'fraction_non_alphanumeric': 0.11857112115343232, 'fraction_numerical': 0.00021519259737465033, 'mean_word_length': 1.6024636058230683, 'pattern_counts': {'":': 0, '<': 3, '<?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': '386420', 'n_tokens_mistral': 1549, 'n_tokens_neox': 1510, 'n_words': 487}