<commit_before>'use strict';

var map = require('../')
  , it = require('it-is')
  , u = require('ubelt')
  , spec = require('stream-spec')
  , from = require('from')
  , Stream = require('stream')
  , es = require('event-stream')

//REFACTOR THIS TEST TO USE es.readArray and es.writeArray

function writeArray(array, stream) {

  array.forEach( function (j) {
    stream.write(j)
  })
  stream.end()

}

function readStream(stream, done) {

  var array = [] 
  stream.on('data', function (data) {
    array.push(data)
  })
  stream.on('error', done)
  stream.on('end', function (data) {
    done(null, array)
  })

} 

//call sink on each write,
//and complete when finished.

function pauseStream (prob, delay) { 
  var pauseIf = (
      'number' == typeof prob 
    ? function () {
        return Math.random() < prob
      } 
    : 'function' == typeof prob 
    ? prob
    : 0.1
  )
  var delayer = ( 
      !delay 
    ? process.nextTick
    : 'number' == typeof delay 
    ? function (next) { setTimeout(next, delay) }
    : delay
  )   

  return es.through(function (data) {    
    if(!this.paused && pauseIf()) {
      console.log('PAUSE STREAM PAUSING')
      this.pause()
      var self = this
      delayer(function () {
        console.log('PAUSE STREAM RESUMING')
        self.resume()
      })
    }
    console.log("emit ('data', " + data + ')')
    this.emit('data', data) 
  })
}

exports ['simple map applied to a stream'] = function (test) {

  var input = [1,2,3,7,5,3,1,9,0,2,4,6]
  //create event stream from

  var doubler = map(function (data, cb) {
    cb(null, data * 2)
  })

  spec(doubler).through().validateOnExit()

  //a map is only a middle man, so it is both readable and writable
  
  it(doubler).has({
    readable: true,
    writable: true,   
  })

  readStream(doubler, function (err, output) {
    it(output).deepEqual(input.map(function (j) {
      return j * 2
    }))
//    process.nextTick(x.validate)
    test.done()
  })
  
  writeArray(input, doubler)
  
}

exports['pipe two maps together'] = function (test) {

  var input = [1,2,3,7,5,3,1,9,0,2,4,6]
  //create event stream from
  function dd (data, cb) {
    cb(null, data * 2)
  }
  var doubler1 = map(dd), doubler2 = map(dd)

  doubler1.pipe(doubler2)
  
  spec(doubler1).through().validateOnExit()
  spec(doubler2).through().validateOnExit()

  readStream(doubler2, function (err, output) {<commit_msg>add failing test for order preservation<commit_after>'use strict';

var map = require('../')
  , it = require('it-is')
  , u = require('ubelt')
  , spec = require('stream-spec')
  , from = require('from')
  , Stream = require('stream')
  , es = require('event-stream')

//REFACTOR THIS TEST TO USE es.readArray and es.writeArray

function writeArray(array, stream) {

  array.forEach( function (j) {
    stream.write(j)
  })
  stream.end()

}

function readStream(stream, done) {

  var array = [] 
  stream.on('data', function (data) {
    array.push(data)
  })
  stream.on('error', done)
  stream.on('end', function (data) {
    done(null, array)
  })

} 

//call sink on each write,
//and complete when finished.

function pauseStream (prob, delay) { 
  var pauseIf = (
      'number' == typeof prob 
    ? function () {
        return Math.random() < prob
      } 
    : 'function' == typeof prob 
    ? prob
    : 0.1
  )
  var delayer = ( 
      !delay 
    ? process.nextTick
    : 'number' == typeof delay 
    ? function (next) { setTimeout(next, delay) }
    : delay
  )   

  return es.through(function (data) {    
    if(!this.paused && pauseIf()) {
      console.log('PAUSE STREAM PAUSING')
      this.pause()
      var self = this
      delayer(function () {
        console.log('PAUSE STREAM RESUMING')
        self.resume()
      })
    }
    console.log("emit ('data', " + data + ')')
    this.emit('data', data) 
  })
}

exports ['simple map applied to a stream'] = function (test) {

  var input = [1,2,3,7,5,3,1,9,0,2,4,6]
  //create event stream from

  var doubler = map(function (data, cb) {
    cb(null, data * 2)
  })

  spec(doubler).through().validateOnExit()

  //a map is only a middle man, so it is both readable and writable
  
  it(doubler).has({
    readable: true,
    writable: true,   
  })

  readStream(doubler, function (err, output) {
    it(output).deepEqual(input.map(function (j) {
      return j * 2
    }))
//    process.nextTick(x.validate)
    test.done()
  })
  
  writeArray(input, doubler)
  
}

exports ['stream comes back in the correct order'] = function (test) {
  var input = [3, 2, 1]

  var delayer = map(function(data, cb){
    setTimeout(function () {
      cb(null, data)
    }, 100 * data)
  })

  readStream(delayer, function (err, output) {
    it(output).deepEqual(input)
    test.done()
  })

  writeArray(input, delayer)
}

exports['pipe two maps together'] = function (test) {

  var input = [1,2,3,7,5,3,1,9,0,2,4,6]
  //create event stream from
  function dd (data, cb) {
    cb(null, data * 2)
  }
  var doubler1 = map(dd), doubler2 = map(dd)

  doubler1.pipe(doubler2)
  
  spec(doubler1).through().validateOnExit()
  spec(doubler2).through().validateOnExit()

  readStream(doubler2, function (err, output) {
{'fraction_non_alphanumeric': 0.15915221579961464, 'fraction_numerical': 0.015028901734104046, 'mean_word_length': 2.772529069767442, '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': '3276748', 'n_tokens_mistral': 1889, 'n_tokens_neox': 1810, 'n_words': 435}