.npmignore000064400000000015152046567360006555 0ustar00node_modules
LICENSE000064400000002067152046567360005574 0ustar00The MIT License (MIT)

Copyright (c) 2015 Mathias Buus

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
.travis.yml000064400000000076152046567360006676 0ustar00language: node_js
node_js:
  - '0.10'
  - '0.12'
  - 'stable'
README.md000064400000001764152046567360006051 0ustar00# stream-each

Iterate all the data in a stream

```
npm install stream-each
```

[![build status](http://img.shields.io/travis/mafintosh/stream-each.svg?style=flat)](http://travis-ci.org/mafintosh/stream-each)

## Usage

``` js
var each = require('stream-each')

each(stream, function (data, next) {
  console.log('data from stream', data)
  // when ready to consume next chunk
  next()
}, function (err) {
  console.log('no more data')
})
```

## API

#### `each(stream, iterator, cb)`

Iterate the data in the stream by calling the iterator function with `(data, next)`
where data is a data chunk and next is a callback. Call next when you are ready to
consume the next chunk. Optionally you can call next with an error to destroy the stream

When the stream ends/errors the callback is called if provided

## License

MIT

## Related

`stream-each` is part of the [mississippi stream utility collection](https://github.com/maxogden/mississippi) which includes more useful stream modules similar to this one.
package.json000064400000003152152046567360007051 0ustar00{
  "_from": "stream-each@^1.1.0",
  "_id": "stream-each@1.2.2",
  "_inBundle": false,
  "_integrity": "sha512-mc1dbFhGBxvTM3bIWmAAINbqiuAk9TATcfIQC8P+/+HJefgaiTlMn2dHvkX8qlI12KeYKSQ1Ua9RrIqrn1VPoA==",
  "_location": "/stream-each",
  "_phantomChildren": {},
  "_requested": {
    "type": "range",
    "registry": true,
    "raw": "stream-each@^1.1.0",
    "name": "stream-each",
    "escapedName": "stream-each",
    "rawSpec": "^1.1.0",
    "saveSpec": null,
    "fetchSpec": "^1.1.0"
  },
  "_requiredBy": [
    "/mississippi",
    "/npm-profile/cacache/mississippi",
    "/npm-profile/mississippi",
    "/npm-registry-fetch/cacache/mississippi"
  ],
  "_resolved": "https://registry.npmjs.org/stream-each/-/stream-each-1.2.2.tgz",
  "_shasum": "8e8c463f91da8991778765873fe4d960d8f616bd",
  "_spec": "stream-each@^1.1.0",
  "_where": "/Users/rebecca/code/npm/node_modules/mississippi",
  "author": {
    "name": "Mathias Buus",
    "url": "@mafintosh"
  },
  "bugs": {
    "url": "https://github.com/mafintosh/stream-each/issues"
  },
  "bundleDependencies": false,
  "dependencies": {
    "end-of-stream": "^1.1.0",
    "stream-shift": "^1.0.0"
  },
  "deprecated": false,
  "description": "Iterate all the data in a stream",
  "devDependencies": {
    "standard": "^5.3.1",
    "tape": "^4.2.1",
    "through2": "^2.0.0"
  },
  "homepage": "https://github.com/mafintosh/stream-each",
  "license": "MIT",
  "main": "index.js",
  "name": "stream-each",
  "repository": {
    "type": "git",
    "url": "git+https://github.com/mafintosh/stream-each.git"
  },
  "scripts": {
    "test": "standard && tape test.js"
  },
  "version": "1.2.2"
}
collaborators.md000064400000000540152046567360007751 0ustar00## Collaborators

stream-each is only possible due to the excellent work of the following collaborators:

<table><tbody><tr><th align="left">maxogden</th><td><a href="https://github.com/maxogden">GitHub/maxogden</a></td></tr>
<tr><th align="left">mafintosh</th><td><a href="https://github.com/mafintosh">GitHub/mafintosh</a></td></tr>
</tbody></table>
test.js000064400000004037152046567360006103 0ustar00var tape = require('tape')
var through = require('through2')
var each = require('./')

tape('each', function (t) {
  var s = through.obj()
  s.write('a')
  s.write('b')
  s.write('c')
  s.end()

  s.on('end', function () {
    t.end()
  })

  var expected = ['a', 'b', 'c']
  each(s, function (data, next) {
    t.same(data, expected.shift())
    next()
  })
})

tape('each and callback', function (t) {
  var s = through.obj()
  s.write('a')
  s.write('b')
  s.write('c')
  s.end()

  var expected = ['a', 'b', 'c']
  each(s, function (data, next) {
    t.same(data, expected.shift())
    next()
  }, function () {
    t.end()
  })
})

tape('each (write after)', function (t) {
  var s = through.obj()
  s.on('end', function () {
    t.end()
  })

  var expected = ['a', 'b', 'c']
  each(s, function (data, next) {
    t.same(data, expected.shift())
    next()
  })

  setTimeout(function () {
    s.write('a')
    s.write('b')
    s.write('c')
    s.end()
  }, 100)
})

tape('each error', function (t) {
  var s = through.obj()
  s.write('hello')
  s.on('error', function (err) {
    t.same(err.message, 'stop')
    t.end()
  })

  each(s, function (data, next) {
    next(new Error('stop'))
  })
})

tape('each error and callback', function (t) {
  var s = through.obj()
  s.write('hello')

  each(s, function (data, next) {
    next(new Error('stop'))
  }, function (err) {
    t.same(err.message, 'stop')
    t.end()
  })
})

tape('each with falsey values', function (t) {
  var s = through.obj()
  s.write(0)
  s.write(false)
  s.write(undefined)
  s.end()

  s.on('end', function () {
    t.end()
  })

  var expected = [0, false]
  var count = 0
  each(s, function (data, next) {
    count++
    t.same(data, expected.shift())
    next()
  }, function () {
    t.same(count, 2)
  })
})

tape('huge stack', function (t) {
  var s = through.obj()

  for (var i = 0; i < 5000; i++) {
    s.write('foo')
  }

  s.end()

  each(s, function (data, cb) {
    if (data !== 'foo') t.fail('bad data')
    cb()
  }, function (err) {
    t.error(err, 'no error')
    t.end()
  })
})
index.js000064400000001773152046567360006237 0ustar00var eos = require('end-of-stream')
var shift = require('stream-shift')

module.exports = each

function each (stream, fn, cb) {
  var want = true
  var error = null
  var ended = false
  var running = false
  var calling = false

  stream.on('readable', onreadable)
  onreadable()

  if (cb) eos(stream, {readable: true, writable: false}, done)
  return stream

  function done (err) {
    if (!error) error = err
    ended = true
    if (!running) cb(error)
  }

  function onreadable () {
    if (want) read()
  }

  function afterRead (err) {
    running = false

    if (err) {
      error = err
      if (ended) return cb(error)
      stream.destroy(err)
      return
    }
    if (ended) return cb(error)
    if (!calling) read()
  }

  function read () {
    while (!running && !ended) {
      want = false

      var data = shift(stream)
      if (data === null) {
        want = true
        return
      }

      running = true
      calling = true
      fn(data, afterRead)
      calling = false
    }
  }
}