@zeg-giang/pipe

基于 RxJS 的业务流程编排库。

将多个 RxJS 操作符串联成管道，返回一个 Promise。适用于编排顺序执行的业务逻辑步骤 — 审批工作流、ETL 数据管道、数据校验链、重试/降级等场景，同时拥有完整的 RxJS 操作符能力。

安装

npm install @zeg-giang/pipe

rxjs 是 peer 依赖，需要确保已安装：

npm install rxjs

使用

链式调用

const add1: OperatorFunction<number, number> = map((x: number) => x + 1);
const double: OperatorFunction<number, number> = map((x: number) => x * 2);
const result = await pipe(add1, double);
// NaN（起始值为 undefined）

数据清洗

const result = await pipe(
  map(() => ({ name: "  Alice  ", email: "  [email protected]  " })),
  map((input) => ({ name: input.name.trim(), email: input.email.trim().toLowerCase() })),
  filter((user) => user.name.length > 0),
);
// { name: "Alice", email: "[email protected]" }

错误处理

// 重试
const result = await pipe(flakyApi, retry(3));

// 降级
const result = await pipe(alwaysFails, catchError(() => of("降级响应")));

API

pipe(...operations): Promise<T>

创建一个起始值为 undefined 的 Observable，依次串联每个 RxJS OperatorFunction，返回 Promise：

• 无任何值发出 → 返回 undefined
• 发出单个值 → 直接返回该值
• 发出多个值 → 返回包含所有值的数组

参数

| 参数 | 类型 | 说明 | |------|------|------| | operations | ...OperatorFunction<any, any>（可变参数） | 要串联的 RxJS 操作符 |

返回值

Promise<T> — 所有操作符执行完毕后的结果。

环境要求

• Node.js >= 16
• rxjs ^7.8.2

License

MIT