wavelength_server/websocket_server/node_modules/from/test/index.js

var from = require('..')
var spec = require('stream-spec')
var a = require('assertions')

function read(stream, callback) {
  var actual = []
  stream.on('data', function (data) {
    actual.push(data)
  })
  stream.once('end', function () {
    callback(null, actual)
  })
  stream.once('error', function (err) {
    callback(err)
  })
}

function pause(stream) {
  stream.on('data', function () {
    if(Math.random() > 0.1) return
    stream.pause()
    process.nextTick(function () {
      stream.resume()
    })
  })
}

exports['inc'] = function (test) {

  var fs = from(function (i) {
    this.emit('data', i)
    if(i >= 99)
      return this.emit('end')
    return true
  })

  spec(fs).readable().validateOnExit() 

  read(fs, function (err, arr) {
    test.equal(arr.length, 100)
    test.done()
  })
}

exports['inc - async'] = function (test) {

  var fs = from(function (i, next) {
    this.emit('data', i)
    if(i >= 99)
      return this.emit('end')
		next();
  })

  spec(fs).readable().validateOnExit() 

  read(fs, function (err, arr) {
    test.equal(arr.length, 100)
    test.done()
  })
}

exports['large stream - from an array'] = function (test) {

  var l = 100000
    , expected = [] 

  while(l--) expected.push(l * Math.random())

  var fs = from(expected.slice())

  spec(fs).readable().validateOnExit() 

  read(fs, function (err, arr) {
		a.deepEqual(arr, expected)
    test.done()
  })
}

exports['large stream - callback return true'] = function (test) {

  var fs = from(function (i, next) {
    this.emit('data', i)
    if(i >= 99999)
      return this.emit('end')
		return true;
  })

  spec(fs).readable().validateOnExit() 

  read(fs, function (err, arr) {
    test.equal(arr.length, 100000)
    test.done()
  })
}

exports['large stream - callback call next()'] = function (test) {

  var fs = from(function (i, next) {
    this.emit('data', i)
    if(i >= 99999)
      return this.emit('end')
		next();
  })

  spec(fs).readable().validateOnExit() 

  read(fs, function (err, arr) {
    test.equal(arr.length, 100000)
    test.done()
  })
}

exports['simple'] = function (test) {

  var l = 1000
    , expected = [] 

  while(l--) expected.push(l * Math.random())

  var t = from(expected.slice())

  spec(t)
    .readable()
    .pausable({strict: true})
    .validateOnExit()

  read(t, function (err, actual) {
    if(err) test.error(err) //fail
    a.deepEqual(actual, expected)
    test.done()
  })

}

exports['simple pausable'] = function (test) {

  var l = 1000
    , expected = [] 

  while(l--) expected.push(l * Math.random())

  var t = from(expected.slice())

  spec(t)
    .readable()
    .pausable({strict: true})
    .validateOnExit()

  pause(t)

  read(t, function (err, actual) {
    if(err) test.error(err) //fail
    a.deepEqual(actual, expected)
    test.done()
  })

}

exports['simple (not strictly pausable) setTimeout'] = function (test) {

  var l = 10
    , expected = [] 
  while(l--) expected.push(l * Math.random())


  var _expected = expected.slice()
  var t = from(function (i, n) {
    var self = this
    setTimeout(function () {
      if(_expected.length)
        self.emit('data', _expected.shift())
      else
        if(!self.ended)
          self.emit('end')
      n()
    }, 3)
  })

  /*
    using from in this way will not be strictly pausable.
    it could be extended to buffer outputs, but I think a better
    way would be to use a PauseStream that implements strict pause.
  */

  spec(t)
    .readable()
    .pausable({strict: false })
    .validateOnExit()

  //pause(t)
  var paused = false
  var i = setInterval(function () {
    if(!paused) t.pause()
    else t.resume()
    paused = !paused
  }, 2)

  t.on('end', function () {
    clearInterval(i)
  })

  read(t, function (err, actual) {
    if(err) test.error(err) //fail
    a.deepEqual(actual, expected)
    test.done()
  })

}