Executing RxJS calls in a sequential manner while allowing inner calls to operate in

Question

Executing RxJS calls in a sequential manner while allowing inner calls to operate in

I want to achieve a scenario using rxjs where a group of http calls are made with the value returned from a previous call. The inner calls should run in parallel, while still being able to return the value from the first call as soon as it's available. Subscription to the result and error handling are also necessary. Importantly, errors from inner calls should not stop sibling calls.

The current code I have works fine but waits to return the initial value until all inner calls are complete due to using forkJoin:

public create(customer: Customer, groupIds: number[]): Observerable<Customer> {
    const response = this.http.post<Customer>('/customer', customer).pipe(mergeMap(
        (created) => {
            return forkJoin([
                of(created),
                forkJoin(groupIds.map(groupId => {
                    const membership = new Membership(created.Id, groupId);
                    return this.http.post<Membership>('/membership', membership);
                )
            ]);
        }
    )).pipe(map(a => a[0]));

    return response;
}

Is there a way to operate such that the created customer is returned without waiting for inner calls to finish? Could the above code be written more succinctly?

(NB: this.http refers to type HttpClient from Angular, which returns an Observable<T>)

typescript rxjs

Answer 1

Answer №1

If you are handling a local variable for the client, you can implement something along these lines:

public add(client: Client, projectIds: number[]): Observerable<Client> {
    const response = this.http.post<Client>('/client', client).pipe(
      tap(added => this.client = added), // <---
      mergeMap((added) => {
            return forkJoin([
                of(added),
                forkJoin(projectIds.map(projectId => {
                    const project = new Project(added.Id, projectId);
                    return this.http.post<Project>('/project', project);
                )
            ])
        }
    )).pipe(map(a => a[0]));

    return response;
}

Integrating a tap before the mergeMap assigns the client to a local variable. This allows access to the newly created client before other operations are finished.

I demonstrated this using stackblitz at: https://stackblitz.com/edit/angular-create-project-johndoe

(Please note that it uses different terminology but follows a similar concept.)

Alternatively

You might also consider the following approach:

  private taskAddedSubject = new BehaviorSubject<Task>(null);
  taskAddedAction$ = this.taskAddedSubject.asObservable();

  private projectSubject = new BehaviorSubject<number[]>([5, 10]);
  projectAction$ = this.projectSubject.asObservable();

  task$ = this.taskAddedAction$.pipe(
    mergeMap(task =>
      this.http.post<Task>(this.taskUrl, task, { headers: this.headers })
    )
  );

  projects$ = combineLatest([this.task$, this.projectAction$]).pipe(
    switchMap(([added, projectIds]) =>
      forkJoin(
        projectIds.map(projectId => {
          const project = {
            userId: added.userId,
            title: "Project:" + projectId,
            description: added.title
          } as Project;
          return this.http
            .post<Project>(this.projectUrl, project, { headers: this.headers })
            .pipe(tap(project => console.log(project)));
        })
      )
    )
  );

This setup defines two distinct streams. task$ mirrors your client and becomes instantly accessible after being posted.

projects$ is akin to your projects.

A respective stackblitz for this style can be found here: https://stackblitz.com/edit/angular-create-declarative-project-johndoe

Answer 2

If you are handling a local variable for the client, you can implement something along these lines:

public add(client: Client, projectIds: number[]): Observerable<Client> {
    const response = this.http.post<Client>('/client', client).pipe(
      tap(added => this.client = added), // <---
      mergeMap((added) => {
            return forkJoin([
                of(added),
                forkJoin(projectIds.map(projectId => {
                    const project = new Project(added.Id, projectId);
                    return this.http.post<Project>('/project', project);
                )
            ])
        }
    )).pipe(map(a => a[0]));

    return response;
}

Integrating a tap before the mergeMap assigns the client to a local variable. This allows access to the newly created client before other operations are finished.

I demonstrated this using stackblitz at: https://stackblitz.com/edit/angular-create-project-johndoe

(Please note that it uses different terminology but follows a similar concept.)

Alternatively

You might also consider the following approach:

  private taskAddedSubject = new BehaviorSubject<Task>(null);
  taskAddedAction$ = this.taskAddedSubject.asObservable();

  private projectSubject = new BehaviorSubject<number[]>([5, 10]);
  projectAction$ = this.projectSubject.asObservable();

  task$ = this.taskAddedAction$.pipe(
    mergeMap(task =>
      this.http.post<Task>(this.taskUrl, task, { headers: this.headers })
    )
  );

  projects$ = combineLatest([this.task$, this.projectAction$]).pipe(
    switchMap(([added, projectIds]) =>
      forkJoin(
        projectIds.map(projectId => {
          const project = {
            userId: added.userId,
            title: "Project:" + projectId,
            description: added.title
          } as Project;
          return this.http
            .post<Project>(this.projectUrl, project, { headers: this.headers })
            .pipe(tap(project => console.log(project)));
        })
      )
    )
  );

This setup defines two distinct streams. task$ mirrors your client and becomes instantly accessible after being posted.

projects$ is akin to your projects.

A respective stackblitz for this style can be found here: https://stackblitz.com/edit/angular-create-declarative-project-johndoe

Answer 3

Answer №2

If the parallel calls do not require the returned values:

public createCustomer(customer: Customer, groupIds: number[]): Observable<Customer> {
  const response = this.http.post <Customer>('/customer', customer)
    .pipe(
      tap(() => groupIds.map(groupId => {
          const membership = new Membership(created.Id, groupId);
          return this.http.post <Membership>('/membership', membership).subscribe();
        }))
      }
    ));

  return response;
}

If you need to use the data, consider using startWith with forkJoin:

public createCustomer(customer: Customer, groupIds: number[]): Observable<Customer> {
    const response = this.http.post<Customer>('/customer', customer).pipe(mergeMap(
        (created) => {
            return forkJoin(groupIds.map(groupId => {
                    const membership = new Membership(created.Id, groupId);
                    return this.http.post<Membership>('/membership', membership);
            ).pipe(startWith(created);
        }
    ));

    return response;
}

Answer 4

If the parallel calls do not require the returned values:

public createCustomer(customer: Customer, groupIds: number[]): Observable<Customer> {
  const response = this.http.post <Customer>('/customer', customer)
    .pipe(
      tap(() => groupIds.map(groupId => {
          const membership = new Membership(created.Id, groupId);
          return this.http.post <Membership>('/membership', membership).subscribe();
        }))
      }
    ));

  return response;
}

If you need to use the data, consider using startWith with forkJoin:

public createCustomer(customer: Customer, groupIds: number[]): Observable<Customer> {
    const response = this.http.post<Customer>('/customer', customer).pipe(mergeMap(
        (created) => {
            return forkJoin(groupIds.map(groupId => {
                    const membership = new Membership(created.Id, groupId);
                    return this.http.post<Membership>('/membership', membership);
            ).pipe(startWith(created);
        }
    ));

    return response;
}

Answer 5

Answer №3

It appears that you are looking to create an Observable that:

Immediately emits the customer upon fetching it, but also returns the value from the initial call as soon as possible.
Makes several parallel HTTP calls using the customer data to save memberships, emitting once all memberships have been saved.

If I understand correctly, here is how I would approach this:

First, a function can be created to handle the parallel calls using `forkJoin`:

function createMemberships(customerId: string, groupIds: number[]) {
  return forkJoin(groupIds.map(gId => 
    createMembershipHttpSimulation(customerId, gId))
  )
}

Then, another function can be created to return the desired Observable:

function create(customer: Customer, groupIds: number[]) {
  const createCustomerObs = createCustomerHttpSimulation(myCustomer).pipe(
      shareReplay(1)
  )
  const createMembershipsObs = createCustomerObs.pipe(
    concatMap(customer => createMemberships(customer.id, groupIds))
  )
  return concat(createCustomerObs, createMembershipsObs)
}

This function first creates an Observable for creating the Customer and ensures it is only called once using `shareReplay(1)`. Then, it sets up another Observable for creating memberships in parallel with the use of `concatMap` and `forkJoin`, followed by concatenating both Observables. This means the result will emit when the Customer is created and then again when all memberships are saved in parallel. Using `shareReplay` helps prevent redundant API calls.

You can view an implementation of this approach on StackBlitz, where delays demonstrate the sequential emission of Customer and Memberships.

Answer 6

It appears that you are looking to create an Observable that:

Immediately emits the customer upon fetching it, but also returns the value from the initial call as soon as possible.
Makes several parallel HTTP calls using the customer data to save memberships, emitting once all memberships have been saved.

If I understand correctly, here is how I would approach this:

First, a function can be created to handle the parallel calls using `forkJoin`:

function createMemberships(customerId: string, groupIds: number[]) {
  return forkJoin(groupIds.map(gId => 
    createMembershipHttpSimulation(customerId, gId))
  )
}

Then, another function can be created to return the desired Observable:

function create(customer: Customer, groupIds: number[]) {
  const createCustomerObs = createCustomerHttpSimulation(myCustomer).pipe(
      shareReplay(1)
  )
  const createMembershipsObs = createCustomerObs.pipe(
    concatMap(customer => createMemberships(customer.id, groupIds))
  )
  return concat(createCustomerObs, createMembershipsObs)
}

This function first creates an Observable for creating the Customer and ensures it is only called once using `shareReplay(1)`. Then, it sets up another Observable for creating memberships in parallel with the use of `concatMap` and `forkJoin`, followed by concatenating both Observables. This means the result will emit when the Customer is created and then again when all memberships are saved in parallel. Using `shareReplay` helps prevent redundant API calls.

You can view an implementation of this approach on StackBlitz, where delays demonstrate the sequential emission of Customer and Memberships.

Executing RxJS calls in a sequential manner while allowing inner calls to operate in

Answer №1

Answer №2

Answer №3

Similar questions

The error message "global.HermesInternal - Property 'HermesInternal' is not found in the type 'Global & typeof globalThis'" appeared

When users install my npm module, they are not seeing text prediction (intellisense) in their editors

Issue with unresolved module in ESLint

The automatic filtering feature does not kick in when the sorting is changed

Struggling with importing aliases in TypeScript for shadcn-ui library

Issue: The observer's callback function is not being triggered when utilizing the rxjs interval

What is the reason behind TypeScript treating numbers as strings when adding them together?

"Error encountered: Array is undefined when using the map and subscribe functions in Ionic

Ways to customize the OverridableComponent interface within Material-UI

Stop Mat-chip from automatically inserting a row upon selection

Angular - postpone function execution until Subject has completed its operation

Encountering an error message about 'resolving symbol values statically' while building an Angular 2 project

I'm having trouble getting my Node.js and TypeScript project to run because I keep encountering the error message ".ts is recognized as an unknown file extension."

Function modifies global variable

Difficulty displaying API information on a web browser with react.js

The Express API controller is unexpectedly receiving empty strings

Unpacking the information in React

Broaden the scope of the generic interface utilized within the package

Strange activities observed during the management of state in react hooks, where the splice() function ends up eliminating the

Is there a way to determine the tally of users who have achieved a score greater than that of [userID]?