Add ap method and a simple test

CHANGELOG.md

@@ -5,6 +5,13 @@ This file does not aim to be comprehensive (you have git history for that),
 rather it lists changes that might impact your own code as a consumer of
 this library.
 
+3.0.0-beta.7
+-----
+
+### New additions
+* `ap` - Applies a stream of function(s) to the stream of value(s).
+  [#643](https://github.com/caolan/highland/pull/643).
+
 3.0.0-beta.6
 -----
 This release contains all changes from [2.12.0](#2120).

lib/index.js

@@ -2237,6 +2237,53 @@ addMethod('flatMap', function (f) {
     return this.map(f).sequence();
 });
 
+/**
+ * Creates a new Stream of values by applying each item in a Stream to each
+ * value in a Stream of functions
+ *
+ * @id ap
+ * @section Higher-order Streams
+ * @name Stream.ap(m)
+ * @param {Stream} m - incoming stream of function(s) to apply to value(s) in stream
+ * @api public
+ *
+ * var asyncUnit = () => _([]);
+ * var asyncAutocomplete = e => _(fetch(`/autocomplete?s=${e.target.value}`));
+ *
+ * var fns = _('change', checkbox)
+ *   .flatMap(e => _.of(e.target.checked ? asyncAutocomplete : asyncUnit));
+ *
+ * _('change', input)
+ *   .ap(fns)
+ *   .sequence()
+ *   .map(showAutocompleteResults);
+ */
+
+addMethod('ap', function(m) {
+    return _([
+        this.map(function (u1) {
+            return {
+                u: u1
+            };
+        }),
+        m.map(function (m1) {
+            return {
+                m: m1
+            };
+        }),
+    ])
+        .merge()
+        .scan1(function (x, y) {
+            return _.extend(y, x);
+        })
+        .filter(function (x) {
+            return x.u && x.m;
+        })
+        .map(function (x) {
+            return x.m(x.u);
+        });
+});
+
 /**
  * Retrieves values associated with a given property from all elements in
  * the collection.

test/test.js

@@ -4997,6 +4997,143 @@ exports['flatMap - map to Stream of Array'] = function (test) {
     });
 };
 
+exports.ap = {
+    setUp: function (callback) {
+        this.clock = sinon.useFakeTimers();
+        callback();
+    },
+    tearDown: function (callback) {
+        this.clock.restore();
+        callback();
+    },
+    'applies values to functions': function (test) {
+        var s = _([1, 2, 3, 4]);
+        var f = _.of(function doubled(x) {
+            return x * 2;
+        });
+
+        _.ap(f, s).toArray(function (xs) {
+            test.same(xs, [2, 4, 6, 8]);
+            test.done();
+        });
+    },
+    'noValueOnError': noValueOnErrorTest(_.ap(_.of(1))),
+    'ArrayStream': function (test) {
+        test.expect(1);
+        var f = _.of(function (x) {
+            return _(function (push, next) {
+                setTimeout(function () {
+                    push(null, x * 2);
+                    push(null, _.nil);
+                }, 10);
+            });
+        });
+        _([1, 2, 3, 4]).ap(f).merge().toArray(function (xs) {
+            test.same(xs, [2, 4, 6, 8]);
+        });
+        this.clock.tick(20);
+        test.done();
+    },
+    'GeneratorStream': function (test) {
+        test.expect(1);
+        var f = _.of(function (x) {
+            return _(function (push, next) {
+                push(null, x * 2);
+                push(null, _.nil);
+            });
+        });
+        var s = _(function (push, next) {
+            push(null, 1);
+            push(null, 2);
+            push(null, 3);
+            push(null, 4);
+            push(null, _.nil);
+        });
+        s.ap(f).merge().toArray(function (xs) {
+            test.same(xs, [2, 4, 6, 8]);
+            test.done();
+        });
+    },
+    'map to Stream of Array': function (test) {
+        test.expect(1);
+        var f = _.of(function (x) {
+            return _([[x]]);
+        });
+        var s = _([1, 2, 3, 4]).ap(f).merge().toArray(function (xs) {
+            test.same(xs, [[1], [2], [3], [4]]);
+            test.done();
+        });
+    },
+    'reflect timing of value and function arrival': function (test) {
+        test.expect(4);
+        var f = _(function (push, next) {
+            setTimeout(function () {
+                push(null, function (x) { return 'g1(' + x + ')'; });
+            }, 20);
+            setTimeout(function () {
+                push(null, function (x) { return 'g2(' + x + ')'; });
+                push(null, _.nil);
+            }, 60);
+        });
+        var s = _(function (push, next) {
+            setTimeout(function () {
+                push(null, 1);
+            }, 10);
+            setTimeout(function () {
+                push(null, 2);
+            }, 50);
+            setTimeout(function () {
+                push(null, 3);
+                push(null, _.nil);
+            }, 70);
+        });
+
+        var results = [];
+        s.ap(f).each(function (x) {
+            results.push(x);
+        });
+        this.clock.tick(20);
+        test.same(results, ['g1(1)']);
+        this.clock.tick(30);
+        test.same(results, ['g1(1)', 'g1(2)']);
+        this.clock.tick(10);
+        test.same(results, ['g1(1)', 'g1(2)', 'g2(2)']);
+        this.clock.tick(10);
+        test.same(results, ['g1(1)', 'g1(2)', 'g2(2)', 'g2(3)']);
+        test.done();
+    },
+    'composition': {
+        'v.ap(u.ap(a.map(f => g => x => f(g(x))))) is equivalent to v.ap(u).ap(a)': function (test) {
+            test.expect(3);
+            var v = _([1, 2, 3]);
+            var u = _.of(function (x) {
+                return 'u(' + x + ')';
+            });
+            var a = _.of(function (x) {
+                return 'a(' + x + ')';
+            });
+            var left = v.fork().ap(u.fork().ap(a.fork().map(function (f) {
+                return function (g) {
+                    return function (x) {
+                        return f(g(x));
+                    };
+                };
+            })));
+            var right = v.observe().ap(u.observe()).ap(a.observe());
+
+            _([left.collect(), right.collect()])
+                .sequence()
+                .apply(function (lefts, rights) {
+                    test.same(lefts, ['a(u(1))', 'a(u(2))', 'a(u(3))']);
+                    test.same(rights, ['a(u(1))', 'a(u(2))', 'a(u(3))']);
+                    test.same(lefts, rights);
+                });
+
+            test.done();
+        },
+    },
+};
+
 exports.pluck = function (test) {
     var a = _([
         {type: 'blogpost', title: 'foo'},