Create observables from a dynamic array

Question

I have something like this currently

onValueChange() {
  let someObj = //dynamically created
  this.processUsers(someObj);
}

processUsers(someObj) {
  from(Object.keys(someObj))
    .pipe(
      concatMap(key => this.getUsers(key)
    )
    .subscribe(res => ...)
}

onValueChange is a method called when something in a form changes and this triggers the method processUsers with an obj that was constructed when onValueChange is called. processUsers then uses this obj to do some async tasks.

This works OK if there is a single trigger but if when processUsers is called consecutively with different objects, the async tasks gets messed up.

So, how can I use rxjs to complete processUsers to be done with the current obj before taking in the next? My first thought is to use an array that I would push someObj into and processUsers should subscribe to the array but I'm not sure how to approach this.

BizzyBob · Accepted Answer · 2023-02-02 20:11:58Z

1

It sounds like you want to handle all your invocations of onValueChange in a single stream. In this case, you could use a Subject to push your objects through. This allows you to subscribe to a single observable to process your items:

private obj$ = new Subject();

onValueChange() {
  let someObj = ...
  this.obj$.next(someObj);
}

private processUsersSub = this.obj$.pipe(
  switchMap(obj => this.processUsers(obj)) // depending on desired behavior you could
).subscribe();                             // use mergeMap, exhaustMap, concatMap

private processUsers(someObj) {
  return from(Object.keys(someObj)).pipe(
    concatMap(key => this.getUsers(key)
  );
}

Depending on the behavior you want when a new emissions of obj$ occurs before the previous one is completed, you can choose the appropriate operator:

switchMap - switch to new source and discard any remaining processing
exhaustMap - discard any new emissions that occur while processing is in progress
concatMap - continue processing old items until complete, then process new items
mergeMap - process all items old and new, all at the same time (can specify concurrency limit)

answered Feb 2, 2023 at 20:11

BizzyBob

14.9k4 gold badges31 silver badges62 bronze badges

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

1 Comment

zer0 Over a year ago

Thank you! this works for me, but I was wondering why processUsersSub assignment is being done here.

Collectives™ on Stack Overflow

Create observables from a dynamic array

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related