如何将包含Observable对象数组的对象转为Observable<MyObject>

核心逻辑是收集对象中所有嵌套的Observable，用forkJoin等待全部流完成后，把返回值回填到原对象结构中，最终输出完整的解包对象。

针对你给出的特定结构实现 #

import { forkJoin, map, Observable, of } from 'rxjs';

// 定义返回值类型
interface ObsItem {
  c: string;
  d: boolean;
}

interface MyObject {
  a: string;
  b: string;
  obs: ObsItem[];
}

const myObject = {
  a: 'foo',
  b: 'bar',
  obs: [{ c: 'foobar', d: of(true) }, { c: 'foobar', d: of(true) }]
};

// 转换函数
function transform(obj: typeof myObject): Observable<MyObject> {
  // 提取所有内部d字段的Observable
  const innerObservables = obj.obs.map(item => item.d);

  return forkJoin(innerObservables).pipe(
    map(dResults => ({
      ...obj,
      obs: obj.obs.map((item, idx) => ({
        ...item,
        d: dResults[idx]
      }))
    }))
  );
}

// 调用测试
transform(myObject).subscribe(result => {
  console.log(result);
  // 输出结果和你要求的结构完全一致
});

通用递归实现（适配任意嵌套Observable的对象） #

如果你的对象结构不固定，Observable可能出现在任意嵌套层级，可以用下面的通用递归函数自动解析所有Observable：

import { forkJoin, isObservable, map, Observable, of } from 'rxjs';

function deepForkJoin<T>(input: T): Observable<T> {
  // 输入本身是Observable直接返回
  if (isObservable(input)) {
    return input as Observable<any>;
  }
  // 输入是数组，递归处理每一项
  if (Array.isArray(input)) {
    return forkJoin(input.map(item => deepForkJoin(item))) as Observable<any>;
  }
  // 输入是普通对象，递归处理每个属性
  if (typeof input === 'object' && input !== null) {
    const entryObservables = Object.entries(input).map(([key, value]) => 
      deepForkJoin(value).pipe(map(res => [key, res]))
    );
    return forkJoin(entryObservables).pipe(
      map(entries => Object.fromEntries(entries))
    ) as Observable<any>;
  }
  // 普通值直接包装成Observable返回
  return of(input) as Observable<any>;
}

// 直接调用即可，不需要关心内部结构
deepForkJoin(myObject).subscribe(result => console.log(result));

补充说明 #

• 上述实现默认和forkJoin行为完全一致，会等待所有内部Observable都完成之后才一次性输出最终结果，注意：如果内部存在不会触发complete回调的流，forkJoin会一直等待，这种场景请改用combineLatest
• 如果需要内部Observable每次发射新值都更新最终对象，直接把代码里的forkJoin替换为combineLatest即可

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