是否可以在循环中组合 Angular NgRx 相关选择器/可观察对象而不取消订阅更新？

Question

我们正在将 Angular (v18) NgRx 用于一个动作频繁的大型应用程序。对于新功能，我们将获得一个值数组，并且需要为从该数组返回的每个项目调用另一个

selector

，以尝试创建视图模型。我们所说的

selector

是某种聚合器，它从

AppState

的众多切片中提取数据。由于它聚合了如此多的数据，因此需要相当长的时间才能完全加载。任何在

unsubscribe

上执行

complete

或

Observable

的解决方案都可能不被允许。

失败的尝试1

export const selectNewFeatureViewModels = createSelector(
  (state: AppState) => state, // <-- dependence on AppState; if literally anything changes this emits
  selectNewFeatureRecords,
  (state, newFeatureRecords) => {
    return newFeatureRecords?.map((newFeatureRecord) => {
      return {
        newFeatureRecord,
        aggregatorRecord: selectAggregatorRecord({
          key: newFeatureRecord.key,
        })(state),
      };
    });
  }
);

另一篇 Stack Overflow 帖子对这一尝试有所帮助。
NGRX 选择器：另一个选择器中的工厂选择器，在 createSelector 方法中没有 prop

尝试 1 确实有效，但它存在与

AppState

绑定的问题。用户实际上会更改应用程序上的任何内容，并且会发出此

selector

，这会导致其他问题。

---

我们还尝试从

Dictionary

创建

Map

/

selectAggregatorRecord

（如果这样会更容易），但每个值都会返回

Observable

。

Map<string, Observable<AggregatorRecord>>

我们也无法进行编译，因为

selector

需要参数。例如

key

。

// broken
export const selectAggregatorMap = (keys: string[]) => createSelector(
  ...keys.map(key => selectAggregatorRecord({key})),
  (value) => value
);

尝试2失败

viewModels: {
  newFeatureRecord: NewFeatureRecord;
  aggregatorRecord: AggregatorRecord;
}[];

ngOnInit(): void {
  this.newFeatureFacade
  .getNewFeatureRecords()
  .pipe(
    tap((newFeatureRecords) => {
      newFeatureRecords.forEach((newFeatureRecord) => {
        this.aggregatorRecordFacade
        .getAggregatorRecord({
          key: newFeatureRecord.key,
        })
        .pipe(
          debounceTime(1000),
          filter(
            (aggregatorRecord) =>
              !!aggregatorRecord?.field1 &&
              !!aggregatorRecord?.field2 &&
              !!aggregatorRecord?.field3
          ),
          map((aggregatorRecord) => {
            return {
              newFeatureRecord,
              aggregatorRecord,
            };
          }),
        ).subscribe((viewModel) => {
          if(this.viewModels?.length < 10){
            this.viewModels.push(viewModel);
            this.changeDetectorRef.markForCheck();
          }
        });
      });
    })
  ).subscribe(() => {
    this.viewModels = [];
    this.changeDetectorRef.markForCheck();
  });
}

尝试2也“奏效”了。它还有一些其他问题，但并不是很干净。我们不想将所有这些逻辑保留在我们的组件中。愚蠢的组件 FTW！嵌套的

subscribe

也让我们紧张。特别是因为我们

subscribing

想要显示更多记录。

getNewFeatureRecords

可以返回 100 条记录，但我们只想显示 10 条。

Facade

类用于在 NgRx 元素和 Angular 组件之间创建分离。组件和 NgRx 之间的所有交互都必须通过这些 Facade 来调度操作并从存储中进行选择。
https://www.rainerhahnekamp.com/en/ngrx-best-practices-series-4-facade-pattern/

@Injectable({ providedIn: 'root' })
export class NewFeatureFacade {
  constructor(private store: Store) {}

  getNewFeatureRecords(): Observable<NewFeatureRecord[]> {
    return this.store.select(selectNewFeatureRecords);
  }
}

@Injectable({ providedIn: 'root' })
export class AggregatorRecordFacade {
  constructor(private store: Store) {}

  getAggregatorRecord({key}: {key: string}): Observable<AggregatorRecord> {
    return this.store.select(selectAggregatorRecord({key});
  }
}

---

我也研究了其他 NgRx 运算符。

• toArray

• mergeAll

• forkJoin

• switchMap

• mergeMap

https://www.learnrxjs.io/learn-rxjs/operators

我们真的很想彻底清理它。你能帮忙吗？

Answer 1

恭喜！请提前原谅我的“我的英语”。 不确定我是否正确地重现了您的情况，但让我们按顺序进行：

应用服务：

import { Injectable } from "@angular/core";
import { BehaviorSubject, Observable, Subject } from "rxjs";

@Injectable({ providedIn: 'root' })
export class AppService {
  constructor() {

  }

  getNewFeatureRecords = (): Observable<Array<any>> => new BehaviorSubject<Array<any>>([1, 2, 3])
  getAggregatorRecord = ({ key }: { key: string }): Observable<any> => new BehaviorSubject<any>({ key })
}

应用程序组件：

import { ChangeDetectorRef, Component, OnDestroy, OnInit } from '@angular/core';
import { RouterOutlet } from '@angular/router';
import { combineLatest, debounceTime, filter, map, Subject, switchMap, takeUntil } from 'rxjs';
import { AppService } from './app.services';

@Component({
  selector: 'app-root',
  standalone: true,
  imports: [RouterOutlet],
  templateUrl: './app.component.html',
  styleUrl: './app.component.scss'
})
export class AppComponent implements OnInit, OnDestroy {
  private readonly unsubscribe$: Subject<void> = new Subject<void>()

  public viewModels: Array<RecordModel> = new Array<RecordModel>()

  private getNewFeatures = () => this._appService.getNewFeatureRecords()
  private getAggregatorState = ({ key }: { key: string }) => this._appService.getAggregatorRecord({ key })

  constructor(
    private _appService: AppService,
    private _cdr: ChangeDetectorRef
  ) {

  }


  ngOnInit(): void {
    this.getNewFeatures().pipe(
      map(el => el.map(el => this.getAggregatorState({ key: el }))),
      map(el => el.map(el => el.pipe(
        debounceTime(1000),
        filter(el => !!el),
        map((el, index) => new RecordModel(index, el)),
        map(model => {
          console.log(model)
          if (this.viewModels.length < 10) { this.viewModels.push(model); this._cdr.markForCheck() }
          return model
        })
      ))),
      // combineLatest(el) – In case you need a set of accumulated values. Maybe it will be useful for building connections
      // merge(...el) – if the order of emit values ​​is not important to you
      switchMap((el) => combineLatest(el)),
      map(el => {
        this.viewModels = []
        this._cdr.markForCheck()
      }),
      takeUntil(this.unsubscribe$)
    ).subscribe()
  }

  ngOnDestroy(): void {
    this.unsubscribe$?.next()
    this.unsubscribe$?.complete()
  }
}

export class RecordModel {
  newFeatureRecord: any
  aggregatorRecord: any

  constructor(newFeatureRecord: any, aggregatorRecord: any) {
    this.newFeatureRecord = newFeatureRecord
    this.aggregatorRecord = aggregatorRecord
  }
}

在上面的示例中，我将您的数据流合并到单个订阅中。如果有必要，您可以添加更多逻辑来在每个 .pipe() 中为相应的 observable 执行计算。

如果需要从子可观察对象中获取所有发出的值，可以使用combineLatest运算符。在这种情况下，您将收到一个数据数组，其中包含管道中内部可观察结果的结果

如果从内部可观察对象发出数据的顺序对您来说并不重要，那么您可以更喜欢 merge() 运算符。

我希望我正确理解您的问题并帮助解决它。