RxJS使用merge与of时lastValueFrom永久挂起问题排查

调用逻辑 #

框架侧对I18nLoader的load方法调用逻辑如下：

if (!translations) {
  translations = await this.loader.load();
}
if (translations instanceof Observable) {
  this.translationsSubject.next(
    await lastValueFrom(translations.pipe(take(1))),
  );
} else {
  this.translationsSubject.next(translations);
}

自定义Loader实现 #

为支持翻译内容动态更新，我实现了返回Observable的load方法，核心逻辑是初始化时返回翻译内容，同时监听localizationService事件触发重新拉取：

private events: Subject<void> = new Subject();

localizationService.subscribe(() => {
  this.events.next();
});

async parseTranslations() {
  // 通过Prisma查询数据库并返回翻译对象
}

async load() {
  return merge(
    of(await this.parseTranslations()),
    this.events.pipe(switchMap(() => this.parseTranslations())),
  );
}

观察到的现象 #

• 初始时translations始终为undefined，会触发load方法调用；
• 按预期，Observable发出第一个值后，take(1)应完成流，lastValueFrom结束等待；
• parseTranslations能正常返回查询结果；
• 即使简化load方法为以下形式，lastValueFrom仍会永久挂起：

async load() {
  return of(await this.parseTranslations())
}

问题原因分析 #

核心问题在于**load方法被定义为async函数，导致返回值被包装成Promise<Observable<Translation>>，而非直接返回Observable<Translation>**。这种不符合接口预期的返回类型，可能导致内部订阅逻辑出现异常；同时async函数与Observable混合使用时，可能出现流的初始化时机与NestJS启动生命周期不匹配的情况，导致lastValueFrom无法及时接收到值。

解决方案 #

1. 移除load方法的async修饰符，直接返回Observable；
2. 使用from()操作符将parseTranslations返回的Promise转换为Observable（适配异步查询逻辑）；
3. 修正load方法的返回类型以匹配NestJS的I18nLoader接口定义。

修正后的代码：

private events: Subject<void> = new Subject();

constructor(private localizationService: LocalizationService) {
  this.localizationService.subscribe(() => {
    this.events.next();
  });
}

private async parseTranslations(): Promise<Record<string, any>> {
  // Prisma数据库查询逻辑，返回翻译对象
  return this.prisma.translation.findMany({
    // 查询条件
  }).then(res => /* 转换为所需的翻译结构 */);
}

load(): Observable<Record<string, any>> {
  return merge(
    from(this.parseTranslations()),
    this.events.pipe(switchMap(() => from(this.parseTranslations())))
  );
}

同时，优化调用逻辑中translations的赋值（无需await，更符合Observable返回类型的规范）：

if (!translations) {
  translations = this.loader.load();
}

修改后，load方法直接返回Observable，from()会正确处理parseTranslations的异步结果，merge后的流会在初始查询完成后立即发出第一个值，take(1)会完成流，lastValueFrom就能正常获取值并结束等待。

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