simple-source/index.js

"use strict";

const Promise = require("bluebird");
const Aborted = require("@ppstreams/aborted");

const { validateOptions } = require("@validatem/core");
const required = require("@validatem/required");
const isFunction = require("@validatem/is-function");
const wrapValueAsOption = require("@validatem/wrap-value-as-option");

module.exports = function simpleSource(_options) {
	let { onRequest, onAbort } = validateOptions(arguments, [
		required,
		wrapValueAsOption("onResult"), {
			onRequest: [ required, isFunction ],
			onAbort: [ isFunction ]
		}
	]);

	let errorReason;
	let peekQueue = [];

	return {
		description: `simple source`,
		peek: function peekValue_simpleSource() {
			return Promise.try(() => {
				if (errorReason != null) {
					return true;
				} else {
					// FIXME: Store Promises in the peekQueue instead? Or would this make it more difficult to deal with concurrent peeks/reads? And should the peek spec be changed to account for this?
					return Promise.try(() => {
						return onRequest();
					}).then((result) => {
						peekQueue.push({ type: "value", value: result });
					}).catch((error) => {
						peekQueue.push({ type: "error", error: error });
					}).then(() => {
						// FIXME: What if there's an EndOfStream marker being produced? Or an abort? Or some other sort of failure? Shouldn't that return/produce something *other than* `true`?
						return true;
					});
				}
			});
		},
		read: function produceValue_simpleSource() {
			return Promise.try(() => {
				if (peekQueue.length > 0) {
					/* FIXME: Move all this logic out into an itemBuffer abstraction of some sort (also useful in from-node-stream?) */
					let item = peekQueue.shift();

					if (item.type === "value") {
						return item.value;
					} else if (item.type === "error") {
						throw item.error;
					} else {
						throw new Error(`Found a queue item of type '${item.type}'; this is a bug in @ppstreams/simple-source, please report it`);
					}
				} else {
					if (errorReason != null) {
						if (errorReason === true) {
							throw new Aborted("Stream was aborted");
						} else if (errorReason instanceof Error) {
							throw new Aborted(`Stream was aborted due to error: ${errorReason.message}`, {reason: errorReason});
						}
					} else {
						return onRequest();
					}
				}
			});
		},
		abort: function abort_simpleSource(reason) {
			return Promise.try(() => {
				if (errorReason == null) {
					if (reason === true || reason instanceof Error) {
						errorReason = reason;

						if (onAbort != null) {
							return onAbort(reason);
						}
					} else {
						throw new Error("You must specify a reason (either `true` or an Error object) when aborting a stream");
					}
				} else {
					// FIXME: Require this behaviour in the spec? Or is there a composability-related reason to permit/require quietly ignoring this, to make it idempotent?
					throw new Error(`The stream is already aborted`);
				}
			});
		}
	};
};