simple-sink/index.js

"use strict";

const Promise = require("bluebird");
const propagateAbort = require("@ppstreams/propagate-abort");
const { isEndOfStream } = require("@ppstreams/end-of-stream-marker");

module.exports = function greedySinkStream(description, callback) {
	return {
		description: `greedy sink stream (${description})`,
		abort: propagateAbort,
		read: function produceValue_greedySinkStream(source) {
			let lastResult;

			function attemptRead() {
				return Promise.try(() => {
					return source.read();
				}).then((value) => {
					return callback(value);
				}).then((result) => {
					lastResult = result;

					return attemptRead();
				}).catch(isEndOfStream, () => {
					/* Don't attempt to do another read, we're done. We return whatever value we got last from the specified callback. */
					return lastResult;
				});
			}

			return attemptRead();
		}
	};
};