observer-vs-generator.ts

import Rx from 'rxjs';

/* library code */

function* count() {
    let i = 0;
    while (true) {
        yield i;
        i += 1;
    }
}

// this will be builtin as of ES2022!
// https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/fromAsync
async function arrayFromAsync<T> (gen: AsyncGenerator<T>) {
    const result: Array<T> = [];
    for await(const a of gen) {
        result.push(a);
    }
    return result;
}

function asyncTake(n: number) {
    return async function* <T> (gen: Generator<T> | AsyncGenerator<T>) {
        for (let i = 0; i < n; i += 1) {
            const { value, done } = await gen.next();
            if (done) break;
            yield value;
        }
    }
}

class Pipe<A> {
    value: A;
    constructor(value: A) {
        this.value = value;
    }

    pipe <B> (fn: (a: A) => B) {
        return new Pipe(fn(this.value));
    }
}

/* business logic */

async function main() {
    Rx.from(count()).pipe(
        Rx.take(10),
        Rx.mergeMap((a) => {
            if (a % 2 === 0) {
                return Rx.of(a, a);
            } else {
                return Rx.of(a);
            }
        }),
        Rx.toArray(),
    ).subscribe((oResult) => {
        console.log(`rxjs observable: ${oResult}`)
    });

    (new Pipe(count())
        .pipe(asyncTake(10))
        .pipe(async function * (g) {
            for await (const a of g) {
                if (a % 2 === 0) {
                    yield* [a, a];
                } else {
                    yield a;
                }
            }
        })
        .pipe(arrayFromAsync)
        .value
    ).then((gResult) => {
        console.log(`js generator:    ${gResult}`);
    });
}
    
main();
// rxjs observable: 0,0,1,2,2,3,4,4,5,6,6,7,8,8,9
// js generator:    0,0,1,2,2,3,4,4,5,6,6,7,8,8,9