Modify data in stream used with angular async pipe reactive way

Question

iam trying to use async pipes instead of manually subscribing in components, in case, iam just displaying data which i get from server, everything works fine. But what if i need to change displayed part of displayed data later, based on some event?

For example, i have a component which displays timetable, which can be later accepted by user. I create timesheet$ observable and i use it in my template with async pipe. For accepting i created subject, so i can emit, when user accepts the timesheet. But how do i combine these two streams(approvementChange$ and timesheet$), so the $timesheet gets updated? I was trying combineLatest, but it returns the latest values, so i cant decide, if the value from approvementChange stream is new or old. Any ideas how to solve this?

export class TimesheetComponent {
  errorMessage: string = '';
  timesheet$: Observable<Timesheet>;
  
  private approvementSubject = new Subject<TimesheetApprovement>();
  approvementChange$ = this.approvementSubject.asObservable();

  constructor(
    private planService: PlanService,
    private statusService: StatusService,
    private notifService: NotificationService
  ) {}

  this.timesheet$ = this.statusService.appStatus.pipe(
    switchMap((status) => {
      return this.planService.getTimesheet(
        status.selectedMonth,
        status.selectedAgency.agNum
      );
    })
  ); //get data every time user changed selected month

  approveTimesheet(isApproved: boolean = true) {
    const approvement: TimesheetApprovement = {
      isApproved: isApproved,
      approveTs: new Date(),
      isLock: true,
    };
    this.approvementSubject.next(approvement);
  }
}

BizzyBob · Accepted Answer · 2022-04-14 13:24:04Z

1

But what if i need to change displayed part of displayed data later, based on some event?

RxJS provides lots of operators for combining, filtering, and transforming observables.

You can use scan to maintain a "state" by providing a function that receives the previous state and the newest emission.

Let's look at a simplified example using a generic Item interface:

interface Item {
  id         : number;
  name       : string;
  isComplete : boolean;
}

We will use 3 different observables:

initialItemState$ - represents initial state of item after being fetched
itemChanges$ - represents modifications to our item
item$ - emits state of our item each time a change is applied to it

  private itemChanges$ = new Subject<Partial<Item>>();

  private initialItemState$ = this.itemId$.pipe(
    switchMap(id => this.getItem(id))
  );

  public item$ = this.initialItemState$.pipe(
    switchMap(initialItemState => this.itemChanges$.pipe(
      startWith(initialItemState),
      scan((item, changes) => ({...item, ...changes}), {} as Item),
    ))
  );

You can see we define item$ by piping the initialItemState$ to the itemChanges$ observable. We use startWith to emit the the initialItemState into the changes stream.

All the magic happens inside the scan, but the set up is really simple. We simply provide a function that accepts the previous state and the new change and returns the updated state of the item. (In this case, I'm just naively apply the changes to the previous state; it's possible this logic would need to be more sophisticated for your case.)

This solution is completely reactive. The end result is a clean item$ observable that will emit the updated state of the current item (based on id), whenever the id changes or the changes occur on the item.

Here's a StackBlitz where you can see this behavior.

answered Apr 14, 2022 at 13:24

BizzyBob

14.9k4 gold badges32 silver badges62 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

Vartex05 Over a year ago

Great solution, but there is one thing i dont understand - how it is possible, that the item$ observable emits, when itemChanges$ observable emits(when it sets item complete). public item$ = this.initialItemState$.pipe(switchMap.=>..) here it uses pipe and swtichmap on initialItemState$ so it should emit only when initialItemState$ emits. Can you give me some explanation or point me towards some resources?

BizzyBob Over a year ago

Your understanding is only partially correct. "it should emit only when initialItemState$ emits" - this isn't entirely true. It's true the switchMap will only emit after initialItemState$ emits, but, it doesn't require a new emission from initialItemState$ to emit more. You see, we pass a function to switchMap that returns an "inner observable". switchMap then emits the emissions from its inner observable, which in the case of our example here, is an observable that will startWith the initial state, then emit the result of our scan operation.

BizzyBob Over a year ago

switchMap is one of several "Higher order mapping operators". All of these operators maintain one or many "inner observables" and emit their emissions. The difference between them is the strategy they use when they receive more than one emission. This article has a pretty good explaination.

Vartex05 Over a year ago

thx a lot, now i get it :)

shutsman · Accepted Answer · 2022-04-14 13:10:25Z

1

you need to track only approvementChange

after that you can pick the latest timesheet via withLatestFrom

approvementChange$
.pipe(withLatestFrom(this.timesheet$))
.subscribe(([accepted, latestTimesheet]) => accepted ? save(latestTimesheet) : void 0)

answered Apr 14, 2022 at 13:10

shutsman

2,5101 gold badge17 silver badges26 bronze badges

Collectives™ on Stack Overflow

Modify data in stream used with angular async pipe reactive way

2 Answers 2

4 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

4 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related