blog/lib/rx-utils.ts

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

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 onComplete<T>(f: (sink: Subscriber<T>) => Promise<void>) {
    return (observable: Observable<T>) =>
        new Observable<T>((subscriber) =>
            observable.subscribe({
                next(value) {
                    subscriber.next(value);
                },
                error(err) {
                    subscriber.error(err);
                },
                complete() {
                    f(subscriber);
                    subscriber.complete();
                },
            })
        );
}

const loadFile = async (path: string): Promise<File> => new File({ path, contents: await readFile(path) });

export const fromGlob = (paths: string | string[]): Observable<File> =>
    from(new Glob(paths, {})).pipe(mergeMap(loadFile));