simple-source

One of the most commonly useful source stream implementations for Promistreams, and meant as a building block for both custom stream libraries and application code.

This library will provide you with a spec-compliant source stream, and the only thing you need to provide is a callback that produces a new value whenever called.

Usage

Making your own custom source stream (which in this case counts from 1 to 10):

const simpleSource = require("simple-source");

function generateNumbers() {
	let i = 0;
	let max = 10;

	return simpleSource(() => {
		if (i < max) {
			i += 1;
			return i;
		} else {
			throw new EndOfStream();
		}
	});
}

// ...

let numbers = await pipe([
	generateNumbers(),
	collect()
]).read(); // numbers = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]

... or using it inline for one-off cases, like turning values provided by some external API into a Promistream:

const simpleSource = require("simple-source");

let values = await pipe([
	simpleSource(async () => {
		return someExternalSource.getTheNextValueSomehow();
	}),
	collect()
]).read(); // numbers = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]

... or if you also need to do something special when the stream is aborted, like disconnecting from such an API:

const simpleSource = require("simple-source");

let values = await pipe([
	simpleSource({
		onValue: async () => {
			return someExternalSource.getTheNextValueSomehow();
		},
		onAbort: async () => {
			return someExternalSource.closeConnection();
		}
	}),
	collect()
]).read(); // numbers = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]

Concurrency

The onValue callback that you specify must be able to deal with concurrency - it may be called a second time (or even many more times) before the first call has finished producing a value.

If your callback cannot do this, then you must combine the source stream with a sequentialize stream to ensure that only one value is requested from it at a time, as described in the Promistreams documentation. (FIXME: Add link)

API

simpleSource(options)

Creates a new 'simple source' stream.

• options: An object with options. If you specify a function instead of an object, then it's assumed to be the onValue option and no other options can be specified.
  • onValue: Required. An asynchronous callback that, when called, provides the next value for the stream.

FIXME: Rest of documentation