pull(source[, ...transforms][, options])

• source <AsyncIterable> | <Iterable> 数据源。
• ...transforms <Function> | <Object> 要应用的零个或多个转换。
• options <Object>
  • signal <AbortSignal> 中止管道。
• 返回：<AsyncIterable> 它的块用 <Uint8Array[]> 填充

创建一个懒惰的异步管道。源转换和可流式协议的分发会在调用 pull() 时发生，但数据直到返回的可迭代对象被消费时才会从 source 读取。已经中止的信号会在源转换后同步抛出。转换按顺序应用。

🌐 Create a lazy async pipeline. Source conversion and streamable protocol dispatch occur when pull() is called, but data is not read from source until the returned iterable is consumed. A signal that is already aborted is thrown synchronously after source conversion. Transforms are applied in order.

import { from, pull, text } from 'node:stream/iter';

const asciiUpper = (chunks) => {
  if (chunks === null) return null;
  return chunks.map((c) => {
    for (let i = 0; i < c.length; i++) {
      c[i] -= (c[i] >= 97 && c[i] <= 122) * 32;
    }
    return c;
  });
};

const result = pull(from('hello'), asciiUpper);
console.log(await text(result)); // 'HELLO'const { from, pull, text } = require('node:stream/iter');

const asciiUpper = (chunks) => {
  if (chunks === null) return null;
  return chunks.map((c) => {
    for (let i = 0; i < c.length; i++) {
      c[i] -= (c[i] >= 97 && c[i] <= 122) * 32;
    }
    return c;
  });
};

async function run() {
  const result = pull(from('hello'), asciiUpper);
  console.log(await text(result)); // 'HELLO'
}

run().catch(console.error);拷贝

使用 AbortSignal：

🌐 Using an AbortSignal:

import { pull } from 'node:stream/iter';

const ac = new AbortController();
const result = pull(source, transform, { signal: ac.signal });
ac.abort(); // Pipeline throws AbortError on next iterationconst { pull } = require('node:stream/iter');

const ac = new AbortController();
const result = pull(source, transform, { signal: ac.signal });
ac.abort(); // Pipeline throws AbortError on next iteration拷贝