RxJS 等待订阅 Observable 完成

Question

我正在尝试使用 NestJS 实现 gRPC，并使用 RxJS Observerable 来处理服务器流。

在下面的代码中，我尝试将值从 observable 放入 results 数组。函数findAllRepos 返回空列表 而subscribe 中的console.log(value) 正确打印所有结果。

  findAllRepos(): Repository[] {
    const results: Repository[] = [];

    const observable = this.mainService.findAllRepos({});
    observable.subscribe({
      next: (value) => {
        console.log(value);
        results.push(value);
      },
      error: (err) => console.log(err),
      complete: () => console.log(results)
    });

    return results;
  }

我认为问题在于函数在subscribe 完成之前返回值。有什么解决方案可以解决这个问题吗？谢谢！

Answer 1

问题

我认为问题在于函数在订阅完成之前返回值。

这些函数（next、error、&complete）都将在未来某个时间被可观察的（不是你）调用。当您深入了解 RxJS 的核心内容时，您可以对这些函数的调用方式进行一些控制，但为了简单起见，最好想象一下这是由 RxJS 库不透明地处理的。

因此，没有办法做你想做的事。

问题比这更糟糕。您遇到的问题存在于同步和异步代码之间的任何交互。您不能同步运行异步代码。在单线程环境（如 JavaScript）中，试图让一段同步的代码等待会立即使整个程序死锁。

考虑以下代码：您可能希望它输出 "a equals 0" 1000 毫秒，然后开始输出 "a equals 1" 永远。然而，实际发生的情况是，setTimeout 中的代码将永远没有机会运行，因为线程将陷入无限循环打印 "a equals 0"

// Start with a = 0
let a = 0;
// After 1 second, set a = 1
setTimeout(() => a = 1, 1000);

// loop forever and print something based on value of a
while(true){ 
  if(a == 0) console.log("a equals 0");
  else console.log("a equals 1");
}

解决方案

管理异步代码的两种最流行的方法是通过 Promise 或 observables。如果你想让一个函数异步返回一些东西，那么让它返回一个 promise 或 observable。

在你的情况下：

findAllRepos(): Observable<Repository[]> {
  
  const observable = this.mainService.findAllRepos({});
  return observable.pipe(
    tap(value => console.log("Repository value being added to array: ", value)),
    toArray(),
    tap({
      next: value => console.log("Result Array (Repository[]) : ", value),
      error: console.log,
      complete: () => console.log("findAllRepos observable complete")
    })
  );

}

然后在别处获取实际值：

findAllRepos().subscribe(repositories => {
  /* Do something with your array of repositories */
});

Answer 2

拥有带有订阅的方法不是很好的做法，您将使响应式代码完全不响应式。为什么不使用类似的东西

findAllRepos(): Observable<Repository[]> {
    return this.mainService.findAllRepos({}).pipe(catchError((e) => {
        console.log('e ', e);
    }), finalize(() => {
        console.log('completed');
    }))
}

然后findAllRepos().subscribe 在客户端？您将在此方法中拥有更大的灵活性，您可以从多个位置调用此方法并根据用例映射您想要的结果。