Projection

requestDelay() 前面要加 await 才行，不然你这样就是开启 10 个请求，后面 requestInstant() 只能排队

你说的这个用 Streams API 很容易实现：

```js
import { setTimeout } from "node:timers/promises";

let id = 0;

async function fetchData() {
console.log("fetchData()");
await setTimeout(Math.random() * 500);
return Array(10)
.fill(null)
.map(() => ({ id: ++id }));
}

const stream = new ReadableStream(
{
async pull(controller) {
const data = await fetchData();
for (const a of data) {
controller.enqueue(a);
}
},
},
new CountQueuingStrategy({ highWaterMark: 0 }), // fully lazy
).values();

async function getD() {
return (await stream.next()).value;
}

for (let i = 0; i < 30; i++) {
console.log("data", await getD());
}
```

输出结果：

fetchData()
data { id: 1 }
data { id: 2 }
data { id: 3 }
data { id: 4 }
data { id: 5 }
data { id: 6 }
data { id: 7 }
data { id: 8 }
data { id: 9 }
data { id: 10 }
fetchData()
data { id: 11 }
data { id: 12 }
data { id: 13 }
data { id: 14 }
data { id: 15 }
data { id: 16 }
data { id: 17 }
data { id: 18 }
data { id: 19 }
data { id: 20 }
fetchData()
data { id: 21 }
data { id: 22 }
data { id: 23 }
data { id: 24 }
data { id: 25 }
data { id: 26 }
data { id: 27 }
data { id: 28 }
data { id: 29 }
data { id: 30 }

这里为了定义你所需要 getD() 显得把代码搞复杂了一点，而实际上你可以根据情况使用 for await... of 。

你从 LLM 获得的代码有轮询的部分，所以强烈不推荐：

await new Promise(resolve => setTimeout(resolve, 100)); // 等待数据获取完成

我用生成器简单实现了一下，代码 https://pastebin.com/hi04KbPd

function delay(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}

const then = Date.now();

async function task(name, input) {
console.log(`[${Date.now() - then}] ${name}: start, input: ${input}`);
await delay(1_000);
console.log(`[${Date.now() - then}] ${name}: done, input: ${input}`);
return input + name;
}

let token = 0;
async function onUpdate(input) {
let current = ++token;
async function* gen() {
let result = await task('a', input);
yield;
result = await task('b', result);
yield;
result = await task('c', result);
console.log(`[${Date.now() - then}] final result: ${result}`);
return result;
}
for await (const _ of gen()) {
if (token !== current) break;
}
}

onUpdate('1');
await delay(500);
onUpdate('2');
await delay(2_000);
onUpdate('3');

输出结果：

[0] a: start, input: 1
[510] a: start, input: 2
[1006] a: done, input: 1
[1526] a: done, input: 2
[1527] b: start, input: 2a
[2516] a: start, input: 3
[2533] b: done, input: 2a
[3520] a: done, input: 3
[3520] b: start, input: 3a
[4528] b: done, input: 3a
[4530] c: start, input: 3ab
[5536] c: done, input: 3ab
[5537] final result: 3abc

@rabbbit 用生成器确实是一个好办法，在可被取消的时间点添加 yield 。暂停后只要不调用 next() 方法就不会继续执行，这样就需要自己写一个执行器，调度的时机可以自己灵活控制，像 co 那种执行器是自动执行的，async 函数也是自动执行的。

给输出结果加个时间线：

[1] Starting task A with input: 1
[514] Starting task A with input: 2
[1010] Finishing task A with input: 1
[1537] Finishing task A with input: 2
[1539] Starting task B with input: 2A
[2019] Starting task A with input: 3
[2542] Finishing task B with input: 2A
[3023] Finishing task A with input: 3
[3024] Starting task B with input: 3A
[4027] Finishing task B with input: 3A
[4028] Starting task C with input: 3AB
[5038] Finishing task C with input: 3AB
[5040] Final result: 3ABC

用 RxJS 可以方便地实现，跟 AI 交流了一会后给出了如下代码：

import { Subject, from } from 'rxjs';
import { switchMap } from 'rxjs/operators';

// 模拟异步任务 A 、B 、C
const taskA = (input) => {
console.log(`Starting task A with input: ${input}`);
return new Promise((resolve) => {
setTimeout(() => {
console.log(`Finishing task A with input: ${input}`);
resolve(input + 'A');
}, 1000); // 假设任务 A 需要 1 秒钟
});
};

const taskB = (input) => {
console.log(`Starting task B with input: ${input}`);
return new Promise((resolve) => {
setTimeout(() => {
console.log(`Finishing task B with input: ${input}`);
resolve(input + 'B');
}, 1000); // 假设任务 B 需要 1 秒钟
});
};

const taskC = (input) => {
console.log(`Starting task C with input: ${input}`);
return new Promise((resolve) => {
setTimeout(() => {
console.log(`Finishing task C with input: ${input}`);
resolve(input + 'C');
}, 1000); // 假设任务 C 需要 1 秒钟
});
};

const subject = new Subject();

subject
.pipe(
switchMap((value) =>
from(taskA(value)).pipe(
switchMap((resultA) => from(taskB(resultA))),
switchMap((resultB) => from(taskC(resultB))),
),
),
)
.subscribe((result) => {
console.log('Final result:', result);
});

// 发出一些值
subject.next('1'); // 发出第一个值
setTimeout(() => subject.next('2'), 500); // 在 0.5 秒后发出第二个值，中止第一个任务并开始新的任务
setTimeout(() => subject.next('3'), 2000); // 在 2 秒后发出第三个值

输出结果：

Starting task A with input: 1
Starting task A with input: 2
Finishing task A with input: 1
Finishing task A with input: 2
Starting task B with input: 2A
Starting task A with input: 3
Finishing task B with input: 2A
Finishing task A with input: 3
Starting task B with input: 3A
Finishing task B with input: 3A
Starting task C with input: 3AB
Finishing task C with input: 3AB
Final result: 3ABC

实际上用全套 RxJS API 更简单，无奈 API 忘了好多

除了 VPN 、SSH 隧道，mTLS 也可以，甚至可以把这三个组合在一起用。