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

• source <AsyncIterable> | <Iterable> 数据源。
• ...transforms <Function> | <Object> 要应用的零个或多个转换。
• options <Object>
  • signal <AbortSignal> 中止管道。
• 返回:AsyncIterable

创建一个惰性异步管道。数据不会从 source 读取,直到返回的可迭代对象被消费。转换按顺序应用。

🌐 Create a lazy async pipeline. Data is not read from source until the returned iterable is consumed. 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拷贝