init

.gitignore

@@ -0,0 +1,7 @@
+
+.DS_Store*
+*.log
+*.gz
+
+node_modules
+coverage

.travis.yml

@@ -0,0 +1,7 @@
+node_js:
+- "0.8"
+- "0.10"
+- "0.11"
+language: node_js
+script: "npm run-script test-travis"
+after_script: "npm install coveralls@2 && cat ./coverage/lcov.info | coveralls"

LICENSE

@@ -0,0 +1,22 @@
+
+The MIT License (MIT)
+
+Copyright (c) 2014 Jonathan Ong [email protected]
+
+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.

README.md

@@ -0,0 +1,108 @@
+
+# promise-transform-streams
+
+[![NPM version][npm-image]][npm-url]
+[![Build status][travis-image]][travis-url]
+[![Test coverage][coveralls-image]][coveralls-url]
+[![Dependency Status][david-image]][david-url]
+[![License][license-image]][license-url]
+[![Downloads][downloads-image]][downloads-url]
+[![Gittip][gittip-image]][gittip-url]
+
+Creates promise-based transform streams.
+`.map()` creates a stream that converts objects to promises.
+`.resolve()` resolves all these promises.
+
+The main use-case for this is to create "concurrent" transform that preserve order.
+This module does not apply any concurrent limits!
+However, you could play with the buffer sizes,
+which indirectly relate to concurrency in this context.
+
+## Example
+
+Suppose you have MongoDB documents that look like this:
+
+```js
+var post = {
+  creator_id: ObjectId('123412341234123412341234'),
+  title: 'some title',
+  markdown: 'blah'
+}
+```
+
+You want to grab all the `.creator`s, but you still want to stream all the results.
+Here's how to do it:
+
+```js
+var PStream = require('promise-transform-stream')
+
+db.posts.find({}).stream()
+// get the creator of each post, returns a stream of promises
+.pipe(PStream.map(getCreator))
+// resolves each promise in the stream in order
+.pipe(PStream.resolve())
+// no we can serialize the results and send it to the client!
+.pipe(JSONStream.stringify())
+.pipe(zlib.createGzip())
+.pipe(res)
+
+function getCreator(post) {
+  return new Promise(function (resolve, reject) {
+    db.users.findOne({
+      _id: post.creator_id
+    }, function (err, user) {
+      if (err) return reject(err)
+      post.creator = user
+      resolve(post)
+    })
+  })
+}
+```
+
+To be more efficient, you'd probably want to query each `creator_id` once per request.
+
+## API
+
+### .create(transform, [flush])
+
+Create a new `Transform` constructor with `._transform` and `._flush` methods.
+`transform` should resolve to the mapped document.
+`flush` should not have any arguments.
+
+```js
+var Transform = PStream.create(function (doc) {
+  return Promise.resolve(doc)
+})
+
+db.posts.find({}).stream()
+.pipe(Transform())
+```
+
+### .map(transform, [flush])
+
+The same as `.create()`, except it returns the transform instance immediately,
+[through2](https://www.npmjs.com/package/through2)-style.
+This is not recommended!
+
+### .resolve([options])
+
+Returns a resolving transform stream.
+
+[gitter-image]: https://badges.gitter.im/stream-utils/promise-transform-streams.png
+[gitter-url]: https://gitter.im/stream-utils/promise-transform-streams
+[npm-image]: https://img.shields.io/npm/v/promise-transform-streams.svg?style=flat-square
+[npm-url]: https://npmjs.org/package/promise-transform-streams
+[github-tag]: http://img.shields.io/github/tag/stream-utils/promise-transform-streams.svg?style=flat-square
+[github-url]: https://github.com/stream-utils/promise-transform-streams/tags
+[travis-image]: https://img.shields.io/travis/stream-utils/promise-transform-streams.svg?style=flat-square
+[travis-url]: https://travis-ci.org/stream-utils/promise-transform-streams
+[coveralls-image]: https://img.shields.io/coveralls/stream-utils/promise-transform-streams.svg?style=flat-square
+[coveralls-url]: https://coveralls.io/r/stream-utils/promise-transform-streams
+[david-image]: http://img.shields.io/david/stream-utils/promise-transform-streams.svg?style=flat-square
+[david-url]: https://david-dm.org/stream-utils/promise-transform-streams
+[license-image]: http://img.shields.io/npm/l/promise-transform-streams.svg?style=flat-square
+[license-url]: LICENSE
+[downloads-image]: http://img.shields.io/npm/dm/promise-transform-streams.svg?style=flat-square
+[downloads-url]: https://npmjs.org/package/promise-transform-streams
+[gittip-image]: https://img.shields.io/gratipay/jonathanong.svg?style=flat-square
+[gittip-url]: https://gratipay.com/jonathanong/

index.js

@@ -0,0 +1,65 @@
+
+var Transform = require('readable-stream/transform')
+var Promise = require('native-or-bluebird')
+var inherits = require('util').inherits
+var assert = require('assert')
+
+/**
+ * Create a new transform constructor.
+ */
+
+exports.create = function (transform, flush) {
+  assert(typeof transform === 'function')
+
+  inherits(Stream, Transform)
+
+  function Stream(options) {
+    if (!(this instanceof Stream)) return new Stream(options)
+    Transform.call(this, options)
+    // always object mode!
+    this._readableState.objectMode =
+    this._writableState.objectMode = true
+  }
+
+  Stream.prototype._transform = function (doc, NULL, cb) {
+    cb(null, Promise.resolve(transform.call(this, doc)))
+  }
+
+  if (typeof flush === 'function') {
+    // TODO: assert that flush doesn't have any arguments
+    Stream.prototype._flush = function (cb) {
+      Promise.resolve(flush.call(this)).then(cb.bind(null, null), cb)
+    }
+  }
+
+  return Stream
+}
+
+/**
+ * Create a new transform instance, `through`-style.
+ * NOT RECOMMENDED!
+ */
+
+exports.map = function (transform, flush) {
+  return exports.create(transform, flush)()
+}
+
+/**
+ * Create a transform stream that resolves all the promises.
+ */
+
+exports.resolve = Resolve
+
+inherits(Resolve, Transform)
+
+function Resolve(options) {
+  if (!(this instanceof Resolve)) return new Resolve(options)
+  Transform.call(this, options)
+  // always object mode!
+  this._readableState.objectMode =
+  this._writableState.objectMode = true
+}
+
+Resolve.prototype._transform = function (doc, NULL, cb) {
+  Promise.resolve(doc).then(cb.bind(null, null)).catch(cb)
+}

package.json

@@ -0,0 +1,36 @@
+{
+  "name": "promise-transform-streams",
+  "description": "Create and resolve a promise of streams",
+  "version": "0.0.0",
+  "author": "Jonathan Ong <[email protected]> (http://jongleberry.com)",
+  "license": "MIT",
+  "repository": "stream-utils/promise-transform-streams",
+  "dependencies": {
+    "native-or-bluebird": "1",
+    "readable-stream": "1"
+  },
+  "devDependencies": {
+    "bluebird": "2",
+    "istanbul": "0",
+    "mocha": "2",
+    "stream-to-array": "2"
+  },
+  "scripts": {
+    "test": "mocha --reporter spec",
+    "test-cov": "istanbul cover node_modules/mocha/bin/_mocha -- --reporter dot",
+    "test-travis": "istanbul cover node_modules/mocha/bin/_mocha --report lcovonly -- --reporter dot"
+  },
+  "keywords": [
+    "streams",
+    "stream",
+    "concurrency",
+    "promises",
+    "promise",
+    "then",
+    "resolve",
+    "concurrent"
+  ],
+  "files": [
+    "index.js"
+  ]
+}

test/test.js

@@ -0,0 +1,39 @@
+
+var Promise = require('native-or-bluebird')
+var toArray = require('stream-to-array')
+var assert = require('assert')
+
+var PStream = require('..')
+
+it('should compose', function () {
+  var stream = PStream.map(function (doc) {
+    return new Promise(function (resolve) {
+      process.nextTick(function () {
+        doc.asdf = true
+        resolve(doc)
+      })
+    })
+  }, function () {
+    this.push({
+      c: 3,
+      asdf: true
+    })
+  })
+
+  stream.write({
+    a: 1
+  })
+
+  stream.write({
+    b: 2
+  })
+
+  stream.end()
+
+  return toArray(stream.pipe(PStream.resolve())).then(function (arr) {
+    assert(arr.length === 3)
+    arr.forEach(function (arr) {
+      assert(arr.asdf)
+    })
+  })
+})