具有异步函数和 toPromise 的 Observable RxJS 的意外行为是什么？

Question

当我只使用订阅方法时，它确实有效，但使用这段代码 - 我不明白结果。

const Observable = require("rxjs").Observable;
let i = 0;
const x = new Observable((o) => {
    setInterval(() => o.next(++i), 1000);
});

(async () => {
    while (true) {
        try {
            console.log("loop");
            console.log("value", await x.toPromise());
        } catch (e) {
            console.log(e);
        }
    }
})();
x.subscribe((value) => {
    console.log("subscribe", value);
});

这段代码的结果是：

loop
subscribe 2
subscribe 4
subscribe 6
subscribe 8
subscribe 10
subscribe 12
subscribe 14

发生什么事了？

它与使用 toPromise

的这种变体一样工作

function a() {
    x.toPromise().then((value) => {
        console.log("promise", value);
        a();
    }).catch((e) => {
        console.log("error", value);
    });
}
a();

Answer 1

toPromise() 在 Observable 完成时执行。由于您的可观察对象从未真正完成，因此它不会执行。使用 take(1) 强制它在 observable 完成之前发出值。

const Observable = require("rxjs").Observable;
let i = 0;
const x = new Observable((o) => {
    setInterval(() => o.next(++i), 1000);
});

(async () => {
    while (true) {
        try {
            console.log("loop");
            console.log("value", await x.take(1).toPromise());//here
        } catch (e) {
            console.log(e);
        }
    }
})();
x.subscribe((value) => {
    console.log("subscribe", value);
});

输出：

loop
subscribe 2
value 1
loop
subscribe 4
value 5
loop
subscribe 7
value 9
loop
subscribe 11
value 14

至于值：

take() 将在至少发出一个值后完成，无论源可观察对象是否完成。所以这真的取决于下一次 toPromise() 被调用时 observable 发出的值是什么

具有异步函数和 toPromise 的 Observable RxJS 的意外行为是什么？

What the unexpected behavior Observable RxJS with async functions and toPromise?

rxjs

rxjs6