gen.re

type generator 'a =
  | End
  | Next 'a (unit => generator 'a);

let rec intsFrom ::start :generator int => Next start (fun () => intsFrom start::(start + 1));

let ints = intsFrom start::0;

let rec tap fn gen =>
  switch gen {
  | End => End
  | Next value next =>
    fn value;
    Next value (fun () => tap fn (next ()))
  };

let rec forEach fn gen =>
  switch gen {
  | End => ()
  | Next value next =>
    fn value;
    forEach fn (next ())
  };

let rec take n gen =>
  switch n {
  | 0 => End
  | n =>
    switch gen {
    | End => End
    | Next value rest => Next value (fun () => take (n - 1) (rest ()))
    }
  };

let log = forEach Js.log;

let rec map fn gen =>
  switch gen {
  | End => End
  | Next value rest => Next (fn value) (fun () => map fn (rest ()))
  };

let rec filter fn gen =>
  switch gen {
  | End => End
  | Next value rest =>
    let next () => filter fn (rest ());
    fn value ? Next value next : next ()
  };

let rec concat b a =>
  switch a {
  | End => b
  | Next value rest => Next value (fun () => concat b (rest ()))
  };

let empty = End;

let singleton a => Next a (fun () => empty);

let rec flatten nested =>
  switch nested {
  | End => End
  | Next gen gens =>
    switch gen {
    | End => flatten (gens ())
    | Next value rest => Next value (fun () => flatten (Next (rest ()) gens))
    }
  };

let rec fromList l =>
  switch l {
  | [] => End
  | [value, ...rest] => Next value (fun () => fromList rest)
  };

let chain fn gen => gen |> map fn |> flatten;

let start = fromList [1, 2, 3];

let even n => n mod 2 == 0;

ints |> filter even |> take 5 |> log;

gen.ts

function flatten<A>(stream: Stream<Stream<A>>): Stream<A> {
    const next = stream.next();
    if (!next) return new EmptyStream();
    return new FlattenStream(next.value, next.rest);
}

abstract class Stream<A> {
    abstract next(): { value: A, rest: Stream<A> } | null;

    map<B>(fn: (a: A) => B): Stream<B> {
        return new MappedStream(fn, this);
    }

    then<B>(fn: (a: A) => Stream<B>): Stream<B> {
        return flatten(this.map(fn));
    }

    take(n: number): Stream<A> {
        return new TakeStream(this, n);
    }

    concat(stream: Stream<A>): Stream<A> {
        return new ConcatStream(this, stream);
    }

    filter(fn: (a: A) => boolean): Stream<A> {
        return new FilteredStream(fn, this);
    }

    forEach(fn: (a: A) => void): void {
        const next = this.next();
        if (next) {
            const { value, rest } = next;
            fn(value);
            rest.forEach(fn);
        }
    }
}

class EmptyStream<A> extends Stream<A> {
    next() {
        return null;
    }
}

class MappedStream<A, B> extends Stream<B> {
    innerStream: Stream<A>;
    fn: (a: A) => B;

    constructor(fn: (a: A) => B, stream: Stream<A>) {
        super();
        this.fn = fn;
        this.innerStream = stream;
    }

    next() {
        const innerNext = this.innerStream.next();
        if (innerNext === null) return null;
        return {
            value: this.fn(innerNext.value),
            rest: new MappedStream(this.fn, innerNext.rest),
        };
    }
}

class TakeStream<A> extends Stream<A> {
    left: number;
    stream: Stream<A>;

    constructor(stream: Stream<A>, left: number) {
        super();
        this.left = left;
        this.stream = stream;
    }

    next() {
        if (this.left === 0) return null;
        const next = this.stream.next();
        if (next === null) return null;
        return {
            value: next.value,
            rest: new TakeStream(next.rest, this.left - 1),
        };
    }
}

class ConcatStream<A> extends Stream<A> {
    conc: Stream<A>;
    base: Stream<A>;

    constructor(stream: Stream<A>, conc: Stream<A>) {
        super();
        this.base = stream;
        this.conc = conc;
    }

    next() {
        const next = this.base.next();
        if (next === null) return this.conc.next();
        return {
            value: next.value,
            rest: new ConcatStream(next.rest, this.conc),
        };
    }
}

class FlattenStream<A> extends Stream<A> {
    first: Stream<A>;
    rest: Stream<Stream<A>>;

    constructor(first: Stream<A>, rest: Stream<Stream<A>>) {
        super();
        this.first = first;
        this.rest = rest;
    }

    next() {
        const firstNext = this.first.next();
        if (firstNext) {
            return {
                value: firstNext.value,
                rest: new FlattenStream(firstNext.rest, this.rest),
            };
        }

        let restNext = this.rest.next();
        while (restNext !== null) {
            const valNext = restNext.value.next();
            if (valNext) {
                return {
                    value: valNext.value,
                    rest: new FlattenStream(valNext.rest, restNext.rest),
                };
            }
            restNext = restNext.rest.next();    
        }

        return null;
    }
}

class FilteredStream<A> extends Stream<A> {
    base: Stream<A>;
    fn: (a: A) => boolean;

    constructor(fn: (a: A) => boolean, base: Stream<A>) {
        super();
        this.base = base;
        this.fn = fn;
    }

    next() {
        let next = this.base.next();
        while (next !== null) {
            if (this.fn(next.value)) {
                return {
                    value: next.value,
                    rest: new FilteredStream(this.fn, next.rest),
                };
            }
            next = next.rest.next();
        }
        return null;
        
    }
}

class IntStream extends Stream<number> {
    val: number;

    constructor(start: number = 0) {
        super();
        this.val = start;
    }

    next() {
        return {
            value: this.val,
            rest: new IntStream(this.val + 1),
        };
    }
}

const ints = new IntStream(0);
const intsFrom = (n: number) => new IntStream(n);

const stream = ints.take(100).then(a => (
    ints.take(100).filter(b => a > b).map(b => [a, b])
));

console.log(stream.next());