Package Exports

pull-fn-stream

This package does not declare an exports field, so the exports above have been automatically detected and optimized by JSPM instead. If any package subpath is missing, it is recommended to post an issue to the original package (pull-fn-stream) to support the "exports" field. If that is not possible, create a JSPM override to customize the exports field for this package.

Readme

pull fn stream

Model an async function call with a pull stream. This is a transform stream that takes arrays and emits streams. Each returned stream emits a start event and a response event.

install

$ npm install pull-fn-stream

example

var S = require('pull-stream')
var flatMerge = require('pull-flat-merge')
var test = require('tape')
var toStream = require('../')

var fns = {
    a: function (cb) {
        setTimeout(function () {
            cb(null, 'a')
        }, 50)
    },
    b: function (cb) {
        setTimeout(function () {
            cb(null, 'b')
        }, 100)
    },
    error: function (cb) {
        setTimeout(function () {
            cb(new Error('test error'))
        }, 0)
    }
}

function asyncOk (arg, cb) {
    process.nextTick(function () {
        cb(null, arg)
    })
}

function asyncErr (msg, cb) {
    process.nextTick(function () {
        cb(new Error(msg))
    })
}

test('call async function', function (t) {
    t.plan(2)
    var stream = toStream(asyncOk, 'key')
    var expected = [
        { type: 'start', op: 'key' },
        { type: 'start', op: 'key' },
        { type: 'key', resp: 'arg' },
        { type: 'key', resp: 'arg2' }
    ]
    S(
        S.values(['arg', 'arg2']),
        stream(),
        flatMerge(),
        S.collect(function (err, res) {
            t.error(err, 'error')
            t.deepEqual(res, expected, 'should emit events')
        })
    )
})

test('async error', function (t) {
    t.plan(1)
    var stream = toStream(asyncErr, 'key')
    S(
        S.once('test msg'),
        stream(),
        flatMerge(),
        S.collect(function (err, res) {
            t.equal(err.message, 'test msg', 'should end with error')
        })
    )
})

test('from object', function (t) {
    t.plan(2)
    var expected = [
        { type: 'start', op: 'b' },
        { type: 'start', op: 'a' },
        { type: 'start', op: 'b' },
        { type: 'a', resp: 'a' },
        { type: 'b', resp: 'b' },
        { type: 'b', resp: 'b' }
    ]
    S(
        S.values(['b', 'a', 'b']),
        toStream.fromObject(fns)(),
        flatMerge(),
        S.collect(function (err, evs) {
            t.error(err, 'error')
            t.deepEqual(evs, expected, 'should emit events in order')
        })
    )
})

test('pass arguments', function (t) {
    var fns = {
        a: function (a,b,c,cb) {
            setTimeout(function () {
                cb(null, [a,b,c])
            }, 50)
        },
        b: function (d,cb) {
            setTimeout(function () {
                cb(null, d)
            }, 100)
        }
    }

    var expected = [
        { type: 'start', op: 'b' },
        { type: 'start', op: 'a' },
        { type: 'a', resp: [1,2,3] },
        { type: 'b', resp: 4 }
    ]

    t.plan(1)
    S(
        S.values([ ['b',4], ['a',1,2,3] ]),
        toStream.fromObject(fns)(),
        flatMerge(),
        S.collect(function (err, evs) {
            t.deepEqual(evs, expected, 'should pass args in array')
        })
    )
})

test('error from object', function (t) {
    t.plan(1)
    S(
        S.values(['error']),
        toStream.fromObject(fns)(),
        flatMerge(),
        S.collect(function (err, res) {
            t.equal(err.message, 'test error', 'should pass error in stream')
        })
    )
})