pull-stream via codata

pull-stream.ts

function pipe<A, B>(a: A, f: (a: A) => B): B;
function pipe<A, B, C>(a: A, f: (a: A) => B, g: (b: B) => C): C;
function pipe<A, B, C, D>(
	a: A,
	f: (a: A) => B,
	g: (b: B) => C,
	h: (c: C) => D
): D;
function pipe<A, B, C, D, E>(
	a: A,
	f: (a: A) => B,
	g: (b: B) => C,
	h: (c: C) => D,
	i: (d: D) => E
): E;
function pipe<A, B, C, D, E, F>(
	a: A,
	f: (a: A) => B,
	g: (b: B) => C,
	h: (c: C) => D,
	i: (d: D) => E,
	j: (e: E) => F
): F;
function pipe<A, B, C, D, E, F, G>(
	a: A,
	f: (a: A) => B,
	g: (b: B) => C,
	h: (c: C) => D,
	i: (d: D) => E,
	j: (e: E) => F,
	k: (f: F) => G
): G;
function pipe<A, B, C, D, E, F, G, H>(
	a: A,
	f: (a: A) => B,
	g: (b: B) => C,
	h: (c: C) => D,
	i: (d: D) => E,
	j: (e: E) => F,
	k: (f: F) => G,
	l: (g: G) => H
): H;
function pipe<A, B, C, D, E, F, G, H, I>(
	a: A,
	f: (a: A) => B,
	g: (b: B) => C,
	h: (c: C) => D,
	i: (d: D) => E,
	j: (e: E) => F,
	k: (f: F) => G,
	l: (g: G) => H,
	m: (h: H) => I
): I;
function pipe(a: any, ...fns: any[]): any {
	return fns.reduce((a, f) => f(a), a);
}

type Stream<T> = null | {
	head(): T;
	tail(): Stream<T>;
};

const countUp = (init: number, step = 1): Stream<number> => {
	return {
		head: () => init,
		tail: () => countUp(init + step),
	};
};

const filter = (predicate: (value: number) => boolean) => {
	return (stream: Stream<number>): Stream<number> => {
		if (stream === null) {
			return null;
		}

		const head = stream.head();

		if (predicate(head)) {
			return {
				head: () => head,
				tail: () => pipe(stream.tail(), filter(predicate)),
			};
		}

		return pipe(stream.tail(), filter(predicate));
	};
};

const takeUntil = <T>(predicate: (value: T) => boolean) => {
	return (stream: Stream<T>): Stream<T> => {
		if (stream === null) {
			return null;
		}

		const head = stream.head();

		if (predicate(head)) {
			return null;
		}

		return {
			head: () => head,
			tail: () => pipe(stream.tail(), takeUntil(predicate)),
		};
	};
};

const map = <T, U>(mapper: (value: T) => U) => {
	return (stream: Stream<T>): Stream<U> => {
		if (stream === null) {
			return null;
		}

		return {
			head: () => mapper(stream.head()),
			tail: () => pipe(stream.tail(), map(mapper)),
		};
	};
};

const take = <T>(total: number) => {
	const consume = (stream: Stream<T>, count = 0): Stream<T> => {
		if (stream === null) {
			return null;
		}

		if (count >= total) {
			return null;
		}

		return {
			head: () => stream.head(),
			tail: () => consume(stream.tail(), count + 1),
		};
	};

	return (stream: Stream<T>) => consume(stream, 0);
};

const reverse = <T>(source: Stream<T>, target: Stream<T> = null): Stream<T> => {
	if (source === null) {
		return target;
	}

	if (target === null) {
		return reverse(source.tail(), {
			head: () => source.head(),
			tail: () => null,
		});
	}

	return reverse(source.tail(), {
		head: () => source.head(),
		tail: () => target,
	});
};

const range = (start: number, end: number, step = 1): Stream<number> => {
	return pipe(
		countUp(start, step),
		takeUntil((value) => value > end)
	);
};

const forEach = <T>(stream: Stream<T>, callback: (value: T) => void): void => {
	if (stream === null) {
		return;
	}

	callback(stream.head());

	forEach(stream.tail(), callback);
};

const sift = (stream: Stream<number>): Stream<number> => {
	if (stream === null) {
		return null;
	}

	const head = stream.head();

	return {
		head: () => head,
		tail: () => pipe(stream.tail(), filter((n) => n % head !== 0), sift),
	};
};

let numbers = countUp(2);

const primes = pipe(numbers, sift);

const test = pipe(
	primes,
	take(20),
)


console.log(`primes:`);

forEach(test, (value) => console.log(value));


numbers = pipe(
  range(0, 10),
  map((value) => value * 2),
  reverse
);

console.log(`reverse:`);

forEach(numbers, console.log);