这是当前的实现Observable[Symbol.asyncIterator]。
下面是一个在数组上实现Symbol.asyncIterator的基本示例：
const dummyPromise = (val, time) => new Promise(res => setTimeout(res, time * 1000, val));
const items = [1, 2, 3];
items[Symbol.asyncIterator] = async function * () {
yield * await this.map(v => dummyPromise(v, v));
}
!(async () => {
for await (const value of items) {
console.log(value);
}
})();
我理解生成器（同步生成器）的方式是它们是可暂停的函数，这意味着你可以立即请求一个值，然后在10秒后请求另一个值。异步生成器遵循相同的方法，只是它们产生的值是异步的，这意味着你必须等待它。
await。
例如：
const dummyPromise = (val, time) => new Promise(res => setTimeout(res, time * 1000, val));
const items = [1, 2, 3];
items[Symbol.asyncIterator] = async function * () {
yield * await this.map(v => dummyPromise(v, v));
}
const it = items[Symbol.asyncIterator]();
(async () => {
await it.next();
setTimeout(async () => {
console.log(await it.next());
}, 2000);
})();
回到
Observable的实现：
async function* coroutine<T>(source: Observable<T>) {
const deferreds: Deferred<IteratorResult<T>>[] = [];
const values: T[] = [];
let hasError = false;
let error: any = null;
let completed = false;
const subs = source.subscribe({
next: value => {
if (deferreds.length > 0) {
deferreds.shift()!.resolve({ value, done: false });
} else {
values.push(value);
}
},
error: err => { },
complete: () => { },
});
try {
while (true) {
if (values.length > 0) {
yield values.shift();
} else if (completed) {
return;
} else if (hasError) {
throw error;
} else {
const d = new Deferred<IteratorResult<T>>();
deferreds.push(d);
const result = await d.promise;
if (result.done) {
return;
} else {
yield result.value;
}
}
}
} catch (err) {
throw err;
} finally {
subs.unsubscribe();
}
}
根据我的理解：
values 用于跟踪同步值
如果你有of(1, 2, 3)，那么在它甚至到达while(true) { }之前，values数组将包含[1, 2, 3]。因为你正在使用for await (const v of ...)，
所以你会像执行it.next(); it.next(); it.next() ...一样请求值。
换句话说，一旦你可以从迭代器中消耗一个值，你就立即请求下一个值，直到数据生产者没有更多可提供的值为止。
deferreds 用于异步值
因此，在第一个it.next()时，values数组为空（意味着observable没有同步发出），因此它将回退到最后一个else，简单地创建一个promise并将其添加到deferreds中，然后等待该promise resolve或reject。
当observable最终发出时，deferreds不会为空，因此等待的promise将resolve为新到达的值。
const src$ = merge(
timer(1000).pipe(mapTo(1)),
timer(2000).pipe(mapTo(2)),
timer(3000).pipe(mapTo(3)),
);
!(async () => {
for await (const value of src$) {
console.log(value);
}
})();
StackBlitz
{{链接1:StackBlitz}}
