在源可观察的每次发射之后结合可观察的最新值

Question

我有一个源 Observable（实际上是一个 Subject），当该 Observable 发出一个值时，我需要启动一个值的加载，这会导致另外 2 个 observable 被更新，当加载完成时，取source 并将其与 2 个“相关” observables 的最新值结合起来，但只有源 observable 发出之后的值。因此，如果 2 个依赖的可观察对象在源发出时具有值，我需要等待直到 2 个依赖对象获得更新，然后再发出所有 3 个可观察对象中的最新。我尝试过使用 withLatestFrom、switchMap 和 combineLatest 的各种咒语，但没有什么能提供我想要的输出：

这将发出正确形状的值，但不是在正确的时间，它使用的是 SelectExpenseSummary 完成其操作之前的值，并且费用详细信息 $ 和费用收据 $ 已更新。

      this.editExpenseSubject.pipe(
        takeWhile(() => this.active),
         tap(x => this.userStore.dispatch(new SelectExpenseSummary(x))),
         tap(x => this.log.verbose({type: 'ExpensesFeatureComponent:editExpenseSubject:tap', x})),
         withLatestFrom(
          this.expenseDetails$,
          this.expenseReceipts$,
          )
        )
      .subscribe(x => {
        this.log.verbose({type: 'ExpensesFeatureComponent:editExpenseSubject:subscribe', x});
      });

这会发出很多次，最后一次具有正确的值，这没关系，但发出错误的形状，它缺少输出中可观察到的源：

      this.editExpenseSubject.pipe(
        takeWhile(() => this.active),
         tap(x => this.userStore.dispatch(new SelectExpenseSummary(x))),
         tap(x => this.log.verbose({type: 'ExpensesFeatureComponent:editExpenseSubject:tap', x})),
         switchMap(() =>
            combineLatest(this.expenseDetails$,
            this.expenseReceipts$,
            )
          )
        )
      .subscribe(x => {
        this.log.verbose({type: 'ExpensesFeatureComponent:editExpenseSubject:subscribe', x});
      });

这具有正确的输出形状，具有所有 3 个可观察对象，但它在订阅中从不发出任何内容，我猜是因为在内部使用了 editExpenseSubject：

      this.editExpenseSubject.pipe(
        takeWhile(() => this.active),
         tap(x => this.userStore.dispatch(new SelectExpenseSummary(x))),
         tap(x => this.log.verbose({type: 'ExpensesFeatureComponent:editExpenseSubject:tap', x})),
         switchMap(x =>
          combineLatest(
            this.editExpenseSubject
            , this.expenseDetails$
            , this.expenseReceipts$
          )
        )
      )
      .subscribe(x => {
        this.log.verbose({type: 'ExpensesFeatureComponent:editExpenseSubject:subscribe', x});
      });

谁能指出正确的咒语来得到我想要的东西，最好能详细解释原因。

---

更新以获取更多信息（我认为这可以帮助您完成整个流程）：

    this.expenseDetails$ = this.expenseStore.pipe(
      select(fromExpenses.getExpenseLines)
    );
    this.expenseReceipts$ = this.expenseStore.pipe(
      select(fromExpenses.getExpenseReceipts)
    );

export const getExpenseLines = createSelector(StateFeatureSelector, x => x.expenseLines);
export const getExpenseReceipts = createSelector(StateFeatureSelector, x => x.expenseReceipts);

export interface State {
  expenseSummaries: Array<IExpenseSummary>;
  expenseLines: Array<IExpenseLine>;
  expenseReceipts: Array<IExpenseReceipt>;
  selectedUser: IDirectReport;
  selectedExpenseSummary: IExpenseSummary;
}


  @Effect() LoadExpenseLines$ = this.actions$.pipe(
    ofType<fromExpense.LoadExpenseLines>(fromExpense.ActionTypes.LoadExpenseLines)
    , tap(action => this.log.debug({type: 'ExpenseEffects:LoadExpenseLines$:filtered', action}))
    , mergeMap(x =>
      this.service.getExpenseLines(x && x.payload)
        .pipe(
          tap(receipts => this.log.debug({type: 'ExpenseEffects:getExpenseLines', receipts}))
          , map(lineItems => new fromExpense.SetExpenseLines(lineItems))
        )
    )
  );
  @Effect() LoadExpenseReceipts$ = this.actions$.pipe(
    ofType<fromExpense.LoadExpenseReceipts>(fromExpense.ActionTypes.LoadExpenseReceipts)
    , tap(action => this.log.debug({type: 'ExpenseEffects:LoadExpenseReceipts$:filtered', action}))
    , mergeMap(x =>
      this.service.getExpenseReceipts(x && x.payload)
        .pipe(
          tap(receipts => this.log.debug({type: 'ExpenseEffects:getExpenseReceipts', receipts}))
          , map(receipts => new fromExpense.SetExpenseReceipts(receipts))
        )
    )
  );

  @Effect() SelectExpenseSummary$ = this.actions$.pipe(
    ofType<fromExpense.SelectExpenseSummary>(fromExpense.ActionTypes.SelectExpenseSummary)
    , tap(action => this.log.debug({type: 'ExpenseEffects:SelectExpenseSummary$:filtered', action}))
    , mergeMap(x =>
        [
          new fromExpense.LoadExpenseLines(x.payload)
          , new fromExpense.LoadExpenseReceipts(x.payload)
        ]
    )
  );

export class SelectExpenseSummary implements Action {
  readonly type = ActionTypes.SelectExpenseSummary;
  constructor(public payload: IExpenseSummary) {}
}

export class LoadExpenseLines implements Action {
  readonly type = ActionTypes.LoadExpenseLines;
  constructor(public payload: IExpenseSummary) {}
}
export class SetExpenseLines implements Action {
  readonly type = ActionTypes.SetExpenseLines;
  constructor(public payload: Array<IExpenseLine>) {}
}

export class LoadExpenseReceipts implements Action {
  readonly type = ActionTypes.LoadExpenseReceipts;
  constructor(public payload: IExpenseSummary) {}
}

export class SetExpenseReceipts implements Action {
  readonly type = ActionTypes.SetExpenseReceipts;
  constructor(public payload: Array<IExpenseReceipt>) {}
}

export function reducer (state = initialState, action: actions.ActionsUnion): State {
  switch (action.type) {
// ...  other actions cut
    case actions.ActionTypes.SetExpenseLines:
      return {
        ...state,
        expenseLines: action.payload && [...action.payload] || []
      };
    case actions.ActionTypes.SetExpenseReceipts:
      return {
        ...state,
        expenseReceipts: action.payload && [...action.payload] || []
      };
    default:
      return state;
  }
}

// from the service class used in the effects

  getExpenseLines(summary: IExpenseSummary): Observable<Array<IExpenseLine>> {
    this.log.debug({type: 'ExpenseService:getExpenseLines', summary, uri: this.detailUri});
    if (summary) {
      return this.http
        .post<Array<IExpenseLine>>(this.detailUri, {ExpenseGroupId: summary.ExpReportNbr})
        .pipe(
          tap(data => this.log.verbose({ type: 'ExpenseService:getExpenseLines:reponse', data }))
        );
    } else {
      this.log.log({ type: 'ExpenseService:getExpenseLines null summary'});
      return of<Array<IExpenseLine>>([])
      .pipe(
        tap(data => this.log.verbose({ type: 'ExpenseService:getExpenseLines:reponse', data }))
      );
    }
  }
  getExpenseReceipts(summary: IExpenseSummary): Observable<Array<IExpenseReceipt>> {
    this.log.debug({type: 'ExpenseService:getExpenseReceipts', summary, uri: this.receiptUri});
    if (summary) {
      return this.http
      .post<Array<IExpenseReceipt>>(this.receiptUri, {ExpenseGroupId: summary.ExpReportNbr})
      .pipe(
        tap(data => this.log.verbose({ type: 'ExpenseService:getExpenseReceipts:reponse', data }))
      );
    } else {
      this.log.log({ type: 'ExpenseService:getExpenseReceipts null summary'});
      return of<Array<IExpenseReceipt>>([])
      .pipe(
        tap(data => this.log.verbose({ type: 'ExpenseService:getExpenseReceipts:reponse', data }))
      );
    }
  }

Answer 1

我从@Reactgular 的假设出发：

您想从主题中获取一个发射值，然后从其他 2 个可观察对象发射它们的最新值。输出是一个包含 3 个项目的数组。

您希望 2 个内部 observable 在外部 observable 发射后获取新数据，而您只需要它们的第一个值。

从那里，想到的运算符是 skip ：它会跳过您要求跳过的值的数量，然后继续。

I have made a sandbox 让您看到它的实际效果：如您所见，两个 observable 都是从单一来源更新的，并且您只获得一个事件，即每个 observable 的最后一个值。

相关代码：

import { BehaviorSubject, combineLatest, fromEvent } from "rxjs";
import {
  delayWhen,
  skip,
  distinctUntilKeyChanged,
  distinctUntilChanged,
  skipUntil,
  switchMap,
  map
} from "rxjs/operators";

const counts = {
  source: 0,
  first: 0,
  second: 0
};

const source$ = new BehaviorSubject("source = " + counts.source);
const first$ = new BehaviorSubject("first = " + counts.first);
const second$ = new BehaviorSubject("second = " + counts.second);

// Allow the source to emit
fromEvent(document.querySelector("button"), "click").subscribe(() => {
  counts.source++;
  source$.next("source = " + counts.source);
});

// When the source emits, the others should emit new values
source$.subscribe(source => {
  counts.first++;
  counts.second++;
  first$.next("first = " + counts.first);
  second$.next("second = " + counts.second);
});

// Final result should be an array of all observables

const final$ = source$.pipe(
  switchMap(source =>
    first$.pipe(
      skip(1),
      switchMap(first =>
        second$.pipe(
          skip(1),
          map(second => [source, first, second])
        )
      )
    )
  )
);

final$.subscribe(value => console.log(value));

Answer 2

如果我理解正确的话......

您想从主题中获取一个发射值，然后从其他 2 个可观察对象发射它们的最新值。输出是一个包含 3 个项目的数组。

您希望 2 个内部 observable 在外部 observable 发射后获取新数据，而您只需要它们的第一个值。

 this.editExpenseSubject.pipe(
        switchMap(editExpense =>
            forkJoin(
               this.expenseDetails$.pipe(first()),
               this.expenseReceipts$.pipe(first()),
            ).pipe(
               map(values => ([editExpense, ...values]))
            )
          )
        )

您可以使用 switchMap() 以便它调用内部 observable 上的 subscribe，并使用 forkJoin() 来获取最终值。添加 first() 运算符以将可观察值限制为 1 个值。然后使用map() 将外部值放回结果数组中。