RxJS 中高效整合与处理多数据流的策略

// 任务接口
export interface Task {
  goal_id: string; // 关联的目标ID
  name: string;
  description: string;
  priority: string;
  taskDate: string; // 任务日期，格式 YYYY-MM-DD
  id: string; // 任务ID
}

// 目标接口
export interface Goal {
  name: string;
  isMainGoal: boolean;
  details: string;
  category: string; // 目标类别
  lifeArea: string;
  creationDate: string;
  priority: string;
  endDate: Date;
  id: string; // 目标ID
}

class MyService {
    // 假设 tasksS 和 goalsS 是返回相应集合 Observable 的服务
    // tasksS.tasksCollection() 返回 Observable<Task[]>
    // goalsS.goalsCollection() 返回 Observable<Goal[]>

    getTasksByCategory(category:string):Observable<any> {
        const daysFromThisWeek = this.getDaysFromThisWeek(); // 获取本周日期列表
        return forkJoin({
          tasks: this.tasksS.tasksCollection(), // 获取所有任务
          goals: this.goalsS.goalsCollection(), // 获取所有目标
        })
        // !!! 第一次操作：处理 Goals !!!
        .pipe(
          // 1. 筛选指定类别的目标
          map(({ tasks, goals }) => { // 此时 tasks 和 goals 都可用
            return goals.filter((item:any) => item.category === category);
          }),
          // 2. 获取筛选后目标的 ID 数组
          map((goals:any) => { // 此时输入只有上一步筛选出的 goals
            const goalsIDs = goals.map((item:any) => item.id);
            return goalsIDs; // 此时只返回 goalsIDs
          })
        )
        // !!! 第二次操作：处理 Tasks !!!
        .pipe( // 这是一个新的 pipe 链，输入是上一个 pipe 的输出 (goalsIDs)
          // 问题：这个 map 接收的参数是 goalsIDs，tasks 已经丢失！
          map(({ tasks, goalsIDs }) => { // 编译错误或运行时错误，tasks 未定义
            let modArr = [] as any;
            goalsIDs.forEach((goalId:any) => {
              const forModArr = tasks.filter((task:any) => task.goal_id === goalId);
              modArr = modArr.concat(forModArr);
          })
          return modArr;
        }),
        map(tasksArr => {
          let finalTasks = [] as any;
          daysFromThisWeek.forEach((day:any) => {
              const forFinalTasks = tasksArr.filter((task:any) => task.taskDate === day);
              finalTasks = finalTasks.concat(forFinalTasks.length);
          })
          return finalTasks;
        })
        )
    }

    getDaysFromThisWeek(): string[] {
        let daysArr: string[] = [];
        // 假设 dayjs 已导入并可用
        for(let i=1; i<=7; i++) {
          daysArr.push(dayjs().startOf('week').add(i, "day").format('YYYY-MM-DD'));
        }
        return daysArr;
    }
}

import { Observable, forkJoin } from 'rxjs';
import { map } from 'rxjs/operators';
import dayjs from 'dayjs'; // 假设 dayjs 已导入

// 假设 Task 和 Goal 接口已定义

class MyService {
    // 假设 tasksS 和 goalsS 是返回相应集合 Observable 的服务
    // tasksS.tasksCollection() 返回 Observable<Task[]>
    // goalsS.goalsCollection() 返回 Observable<Goal[]>
    // 注意：原始问题中存在方法名混淆，这里假设 goalsS.goalsCollection() 获取 Goal[]，
    // tasksS.tasksCollection() 获取 Task[]。请根据实际服务方法名进行调整。

    constructor(private goalsS: any, private tasksS: any) {} // 注入服务

    getTasksByCategory(category: string): Observable<number[]> { // 明确返回类型
        const daysFromThisWeek = this.getDaysFromThisWeek();

        // 1. 预处理 goals 数据流，只提取出符合条件的 goal ID
        const goalIds$: Observable<string[]> = this.goalsS.goalsCollection().pipe(
            map((goals: Goal[]) =>
                goals
                    // 筛选指定类别的目标
                    .filter((goal: Goal) => goal.category === category)
                    // 提取这些目标的 ID
                    .map((goal: Goal) => goal.id)
            )
        );

        // 2. 获取所有任务数据流
        const tasks$: Observable<Task[]> = this.tasksS.tasksCollection();

        // 3. 使用 forkJoin 合并预处理后的 goalIds$ 和 tasks$
        return forkJoin({
            goalIds: goalIds$, // 现在 goalIds$ 会直接发出 string[]
            tasks: tasks$,     // tasks$ 会发出 Task[]
        }).pipe(
            // 4. 在一个 pipe 链中处理所有合并后的数据
            // 获取与 goal ID 匹配的任务
            map(({ tasks, goalIds }) => { // 此时 tasks 和 goalIds 都可用
                let matchedTasks: Task[] = [];
                goalIds.forEach((goalId: string) => {
                    const forMatchedTasks = tasks.filter((task: Task) => task.goal_id === goalId);
                    matchedTasks = matchedTasks.concat(forMatchedTasks);
                });
                return matchedTasks; // 返回匹配到的任务数组
            }),
            // 5. 统计匹配任务在当前周每天的数量
            map((tasksArr: Task[]) => { // 接收上一步的 matchedTasks
                let finalTasksCounts: number[] = [];
                daysFromThisWeek.forEach((day: string) => {
                    const dailyTasks = tasksArr.filter((task: Task) => task.taskDate === day);
                    finalTasksCounts = finalTasksCounts.concat(dailyTasks.length);
                });
                return finalTasksCounts; // 返回每天任务数量的数组
            })
        );
    }

    /**
     * 获取当前周（从周日开始）每天的日期字符串数组。
     * @returns 格式为 'YYYY-MM-DD' 的日期字符串数组。
     */
    private getDaysFromThisWeek(): string[] {
        let daysArr: string[] = [];
        for(let i = 0; i < 7; i++) { // 通常周日是每周的第一天，索引为0
          daysArr.push(dayjs().startOf('week').add(i, "day").format('YYYY-MM-DD'));
        }
        return daysArr;
    }
}