这是当前的实现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}}