You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
|
|
|
"use strict";
|
|
|
|
|
|
|
|
const Promise = require("bluebird");
|
|
|
|
const propagateAbort = require("@promistream/propagate-abort");
|
|
|
|
const propagatePeek = require("@promistream/propagate-peek");
|
|
|
|
|
|
|
|
const { validateArguments } = require("@validatem/core");
|
|
|
|
const required = require("@validatem/required");
|
|
|
|
const isFunction = require("@validatem/is-function");
|
|
|
|
|
|
|
|
module.exports = function mapStream(_filterPredicate) {
|
|
|
|
let [ filterPredicate ] = validateArguments(arguments, {
|
|
|
|
filterPredicate: [ required, isFunction ]
|
|
|
|
});
|
|
|
|
|
|
|
|
return {
|
|
|
|
_promistreamVersion: 0,
|
|
|
|
description: `filter stream`,
|
|
|
|
abort: propagateAbort,
|
|
|
|
peek: propagatePeek,
|
|
|
|
read: function produceValue_mapStream(source) {
|
|
|
|
function attemptRead() {
|
|
|
|
return Promise.try(() => {
|
|
|
|
return source.read();
|
|
|
|
}).then((result) => {
|
|
|
|
return Promise.try(() => {
|
|
|
|
return filterPredicate(result);
|
|
|
|
}).then((itemPassed) => {
|
|
|
|
if (itemPassed) {
|
|
|
|
return result;
|
|
|
|
} else {
|
|
|
|
return attemptRead();
|
|
|
|
}
|
|
|
|
});
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
return attemptRead();
|
|
|
|
}
|
|
|
|
};
|
|
|
|
};
|