Angular multiple streams one observable

Question

I have data I am pulling from my server based on a specific date. So when the date changes I query from the server again to get the new data. I also have the ability to manually trigger a refresh from the server ( for cases when I pull data from so I need to re-pull )

private _date = new BehaviorSubject(new Date())
private _refresh = new BehaviorSubject<void>(undefined); 
readonly data$: Observable<Model[]> = combineLatest([this._date, this._refresh])
   .pipe( 
      switchMap(([date, refresh]) => this.dataService.get(date)),
      shareReplay(1) 
   )

My view is displaying data$ using an async pipe

I need to be able to manipulate the list in data$ so that I can keep track of modified items in the list.

Model {
   ...
   isModified: boolean
}

My model is like this so if I make a change to an element I want to be able to push that into the list of data ( remove old item -> replace with new modified item ) then have data$ emit the update list.

Then if the _date emits $data would emit that and everything modified would be wiped out.

I was trying not to mutate the items in the list directly so I can utilize ChangeDetectionStrategy.OnPush

Is there a way I can combine the steams of manually updating the list and _date updating the list but when pulling from the server that would take precedence.

Only maintaining one list should allow me to easier get aggregates based on the data. As items update my aggregate values would be updated based on the modified list. If I maintained a second list of modified items I would need to compare the two lists if I wanted the aggregate values in the modified state ( get all items in main list which aren't modified and all items from the modified list).

Batch updating should then just be as easy as

data$.pipe(
   filter(x => x.isModified), 
   switchMap(x => dataService.update(x))
)

Here is a stackblitz that I believe will help demonstrate.

Here you can see I have a component which allows for updating the name of the items returned from my observable. I would like to be able to modify those items and then have them stored back into the observable. This way if I wanted to get all modified item i could just apply a filter on that observable to get the modified items. Also then if you change the date that observable would repull from the server and the list of modified items would be wiped out.

See DataService.itemNameChanged

Could you give an example of your expected input & output? Then if the _date emits $data would emit that and everything modified would be wiped out. - this means that when _date emits, data$ should emit what it currently has and then remove everything that's modified? Could you elaborate on this? — Andrei Gătej
– Andrei Gătej, Commented Jul 29, 2020 at 21:10
@AndreiGătej I added a stackblitz that I hope will help clarify what I am looking to achieve. When _date emits we should repull from the server which would effectively remove all modified items as we now have a fresh list from the server — rpascal
– rpascal, Commented Jul 29, 2020 at 23:31

bernatsampera · Accepted Answer · 2020-07-30 13:15:23Z

To solve this problem you could use the scan operator to hold the value of the new items in the array.

Here a BehaviorSubject is created to emit the values of each new item modified, it's initialized with the first value of the default array.

  private itemModified = new BehaviorSubject<Model>(data[0]);

Every time that the base observables for dateRefresh$ are updated this value should be reseted.

  readonly dateRefreshHttp$: Observable<Model[]> = combineLatest([
    this._date,
    this._refresh
  ]).pipe(
    tap(() => this.itemModified.next(data[0])),
    switchMap(([date, refresh]) =>

The scan operator is added inside the switchMap, in order to restart the accumulate whenever it's changed. Note the debounce operator added to allow multiple input of characters without loosing focus.

   switchMap(([date, refresh]) =>
      combineLatest([from([data]), this.itemModified]).pipe(
        scan(
          (acc: any[], [data, itemModified], index: number) => {
            return acc.map(item =>
              item.id == itemModified.id ? itemModified : item
            );
          },
          [...data]
        )
      )
    ),
    debounceTime(300)
  );

Finally, a new value is emited through the itemNameChanged function.

  itemNameChanged(newName: string, item: Model) {
    const newItem: Model = {
      id: item.id,
      name: newName,
      isModified: true
    };

    this.itemModified.next(newItem);
  }

Live example, Stackblitz

Collectives™ on Stack Overflow

Angular multiple streams one observable

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related