pipelines! in ts

gistfile1.txt

export interface PipelineContext<T> {
  task: T;
  stop: boolean;
}

export type Middleware<T> = (context: PipelineContext<T>) => Promise<PipelineContext<T>>;

/**
 * Stops the pipeline execution by setting the stop flag to true.
 *
 * @template T The type of task being processed through the pipeline
 * @param context The current pipeline context
 * @returns The updated pipeline context with the stop flag set to true
 */
export function stopPipeline<T>(context: PipelineContext<T>) {
  return { ...context, stop: true };
}

/**
 * A Pipeline class that processes a task through a series of middleware functions.
 * Each middleware can transform the task and optionally stop the pipeline execution.
 *
 * @template T The type of task being processed through the pipeline
 */
export class Pipeline<T> {
  private middlewares: Array<Middleware<T>> = [];

  use(fn: Middleware<T>) {
    this.middlewares.push(fn);
    return this;
  }

  async execute(task: T) {
    let context: PipelineContext<T> = { task, stop: false };

    for (const middleware of this.middlewares) {
      context = await middleware(context);

      if (context.stop) {
        break;
      }
    }

    return context.task;
  }
}

Usage

  async doSomething(data) {
    const pipeline = new Pipeline<any>()
      .use(async context => {
        if (context.task.status === 'skip') {
          console.log('Skipping...');
          return stopPipeline(context);
        }
        return context;
      })
      .use(this.doThingOne.bind(this))
      .use(this.doThingTwo.bind(this))
      .use(this.doThingThree.bind(this));
    await pipeline.execute(data);
    return;
  }