blog/lib/rx-utils.ts

import { Observable, from, map } from "rxjs";
import File from "vinyl";
import { Glob } from "glob";
import { readFile } from "node:fs/promises";
import { join, dirname } from "node:path";
import { existsSync } from "node:fs";
import { writeFile, mkdir } from "node:fs/promises";

export function src(glob: string | string[]): Observable<Promise<File>> {
    return from(new Glob(glob, {})).pipe(
        map(async (path) => new File({ path, contents: await readFile(path) })),
    );
}

export function then<T, U>(f: (v: T) => Promise<U>) {
    return (observable: Observable<Promise<T>>): Observable<Promise<U>> =>
        observable.pipe(map((v) => v.then(f)));
}

export function dest(prefix: string) {
    return async (file: File) => {
        const actualPath = join(prefix, file.path);
        if (!existsSync(dirname(actualPath))) {
            await mkdir(dirname(actualPath), { recursive: true });
        }
        await writeFile(actualPath, file.contents as Buffer);
        console.log("[-] Written", actualPath);

        return file;
    };
}

export function synchronise<T>() {
    return (observable: Observable<Promise<T>>): Observable<T> =>
        new Observable<T>((subscriber) => {
            const runningPromises = new Set<Promise<void>>();
            let done = false;

            observable.subscribe({
                next(value) {
                    const promise = value.then((v) => {
                        runningPromises.delete(promise);

                        subscriber.next(v);
                        if (runningPromises.size === 0 && done) {
                            subscriber.complete();
                        }
                    });
                    runningPromises.add(promise);
                },
                complete() {
                    done = true;
                    if (runningPromises.size === 0) {
                        subscriber.complete();
                    }
                },
            });
        });
}