Is it possible to combine Angular NgRx dependent selectors/Observables in a loop without unsubscribing from updates?

Question

We're using Angular (v18) NgRx for a large app with actions firing a lot. For a new feature we will get an array of values and we need to call another selector for each item returned from that array in an attempt to create a view model. The selector we are calling is some kind of aggregator that pulls data from numerous slices of the AppState. Since it's aggregating so much data it takes a decent amount of time to completely load. Any solution that does unsubscribe or complete on the Observable will likely not permitted.

Failed Attempt 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),
      };
    });
  }
);

Another Stack Overflow post helped with this attempt.
NGRX selectors: factory selector within another selector without prop in createSelector method

Attempt 1 did work, but it had a problem of being tied to AppState. Users would change literally anything on the app and this selector would emit, which caused other problems.

---

We also tried creating a Dictionary/Map from the selectAggregatorRecord if that would've been easier, but each value is returning an Observable.

Map<string, Observable<AggregatorRecord>>

We also can't get those attempts to compile since the selector requires parameters. e.g. key.

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

Failed Attempt 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();
  });
}

Attempt 2 also did "work". It has some other problems, but it's not really clean. We don't want to keep all this logic inside our components. Dumb components FTW! The nested subscribe makes us nervous too. Especially since we're subscribing to more records then we want to display.

getNewFeatureRecords could return 100's of records, yet we only want to show 10.

The Facade classes are used to create separation between NgRx elements and Angular components. All interactions between components and NgRx have to go through these Facades for dispatching actions and selecting from the store.
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});
  }
}

---

I looked at these other NgRx operators too.

• toArray
• mergeAll
• forkJoin
• switchMap
• mergeMap

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

We really want to clean this up a lot. Can you help?

Answer 1

Congratulations! Forgive me in advance for "my English". Not sure if I recreated your situation correctly, but let's go in order:

AppService:

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 })
}

AppComponent:

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
  }
}

In the example above, I've combined your data streams within a single subscription. If necessary, you can add more logic to perform calculations in each .pipe() for the corresponding observable.

You can use the combineLatest operator if you need to get all emitted values ​​from child observables. In this case, you will receive a data array with results from internal observables in the pipeline

If the order of emitting data from the internal observable does not matter to you, then you can prefer the merge() operator.

I hope that I correctly understood your problem and helped to solve it.