<commit_before>var es = require('event-stream')

function MuxDemux (name) {

  function createID() {
    return (
      Math.random().toString(16).slice(2) +
      Math.random().toString(16).slice(2)
    )
  }

  var streams = {}, streamCount = 0
  var md = es.through(function (data) {
    var id = data.shift()
    var event = data[0]
    var s = streams[id]
    if(!s) {
      if(event != 'new')
        return md.emit('error', new Error('does not have stream:' + id))
      md.emit('connection', createStream(id, data[1].meta, data[1].opts))
    } 
    else if (event === 'pause')
      s.paused = true
    else if (event === 'resume') {
      var p = s.paused
      s.paused = false
      if(p) s.emit('drain')
    }
    else {
      s.emit.apply(s, data)
    }
  })

  function destroyAll (_err) {
    md.removeListener('end', destroyAll) 
    md.removeListener('error', destroyAll) 
    md.removeListener('close', destroyAll) 
    var err = _err || new Error ('unexpected disconnection')
    for (var i in streams) {
      var s = streams[i]
      s.emit('error', err)
      s.destroy()
    }
  }


  //the problem here, is that this is registering the first
  //event listener.
  //and so in this test, the close message is 
  //getting to the other side first
 md.pause = function () {}
  md.resume = function () {}

  function createStream(id, meta, opts) {
    var s = es.through(function (data) {
      if(!this.writable)
        throw new Error('stream is not writable')
      md.emit('data', [s.id, 'data', data])
    }, function () {
      md.emit('data', [s.id, 'end']) 
    })
    s.pause = function () {<commit_msg>turn errors off<commit_after>var es = require('event-stream')

function MuxDemux (opts) {

  function createID() {
    return (
      Math.random().toString(16).slice(2) +
      Math.random().toString(16).slice(2)
    )
  }

  var streams = {}, streamCount = 0
  var md = es.through(function (data) {
    var id = data.shift()
    var event = data[0]
    var s = streams[id]
    if(!s) {
      if(event != 'new')
        return md.emit('error', new Error('does not have stream:' + id))
      md.emit('connection', createStream(id, data[1].meta, data[1].opts))
    } 
    else if (event === 'pause')
      s.paused = true
    else if (event === 'resume') {
      var p = s.paused
      s.paused = false
      if(p) s.emit('drain')
    }
    else {
      s.emit.apply(s, data)
    }
  })

  function destroyAll (_err) {
    md.removeListener('end', destroyAll) 
    md.removeListener('error', destroyAll) 
    md.removeListener('close', destroyAll) 
    var err = _err || new Error ('unexpected disconnection')
    for (var i in streams) {
      var s = streams[i]
      if (opts && opts.error === false) {
        s.end()
      } else {
        s.emit('error', err)
        s.destroy() 
      }
    }
  }


  //the problem here, is that this is registering the first
  //event listener.
  //and so in this test, the close message is 
  //getting to the other side first
 md.pause = function () {}
  md.resume = function () {}

  function createStream(id, meta, opts) {
    var s = es.through(function (data) {
      if(!this.writable)
        throw new Error('stream is not writable')
      md.emit('data', [s.id, 'data', data])
    }, function () {
      md.emit('data', [s.id, 'end']) 
    })
    s.pause = function () {
{'fraction_non_alphanumeric': 0.15891472868217055, 'fraction_numerical': 0.005963029218843173, 'mean_word_length': 2.375251509054326, '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': '386416', 'n_tokens_mistral': 1139, 'n_tokens_neox': 1101, 'n_words': 300}