Skip to content

Implement flatTap #651

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tomwhale wants to merge 8 commits into
base: master
Choose a base branch
from
Open

Implement flatTap #651

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9b41102
Implement flatMap
Jun 13, 2018
2d8a712
Tests for flatTap
Jun 13, 2018
7172f19
Same stream test for flatTap
Jun 13, 2018
1a2fb23
flatTap - noValueOnError test
Jun 14, 2018
d3ee895
flatTap - More tests for each stream source.
Jun 14, 2018
768f6c2
Update tests & use consume instead
Oct 22, 2018
18a7add
FlatTap - Improve tests & Use consume rather than just FlatMap
Oct 24, 2018
689c356
FlatTap - Push errors back onto the stream, and respective tests
Oct 25, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions lib/index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2167,6 +2167,44 @@ addMethod('doto', function (f) {
Stream.prototype.tap = Stream.prototype.doto;
_.tap = _.doto;

/**
* Applies a function, which must return a (possibly empty) highland stream, to each value from the source, and re-emits the
* source value when the function return ends.
*
* @id flatTap
* @section Transforms
* @name Stream.flatTap(f)
* @param {Function} f - the function to apply
* @api public
*
* const httpLog = _(new Promnise(res => {
* setTimeout(() => {
* console.log('Log');
* res('Success')
* }, 3000)
* }))
*
* _([1, 2, 3]).flatTap(httpLog) // Will emit [1,2,3]
*/

addMethod('flatTap', function (f) {
return this.flatMap(function (x) {
return f(x).consume(function (err, y, push, next) {
if (err) {
push(err);
push(null, _.nil);
}
else if (y === _.nil) {
push(null, x);
push(null, _.nil);
}
else {
next();
}
});
});
});

/**
* Limits number of values through the stream to a maximum of number of values
* per window. Errors are not limited but allowed to pass through as soon as
Expand Down
111 changes: 111 additions & 0 deletions test/test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4909,6 +4909,117 @@ exports['tap - doto alias'] = function (test) {
test.done();
};


exports.flatTap = {
'flatTap - noValueOnError': noValueOnErrorTest(_.flatTap(function (x) { return _(); })),
'flatTap - returnsSameStream': returnsSameStreamTest(function (s) {
return s.flatTap(function (x) { return _([2]); });
}, [1], [1]),
'flatTap - On a empty stream': function (test) {
test.expect(1);
var s = _([]).flatTap(function (x) {
return _([6]);
});

s.pull(valueEquals(test, _.nil));
test.done();
},
'flatTap - Returns an Empty stream': function (test) {
test.expect(1);
var s = _([3]).flatTap(function (x) {
return _([]);
});

s.pull(valueEquals(test, 3));
test.done();
},
'flatTap - Emits an multiple values': function (test) {
test.expect(1);
var s = _([3]).flatTap(function (x) {
return _([5, 6, 7, 8]);
});

s.pull(valueEquals(test, 3));
test.done();
},
'flatTap - Emits an error values': function (test) {
test.expect(4);
var s = _([1, 2, 3, 4]).flatTap(function (x) {
return _(function (push) {
push({errorMessage: 'error'});
push(null, _.nil);
});
});

s.pull(anyError(test));
s.pull(anyError(test));
s.pull(anyError(test));
s.pull(anyError(test));
test.done();
},
'flatTap - argument function throws': function (test) {
Copy link
Collaborator

@vqvu vqvu Oct 25, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you add a similar test to this, but for the stream (i.e., the return value of the argument function) emitting an error instead.

Something like this,

_([1, 2, 3, 4]).flatTap((x) => _((push) => {
    push(err);
    push(null, _.nil);
}));

test.expect(4);
var err = new Error('error');
var s = _([1, 2, 3, 4]).flatTap(function (x) {
if (x === 1) { throw err; }
if (x === 2) { _(['hello']); }
if (x === 3) { throw err; }
return _(['world']);
});

s.pull(errorEquals(test, 'error'));
s.pull(valueEquals(test, 2));
s.pull(anyError(test));
s.pull(valueEquals(test, 4));
test.done();
},
'flatTap - ArrayStream': function (test) {
var seen = [];
var f = function (x) {
return _(function (push, next) {
setTimeout(function () {
var y = x * 2;
seen.push(y);
push(null, y);
push(null, _.nil);
}, 10);
});
};
_([1, 2, 3, 4]).flatTap(f).toArray(function (xs) {
test.same(xs, [1, 2, 3, 4]);
test.same(seen, [2, 4, 6, 8]);
test.done();
});
},
'flatTap - GeneratorStream': function (test) {
var seen = [];
var f = function (x) {
return _(function (push, next) {
setTimeout(function () {
var y = x * 2;
seen.push(y);
push(null, y);
push(null, _.nil);
}, 10);
});
};
var s = _(function (push, next) {
push(null, 1);
setTimeout(function () {
push(null, 2);
push(null, 3);
push(null, 4);
push(null, _.nil);
}, 10);
});
s.flatTap(f).toArray(function (xs) {
test.same(xs, [1, 2, 3, 4]);
test.same(seen, [2, 4, 6, 8]);
test.done();
});
}
};

exports.flatMap = function (test) {
var f = function (x) {
return _(function (push, next) {
Expand Down