blog/lib/rx-utils.ts

import { Observable, Subscriber } from "rxjs";
import File from "vinyl";
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 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 const loadFile = async (path: string): Promise<File> =>
    new File({ path, contents: await readFile(path) });

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();
                },
            }),
        );
}