feat(async): add waterfall and async-queue functionality with tests

src/async/queue.ts

@@ -0,0 +1,114 @@
+import { withResolvers } from "radashi";
+
+export type QueueTask<T, TResult> = {
+  task: T;
+  resolve: (result: TResult) => void;
+  reject: (error: Error) => void;
+};
+
+export type IQCallback<TResult> = (err: Error | undefined, result: TResult) => void;
+export type IQListener<T> = (err: Error, task: T) => void;
+
+export type IQParams<T, TResult> = {
+  input: T | T[];
+  callback?: IQCallback<TResult>;
+}
+
+export interface Queue<T, TResult> {
+  push: (tasks: T | T[], callback?: IQCallback<TResult>) => void;
+  unshift: (tasks: T | T[], callback?: IQCallback<TResult>) => void;
+  drain: {
+    () : Promise<void>,
+    (listener: () => void): void,
+  };
+  error: (listener: IQListener<T>) => void;
+};
+
+
+export function queue<T, TResult>(
+  worker: (task: T) => TResult | PromiseLike<TResult>,
+  concurrency: number,
+): Queue<T, TResult> {
+
+  // internal state
+  const tasks: Array<{ task: T, callback?: IQCallback<TResult> }> = []; // FIFO butter
+  let running = 0;
+  let errorListener: IQListener<T> | null = null;
+  let drainListener: (() => void) | null = null;
+  let drainPromise = withResolvers<void>();
+
+  // core scheduler
+  const schedule = () => {
+    while (running < concurrency && tasks.length > 0) {
+      const item = tasks.shift()!;
+      running++;
+
+      processTask(item);
+    }
+    checkDrain();
+  }
+
+  const processTask = async (item: { task: T, callback?: IQCallback<TResult> }) => {
+    try {
+      const result = await worker(item.task);
+      item.callback?.(undefined, result);
+    }catch(err) {
+      errorListener?.(err as Error, item.task);
+      item.callback?.(err as Error, undefined as never);
+    } finally {
+      running--;
+      schedule();
+    }
+  }
+
+  const checkDrain = () => {
+    if(running === 0 && tasks.length === 0) {
+      drainListener?.();
+      drainPromise.resolve();
+      drainPromise = withResolvers<void>();
+    }
+  }
+
+
+  // insert the task or 's to the queue 
+  const push = (
+    input: IQParams<T, TResult>["input"], 
+    callback?: IQParams<T, TResult>["callback"]
+  ) => {
+    const items = Array.isArray(input) ? input : [input]
+    if(items.length === 0) return;
+
+    for(const task of items) {
+      tasks.push({ task, ...(callback ? { callback } : {}) });
+    }
+    schedule();
+  }
+
+  // moves the task iteratively to the front from the back of the queue.
+  const unshift = (
+    input: IQParams<T, TResult>["input"], 
+    callback?: IQParams<T, TResult>["callback"]
+  ) => {
+    const items = Array.isArray(input) ? input : [input];
+    if(items.length === 0) return;
+
+    for(let i = items.length - 1; i >= 0; i--) {
+      tasks.unshift({ task: items[i], ...(callback ? { callback } : {}) });
+    }
+    schedule();
+  }
+
+  const drain = ((listener?: () => void) => {
+    if(listener) {
+      drainListener = listener;
+      return;
+    }
+    return drainPromise.promise;
+  }) as Queue<T, TResult>["drain"];
+
+  const error = (listener: IQListener<T>) => {
+    errorListener = listener;
+  }
+
+  return { push, unshift, drain, error };
+}

src/async/waterfall.ts

@@ -0,0 +1,16 @@
+/**
+ * Executes a sequence of asynchronous functions in order, where the output
+ * of each function becomes the input to the next, and returns the final result.
+ *
+ * Execution:
+ * initialInput → fn1() → fn2() → fn3() → ... → final result
+ */
+export async function waterfall<T>(
+  funcs: Array<(input: T) => Promise<T>>,
+  initialInput?: T,
+): Promise<T> {
+  return funcs.reduce<Promise<T>>(
+    async (acc, fn) => fn(await acc),
+    Promise.resolve(initialInput as T),
+  );
+}

src/mod.ts

@@ -39,6 +39,7 @@ export * from './async/defer.ts'
 export * from './async/guard.ts'
 export * from './async/map.ts'
 export * from './async/parallel.ts'
+export * from './async/queue.ts'
 export * from './async/queueByKey.ts'
 export * from './async/reduce.ts'
 export * from './async/retry.ts'
@@ -47,6 +48,7 @@ export * from './async/timeout.ts'
 export * from './async/toResult.ts'
 export * from './async/tryit.ts'
 export * from './async/withResolvers.ts'
+export * from './async/waterfall.ts';
 
 export * from './curry/callable.ts'
 export * from './curry/chain.ts'

tests/async/queue.test.ts

@@ -0,0 +1,192 @@
+import * as _ from 'radashi'
+
+describe('queue', () => {
+  beforeEach(() => {
+    vi.useFakeTimers({ shouldAdvanceTime: true })
+  })
+
+  test('executes tasks in FIFO order respecting concurrency', async () => {
+    const results: string[] = []
+    const worker = async (task: string) => {
+      await _.sleep(5)
+      results.push(task)
+    }
+    const q = _.queue(worker, 2)
+    q.push('a')
+    q.push('b')
+    q.push('c')
+    await q.drain()
+    expect(results).toEqual(['a', 'b', 'c'])
+  })
+
+  test('limits concurrent execution to concurrency', async () => {
+    let running = 0
+    let maxRunning = 0
+    const worker = async (task: number) => {
+      running++
+      maxRunning = Math.max(maxRunning, running)
+      await _.sleep(100)
+      running--
+    }
+    const q = _.queue(worker, 2)
+    q.push([1, 2, 3, 4])
+    await q.drain()
+    expect(maxRunning).toBeLessThanOrEqual(2)
+    expect(maxRunning).toBe(2)
+  })
+
+  test('processes batch push in order', async () => {
+    const results: number[] = []
+    const worker = async (task: number) => {
+      await _.sleep(10)
+      results.push(task)
+    }
+    const q = _.queue(worker, 1)
+    q.push([1, 2, 3])
+    await q.drain()
+    expect(results).toEqual([1, 2, 3])
+  })
+
+  test('unshift adds tasks to front of pending queue', async () => {
+    const results: string[] = []
+    const worker = async (task: string) => {
+      await _.sleep(10)
+      results.push(task)
+    }
+    const q = _.queue(worker, 1)
+    q.push('b')
+    q.push('c')
+    q.unshift('a')
+    await q.drain()
+    expect(results).toEqual(['b', 'a', 'c'])
+  })
+
+  test('unshift with batch preserves relative order', async () => {
+    const results: string[] = []
+    const worker = async (task: string) => {
+      await _.sleep(10)
+      results.push(task)
+    }
+    const q = _.queue(worker, 1)
+    q.push('c')
+    q.unshift(['a', 'b'])
+    await q.drain()
+    expect(results).toEqual(['c', 'a', 'b'])
+  })
+
+  test('drain callback fires when queue becomes idle', async () => {
+    const spy = vi.fn()
+    const worker = async (task: string) => {
+      await _.sleep(10)
+    }
+    const q = _.queue(worker, 2)
+    q.drain(spy)
+    q.push('a')
+    q.push('b')
+    await vi.advanceTimersToNextTimerAsync()
+    expect(spy).toHaveBeenCalledTimes(1)
+  })
+
+  test('await q.drain() resolves after all tasks complete', async () => {
+    const results: string[] = []
+    const worker = async (task: string) => {
+      await _.sleep(10)
+      results.push(task)
+    }
+    const q = _.queue(worker, 2)
+    q.push('a')
+    q.push('b')
+    await q.drain()
+    expect(results).toEqual(['a', 'b'])
+  })
+
+  test('queue is reusable after drain', async () => {
+    const results: number[] = []
+    const worker = async (task: number) => {
+      await _.sleep(10)
+      results.push(task)
+    }
+    const q = _.queue(worker, 2)
+    q.push([1, 2])
+    await q.drain()
+    expect(results).toEqual([1, 2])
+    q.push([3, 4])
+    await q.drain()
+    expect(results).toEqual([1, 2, 3, 4])
+  })
+
+  test('error callback receives errors from worker', async () => {
+    const errorSpy = vi.fn()
+    const worker = async (task: string) => {
+      if (task === 'fail') {
+        throw new Error('oops')
+      }
+      await _.sleep(10)
+    }
+    const q = _.queue(worker, 1)
+    q.error(errorSpy)
+    q.push('ok')
+    q.push('fail')
+    q.push('ok2')
+    await q.drain()
+    expect(errorSpy).toHaveBeenCalledTimes(1)
+    expect(errorSpy).toHaveBeenCalledWith(expect.any(Error), 'fail')
+  })
+
+  test('per-task callback receives error or result', async () => {
+    const okSpy = vi.fn()
+    const failSpy = vi.fn()
+    const worker = async (task: string) => {
+      if (task === 'fail') {
+        throw new Error('oops')
+      }
+      await _.sleep(10)
+      return task.toUpperCase()
+    }
+    const q = _.queue(worker, 1)
+    q.push('ok', okSpy)
+    q.push('fail', failSpy)
+    await q.drain()
+    expect(okSpy).toHaveBeenCalledWith(undefined, 'OK')
+    expect(failSpy).toHaveBeenCalledWith(expect.any(Error), undefined)
+  })
+
+  test('queue continues processing after a task error', async () => {
+    const results: string[] = []
+    const worker = async (task: string) => {
+      await _.sleep(10)
+      if (task === 'fail') {
+        throw new Error('oops')
+      }
+      results.push(task)
+    }
+    const q = _.queue(worker, 2)
+    q.push(['ok1', 'fail', 'ok2'])
+    await q.drain()
+    expect(results).toEqual(['ok1', 'ok2'])
+  })
+
+  test('push with empty array does nothing', () => {
+    const worker = vi.fn()
+    const q = _.queue(worker, 2)
+    q.push([])
+    expect(worker).not.toHaveBeenCalled()
+  })
+
+  test('unshift with empty array does nothing', () => {
+    const worker = vi.fn()
+    const q = _.queue(worker, 2)
+    q.unshift([])
+    expect(worker).not.toHaveBeenCalled()
+  })
+
+  test('unshift with callback', async () => {
+    const cb = vi.fn()
+    const worker = async (task: string) => task.toUpperCase()
+    const q = _.queue(worker, 2)
+    q.push('b')
+    q.unshift('a', cb)
+    await q.drain()
+    expect(cb).toHaveBeenCalledWith(undefined, 'A')
+  })
+})