NestJS gRPC流处理：RxJS订阅Observable完成前返回空如何解决

问题本质 #

你的猜测完全正确，这是典型的异步时序问题：RxJS的subscribe回调是异步执行的，findAllRepos函数会在流还没发射任何值、也没触发complete回调之前，就同步返回了初始化的空results数组，只有后续流触发next回调的时候才会往数组里推值，所以你拿不到返回的结果。

解决方案 #

方案1：直接返回Observable流（推荐，适配NestJS gRPC生态）

NestJS的gRPC服务端原生支持返回Observable类型的响应，不需要自己转数组再返回。如果需要把多个流结果合并成单个数组返回，用toArray操作符即可：

import { toArray } from 'rxjs';

findAllRepos(): Observable<Repository[]> {
  return this.mainService.findAllRepos({}).pipe(
    toArray() // 等待流complete之后，把所有next发射的值合并为一个数组发射
  );
}

上层调用方可以直接订阅这个Observable，或者转成Promise用async/await获取结果。

方案2：转成Promise用async/await

如果你需要用同步写法拿结果，把函数改成异步，用lastValueFrom工具把Observable转成Promise：

import { lastValueFrom, toArray } from 'rxjs';

async findAllRepos(): Promise<Repository[]> {
  const observable = this.mainService.findAllRepos({}).pipe(
    toArray()
  );
  // 等待流执行完成，返回最终的数组结果
  return lastValueFrom(observable);
}

调用该函数时用await findAllRepos()就能拿到完整的结果数组。

注意事项 #

• 不要在函数内部手动subscribe后同步返回结果，这种写法违背异步逻辑的执行时序，永远无法拿到实时填充后的结果
• 如果你要给gRPC客户端返回服务端流，直接返回原始的Observable<Repository>即可，不需要转成数组，客户端会逐帧接收流数据

内容的提问来源于stack exchange，提问作者fahmiduldul