javascript - Observable 中的 Async/Await

Question

如何在 Observable 中使用 async/await？ 使用此代码，我无法在可观察范围内触发取消订阅功能，因此不会清除间隔。

const { Observable } = require("rxjs");

const test = () => new Observable(async (subscriber) => {
  await Promise.resolve();

  const a = setInterval(() => {
    subscriber.next(Math.random());
    console.log("zz");
  }, 500);

  return () => {
    console.log("asdsad");
    clearInterval(a);
  };
});

const xyz = test().subscribe(console.log);


setTimeout(() => {
  xyz.unsubscribe();
}, 3000);

Answer 1

不支持 observable 中的 Async/Await。但是，它可以通过行为主体和异步嵌套函数来完成。

创建一个行为主体，将其转换为可观察对象(.asObservable())，执行异步嵌套函数，返回可观察对象。这是一个例子。

function getProgress() {

    // Change this value with latest details
    const value = new BehaviorSubject('10%');
    const observable = value.asObservable();

    // Create an async function
    const observer = async() => {
      // Perform all tasks in here
      const wait1 = await new Promise(resolve => setTimeout(resolve, 3000));
      value.next('66%');

      const wait2 = await new Promise(resolve => setTimeout(resolve, 3000));
      value.next('100%');

      // Complete observable
      value.complete();
    }

    // Call async function & return observable
    observer();
    return observable;
  }

它的可读性很强，而且很有魅力。