then pattern

lib/rx-utils.ts

@@ -1,62 +1,56 @@
-import { Observable, from, map, mergeAll } from "rxjs";
+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, mkdirSync, writeFileSync } from "node:fs";
+import { existsSync } from "node:fs";
+import { writeFile, mkdir } from "node:fs/promises";
 
-export function src(glob: string | string[]): Observable<File> {
+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) })),
-        map(from),
-        mergeAll(),
     );
 }
 
-export function dist(prefix: string) {
+export function then<T, U>(f: (v: T) => Promise<U>) {
-    return (observable: Observable<File>): Observable<File> =>
+    return (observable: Observable<Promise<T>>): Observable<Promise<U>> =>
-        new Observable((subscriber) =>
+        observable.pipe(map((v) => v.then(f)));
-            observable.subscribe({
+}
-                next(value) {
+
-                    const actualPath = join(prefix, value.path);
+export function dest(prefix: string) {
+    return async (file: File) => {
+        const actualPath = join(prefix, file.path);
         if (!existsSync(dirname(actualPath))) {
-                        mkdirSync(dirname(actualPath), { recursive: true });
+            await mkdir(dirname(actualPath), { recursive: true });
         }
-                    writeFileSync(actualPath, value.contents as Buffer);
+        await writeFile(actualPath, file.contents as Buffer);
         console.log("[-] Written", actualPath);
 
-                    subscriber.next(value);
+        return file;
-                },
+    };
-                complete() {
-                    subscriber.complete();
-                },
-            }),
-        );
 }
 
 export function synchronise<T>() {
     return (observable: Observable<Promise<T>>): Observable<T> =>
         new Observable<T>((subscriber) => {
-            const promiseArray = [];
+            const runningPromises = new Set<Promise<void>>();
             let done = false;
 
             observable.subscribe({
                 next(value) {
                     const promise = value.then((v) => {
-                        const i = promiseArray.findIndex((i) => i == promise);
+                        runningPromises.delete(promise);
-                        promiseArray.splice(i, 1);
 
                         subscriber.next(v);
-                        if (promiseArray.length === 0 && done) {
+                        if (runningPromises.size === 0 && done) {
                             subscriber.complete();
                         }
                     });
-                    promiseArray.push(promise);
+                    runningPromises.add(promise);
                 },
                 complete() {
                     done = true;
-                    if (promiseArray.length === 0) {
+                    if (runningPromises.size === 0) {
-                        console.log("[synchronise] complete (from complete)");
                         subscriber.complete();
                     }
                 },

lib/tasks/articles.ts

@@ -18,8 +18,8 @@ import { renderToStaticMarkup } from "preact-render-to-string";
 import { SITE_TITLE, SITE_DESCRIPTION } from "../constants.js";
 import { JSX } from "preact/jsx-runtime";
 import { reloadAssets } from "../assets.js";
-import { map, lastValueFrom } from "rxjs";
+import { lastValueFrom } from "rxjs";
-import { src, synchronise, dist } from "../rx-utils.js";
+import { src, synchronise, dest, then } from "../rx-utils.js";
 
 const Asciidoctor = asciidoctor();
 const EXTENSION_REGISTRY = Asciidoctor.Extensions.create();
@@ -75,7 +75,7 @@ export async function articles(): Promise<void> {
 
     await lastValueFrom(
         src("articles/**/*.asciidoc").pipe(
-            map(async (file) => {
+            then(async (file) => {
                 const slug = basename(file.path, ".asciidoc");
                 const document = Asciidoctor.load(file.contents.toString(), {
                     extension_registry: EXTENSION_REGISTRY,
@@ -119,8 +119,8 @@ export async function articles(): Promise<void> {
 
                 return file;
             }),
+            then(dest("dist")),
             synchronise(),
-            dist("dist"),
         ),
     );
     images.forEach(_output);
@@ -133,7 +133,7 @@ export async function articles(): Promise<void> {
             "og:type": "website",
             "og:url": `https://adaedra.eu`,
         },
-        Content: () => renderIndex({ articles: articles }),
+        Content: () => renderIndex({ articles }),
     });
 
     _output(