# @effectionx/stream-helpers 中文文档

规范地址：https://npmcn.com/package/%40effectionx/stream-helpers/

版本：0\.8\.3
npm 元数据更新时间：2026-09-30T21:08:29.243Z
中文译文更新时间：2026-11-13T08:27:25.886Z
软件包许可：MIT

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40effectionx/stream-helpers
原文对照：https://npmcn.com/package/%40effectionx/stream-helpers/?lang=original
项目主页：https://github.com/thefrontside/effectionx#readme
源码仓库：https://github.com/thefrontside/effectionx

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：197 kB，96 个文件
- 模块格式：ES Module
- TypeScript 类型：包内自带
- 运行依赖：4 个
- peer 依赖：effection ^3 \|\| ^4
- 首次发布：2025-05-19T15:39:06.750Z
- 最新版本发布：2026-06-16T16:32:24.113Z
- 近 30 天下载量：47680（2026-10-13 至 2026-11-11，npm 官方下载统计）
- 已发布版本数：16（近 12 个月 12 个）

## 中文简介

诸如过滤器、映射、减少和预览等安全流操作器

## 安装

```sh
npm install @effectionx/stream-helpers
```

也可以使用：`pnpm add @effectionx/stream-helpers`、`yarn add @effectionx/stream-helpers`、`bun add @effectionx/stream-helpers`

## README

# 串流帮助

顶上建有一组安全类型流帮助器 [效果](https://github.com/thefrontside/effection) 提高能效和 控制流处理.

## 包含的帮助者

### 过滤器

这个 `filter` 帮助缩小流,只通过匹配的项目 上观相.

```typescript
import { filter } from "@effectionx/stream-helpers";
import { each } from "effection";

// Example: Synchronous filtering
function* syncExample(source: Stream<number, unknown>) {	

  const gt5 = filter<number>(function* (x) { return x > 5 });

  for (const value of yield* each(gt5(stream))) {
    console.log(value); // Only values > 5
    yield* each.next();
  }
};

// Example: Asynchronous filtering
function* asyncExample(source: Stream<number, unknown>) {

  const evensOf = filter<number>(function* (x) {
    yield* sleep(100); // Simulate async operation
    return x % 2 === 0; // Keep only even numbers
  });

  for (const value of yield* each(evensOf(stream))) {
    console.log(value); // Only even numbers
    yield* each.next();
  }
});
```

### 地图

这个 `map` 帮助在流中使用提供的函数转换每个项目. 用于您需要处理每个数据转换操作 各个项目.

```typescript
import { map } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example(stream: Stream<number, unknown>) {
  const double = map<number>(function* (x) {
    return x * 2;
  });

  for (const value of yield* each(double(stream))) {
    console.log(value); // Each value is doubled
    yield* each.next();
  }
}
```

### 减少

这个 `reduce` 帮助工具通过应用在流中转换每个项目 累积值.

```typescript
import { reduce } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example(stream: Stream<number, unknown>) {
  let sum = reduce(function* (total: number, current: number) {
    return total + current;
  }, 0);

  for (let value of yield* each(sum(streamOf([1,2,3]))) {
    console.log(value) // logs 1 -> 3 -> 6
    yield* each.next();
  }
}
```

### 批次

这个 `batch` 帮助工具在您想要转换单个通过的项目时有用 通过流入数组项目。 批量可以由 指定最大时间或最大大小。 如果两者都明确,则批次 将在满足任一条件时创建.

```typescript
import { batch } from "@effectionx/stream-helpers";
import { each } from "effection";

// Example: Batch by size
function* exampleBySize(stream: Stream<number, unknown>) {
  const byThree = batch({ maxSize: 3});

  for (const items of yield* each(byThree(stream))) {
    console.log(batch); // [1, 2, 3], [4, 5, 6], ...
    yield* each.next();
  }
};

// Example: Batch by time
function* exampleByTime(stream: Stream<number, unknown>) {
  const stream = batch({ maxTime: 1000 })(sourceStream);

  for (const batch of yield* each(stream)) {
    console.log(batch); // Items received within 1 second
    yield* each.next();
  }
});

// Example: Combined batching
function* exampleCombined(stream: Stream<number, unknown>) {

  const batched = batch({
    maxSize: 5,
    maxTime: 1000,
  });

  for (const batch of yield* each(batched(stream))) {
    console.log(batch); // Up to 5 items within 1 second
    yield* each.next();
  }
});
```

### 阀门

允许对源流应用回压来防止压倒 下游消费者。 这对任何生成更快项目的流都有用 比消费者可以消费它们。 它最初设计用于: Kafka , 制作人可以让服务耗尽内存 制作人生产出许多比消费者更快的处理信件. 它会 用于任何无限流的缓冲.

```typescript
import { valve } from "@effectionx/stream-helpers";
import { each } from "effection";

function* example() {
  const regulated = valve({
    // buffer size threshold when close operation will invoked
    closeAt: 1000,
    *close() {
      // pause the source stream
    },

    // buffer size threshold when open operation will be invoked
    openAt: 100,
    *open() {
      // resume the source stream
    },
  })(stream);

  for (const value of yield* each(regulated)) {
    console.log(value);
    yield* each.next();
  }
}
```

### 下水道

这个 `drain` 帮助将流耗尽,丢弃所有生成的值,然后返回 近值。 当你只关心最后的结果时,这个是有用的 流,而不是中间值.

```typescript
import { drain } from "@effectionx/stream-helpers";

function* example() {
  // Get the response from a request channel (ignoring any progress)
  const channel = yield* transport.send(request);
  const response = yield* drain(channel);
  console.log(response); // The close value
}
```

### 第一届

这个 `first` 帮助返回由流生成的第一个值,或 `undefined` 如果流关闭而不产生任何值。 使用 `first.expect()` 要掷出 错误而不是返回 `undefined`。 。 。 .

```typescript
import { first } from "@effectionx/stream-helpers";
import { streamOf } from "@effectionx/stream-helpers";

function* example() {
  const stream = streamOf([1, 2, 3]);
  const value = yield* first(stream);
  console.log(value); // 1

  const empty = streamOf([]);
  const none = yield* first(empty);
  console.log(none); // undefined

  // Use first.expect() to throw if stream is empty
  const required = yield* first.expect(stream); // throws if empty
}
```

### 最后一个

这个 `last` 帮助返回由流产生的最后一个值,或 `undefined` 如果流关闭而不产生任何值。 它会耗尽整个溪流 查找最后一个值。 使用 `last.expect()` 来丢出一个错误而不是 返回时 `undefined`。 。 。 .

```typescript
import { last } from "@effectionx/stream-helpers";
import { streamOf } from "@effectionx/stream-helpers";

function* example() {
  const stream = streamOf([1, 2, 3]);
  const value = yield* last(stream);
  console.log(value); // 3

  const empty = streamOf([]);
  const none = yield* last(empty);
  console.log(none); // undefined

  // Use last.expect() to throw if stream is empty
  const required = yield* last.expect(stream); // throws if empty
}
```

### 拿着

这个 `take` 帮助创建第一个生成的流变压器 `n` 数值 从源流中取出,然后用最后的取值关闭.

```typescript
import { take, forEach, streamOf } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  const stream = streamOf([1, 2, 3, 4, 5]);
  // yields 1, 2, then closes with 3
  const closeValue = yield* forEach(function* (value) {
    console.log(value); // 1, then 2
  }, take(3)(stream));
  console.log(closeValue); // 3
}

// Works with pipe
const limited = pipe(source, take(3));
```

### 慢慢来

这个 `takeWhile` 帮助创建流式变压器,在同时生成值 上游返回真实。 当上游返回时,溪会关闭 立即(不包括失败的值).

```typescript
import { takeWhile, forEach, streamOf } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  const stream = streamOf([1, 2, 3, 4, 5]);
  // yields 1, 2 (stops when value >= 3)
  yield* forEach(function* (value) {
    console.log(value); // 1, then 2
  }, takeWhile((x) => x < 3)(stream));
}

// Works with pipe
const filtered = pipe(source, takeWhile((x) => x.isValid));
```

### 使用到结束

这个 `takeUntil` 帮助创建流式变压器,该变压器输出值直到 上游返回真实。 当上游匹配时,溪流会关闭 匹配值。 这对“在满足条件之前”有用 模式.

```typescript
import { takeUntil, forEach } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

function* example() {
  // Iterate validation progress until we get a terminal status
  const result = yield* forEach(
    function* (progress) {
      showSpinner(progress.status);
    },
    takeUntil((p) => p.status === "valid" || p.status === "invalid")(
      validationStream,
    ),
  );

  // result is the validation object with terminal status
  if (result.status === "valid") {
    console.log("Validation passed!");
  }
}

// Works with pipe
const untilDone = pipe(source, takeUntil((x) => x.done));
```

**关键差异 `takeWhile` 财务报告和审定财务报表 `takeUntil`数字 :**

| 帮助 | 也门 | 关闭方式 | 使用大小写 |
| --- | --- | --- | --- |
| `takeWhile` | 虽然上游是真的 | `undefined` | 继续走,继续走 |
| `takeUntil` | 直到上游是真实的 | 匹配值 | "当你找到它时停止" |

### 每个

这个 `forEach` 帮助为每个通过流的项目引用函数. 当您需要在每个项目上执行副作用或操作时,此功能是有用的 不改变溪流本身。 不同于返回的其他流帮助 变了的溪流, `forEach` 消耗整个溪流并返回 流的近值.

```typescript
import { forEach } from "@effectionx/stream-helpers";
import { createSignal, spawn } from "effection";

function* example() {
  const stream = createSignal<number, void>();

  // Process each item in the stream
  yield* spawn(() =>
    forEach(function* (item) {
      console.log(`Processing: ${item}`);
      // Perform any side effects here
    }, stream)
  );

  stream.send(1);
  stream.send(2);
  stream.send(3);
  stream.close();
}

// Example: Handling stream close value
function* exampleWithCloseValue() {
  const stream = createSignal<string, number>();

  const result = yield* spawn(() =>
    forEach(function* (item) {
      console.log(`Item: ${item}`);
    }, stream)
  );

  stream.send("hello");
  stream.send("world");
  stream.close(42); // Close with value 42

  const closeValue = yield* result;
  console.log(`Stream closed with: ${closeValue}`); // 42
}
```

### 议题

主题帮助器将任意流转换成重播的多播流 新用户的最新值。 类似 [RxJS 行为主体](https://www.learnrxjs.io/learn-rxjs/subjects/behaviorsubject)。 。 。 .

```typescript
import { createSubject } from "@effectionx/stream-helpers";
import { createChannel, spawn } from "effection";

function* example() {
  const subject = createSubject<number>();
  const channel = createChannel<number, void>();
  const downstream = subject(channel);

  // First subscriber
  const sub1 = yield* downstream;

  yield* channel.send(1);
  yield* channel.send(2);

  console.log(yield* sub1.next()); // { done: false, value: 1 }
  console.log(yield* sub1.next()); // { done: false, value: 2 }

  // Late subscriber gets the latest value immediately
  const sub2 = yield* downstream;
  console.log(yield* sub2.next()); // { done: false, value: 2 }
}
```

使用管道操作符将任何流转换成行为主体 :

```typescript
import { createSubject, map } from "@effectionx/stream-helpers";
import { pipe } from "remeda";

const subject = createSubject<string>();

const stream = pipe(
  source,
  map(function* (x) {
    return x.toString();
  }),
  subject,
);
```

### 通过跟踪器

通过 Tracker 流帮助提供了一种了解所有项目的方法 穿过溪流已经处理完毕。 这对你特别有帮助 想要确保所有物品在操作完成前都得到处理.

和其他小溪帮手不一样 因为你必须先打电话 `createTracker` 函数重击对象。 实际的帮助者在 `passthrough` 可以调用的方法,并与其他方法一样链条 帮手.

```typescript
import { each, signal } from "effection";
import { createTracker } from "@ffectionx/stream-helpers"

const source = signal(0);

// create the tracker
const tracker = yield* createTracker();

// create  passthrough stream helper
const track = tracker.passthrough();

for (const value of yield* each(track(source))) {
  // mark items 
  tracker.markOne(value);
  yield* each.next();
}

// will resolve when all items that passed through the stream were seen
yield* tracker;
```

### 编组流帮助

你可以用简单的 `pipe()` 组成一系列的溪流帮手 单位 这个例子,我们用一个来自 [医生](https://remedajs.com/docs/#pipe), (中文(简体) ),

```typescript
import { batch, filter, forEach, map, valve } from "@effectionx/stream-helpers";
// any standard pipe function should work
import { pipe } from "remeda";

function* example(source: Stream<number, unknown>) {
  // Compose stream helpers using pipe
  const stream = pipe(
    source,
    valve({ open, close, openAt: 100, closeAt: 100 }),
    filter(function* (x) {
      return x > 0;
    }),
    map(function* (x) {
      return x * 20;
    }),
    batch({ maxSize: 50 }),
  );

  // Use forEach to process each value in the composed stream
  yield* forEach(function* (value) {
    console.log(value);
  }, stream);
}
```

## 测试流

库包括测试工具, 帮助您测试您的流处理 代码。 这些表格载于 `@effectionx/stream-helpers/test-helpers` 输出.

### 福赛特

这个 `useFaucet` 函数创建可用于测试行为的流 使用后压的溪流 在你的测试中特别有用 需要一个可控源流 .

```typescript
import { useFaucet } from "@effectionx/stream-helpers/test-helpers";
import { each, run, spawn } from "effection";

await run(function* () {
  const faucet = yield* useFaucet<number>({ open: true });

  // Remember to spawn the stream subscription before sending items to the stream
  yield* spawn(function* () {
    for (let i of yield* each(faucet)) {
      console.log(i);
      yield* each.next();
    }
  });

  // Pass an array of items to send items to the stream one at a time synchronously
  yield* faucet.pour([1, 2, 3]);

  // Pass an operation to control the rate at which items are sent to the stream
  yield* faucet.pour(function* (send) {
    yield* sleep(10);
    yield* send(5);
    yield* sleep(30);
    yield* send(6);
    yield* sleep(10);
    yield* send(7);
  });

  // You can close the faucet to stop items from being sent
  faucet.close();

  // And open it again when needed
  faucet.open();
});
```

当函数传递到 `faucet.pour`,则 `send` 函数将返回一个 操作。 水龙头关闭后 行动就会被阻止 这是 在测试后压时特别有用, 因为您可以发送许多事件 不用担心 `open/close` 状态.

## 常见问题

### @effectionx/stream-helpers 是什么？

诸如过滤器、映射、减少和预览等安全流操作器

### 如何安装 @effectionx/stream-helpers？

在项目目录运行 npm install @effectionx/stream-helpers，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：effection（^3 \|\| ^4）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

### @effectionx/stream-helpers 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @effectionx/stream-helpers 支持 ESM 还是 CommonJS？

以 ES Module 形式发布，推荐使用 import 引入。在 CommonJS 项目中，Node\.js 20\.19\+ / 22\.12\+ 可以直接 require\(\) 不含顶层 await 的 ES Module，更早的版本需要改用动态 import\(\)。

### @effectionx/stream-helpers 可以商用吗？

@effectionx/stream-helpers 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

### @effectionx/stream-helpers 的下载量有多少？

根据 npm 官方下载统计，@effectionx/stream-helpers 在 2026/10/13 – 2026/11/11 这 30 天内被下载 47,680 次，日均约 1,589 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @effectionx/stream-helpers 还在维护吗？

最新版本 0\.8\.3 发布于 2026/06/17（4 个月前）；至今共发布 16 个版本，其中过去 12 个月发布了 12 个；首次发布于 2025/05/19。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。