如何用RxJS将Project按CurrentProjectPhase关联到Phase的Observable？

核心需求：将Project按CurrentProjectPhase关联到对应Phase，全程保持Observable流形式，避免嵌套订阅。

问题分析 #

你之前的代码存在两个关键问题：

• 使用tap操作符修改Phase的Projects$，但tap仅用于执行副作用，不会改变流的输出值，最终phases$并未携带修改后的Projects$。
• 直接复用this.projects$的管道，若projects$是冷Observable（比如HTTP请求），每次订阅Projects$都会重复发起请求，造成资源浪费。

可行实现 #

根据你的场景（两个HTTP服务返回一次性Observable），推荐使用forkJoin合并两个流，统一处理关联逻辑：

import { forkJoin, of } from 'rxjs';
import { map } from 'rxjs/operators';

// 合并 phases 和 projects 请求，待两者都完成后处理数据
this.phases$ = forkJoin([
  this.phaseService.getPhases(),
  this.projectService.getProjects()
]).pipe(
  map(([phases, projects]) => {
    // 遍历每个Phase，生成包含对应过滤后项目Observable的新对象
    return phases.map(phase => ({
      ...phase,
      Projects$: of(projects.filter(project => project.CurrentProjectPhase === phase.PhaseId))
    }));
  })
);

// 若需单独维护项目列表的Observable
this.projects$ = this.projectService.getProjects();

如果projects$是会持续更新的流（比如实时同步的项目列表），则改用combineLatest，确保phases$随projects$更新自动同步：

import { combineLatest, of } from 'rxjs';
import { map } from 'rxjs/operators';

this.projects$ = this.projectService.getProjects();

this.phases$ = combineLatest([
  this.phaseService.getPhases(),
  this.projects$
]).pipe(
  map(([phases, projects]) => {
    return phases.map(phase => ({
      ...phase,
      Projects$: of(projects.filter(project => project.CurrentProjectPhase === phase.PhaseId))
    }));
  })
);

关键说明 #

• forkJoin：适用于多个一次性Observable，等待所有流完成后发射结果数组，避免冗余请求。
• combineLatest：适用于有更新需求的场景，任一源流发射新值时，重新计算并发射最新的关联结果。
• of()：将过滤后的项目数组转为Observable，既满足Phase接口的类型要求，又避免重复发起HTTP请求。

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