filter/index.js

"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();
		}
	};
};