Tips for serializing the execution of an array of observables

Question

Tips for serializing the execution of an array of observables

I am currently working on a validation process that goes through data in a table row by row. Due to the fact that each row validation requires access to a shared resource, it is crucial that the access to this resource is serialized.

public validate():Observable<boolean>{
    const rowValidations:Observable<boolean>[] = dataRows.map(row=>this.validateSingleRow(row);
    return forkJoin(...rowValidations).pipe(
      map(results=>results.every(r=>r))
    )
}

My understanding is that forkJoin does not wait for each observable to finish before moving on to the next one, unlike concat, so this approach may fail. On the other hand, concat serializes all observables into a single stream.

Is there a way to achieve the subscription order of concat while still obtaining an array of results from each observable like what forkJoin provides, effectively synchronizing the execution of each inner observable (similar to Java's synchronized validateSingleRow)?

javascript typescript rxjs

Answer 1

Answer №1

If you are certain that each this.validateSingleRow(row) will emit only once, utilizing toArray() can simplify the process:

concat(...rowValidations).pipe(
  toArray(),
);

The use of concat ensures the correct order, while toArray() gathers all emissions into a single array and emits it after the source Observable completes.

However, in scenarios where validateSingleRow may emit multiple times and you only need its latest value, employing scan might be more appropriate:

const indexedRowValidations = rowValidations.map((o, index) => o.pipe(
  map(result => [index, result]),
));

concat(...indexedRowValidations ).pipe(
  scan((acc, [index, result]) => {
    acc[index] = result;
    return acc;
  }, {}),
  takeLast(1),
);

(I haven't tested this code, but I believe you grasp the concept :)).

Answer 2

If you are certain that each this.validateSingleRow(row) will emit only once, utilizing toArray() can simplify the process:

concat(...rowValidations).pipe(
  toArray(),
);

The use of concat ensures the correct order, while toArray() gathers all emissions into a single array and emits it after the source Observable completes.

However, in scenarios where validateSingleRow may emit multiple times and you only need its latest value, employing scan might be more appropriate:

const indexedRowValidations = rowValidations.map((o, index) => o.pipe(
  map(result => [index, result]),
));

concat(...indexedRowValidations ).pipe(
  scan((acc, [index, result]) => {
    acc[index] = result;
    return acc;
  }, {}),
  takeLast(1),
);

(I haven't tested this code, but I believe you grasp the concept :)).

Answer 3

Answer №2

Would this code snippet serve your needs?

class ValidationClass {
  dataRows = [1, 2, 3];

  public checkValidity(): Observable<boolean[]> {
    return this.processSequentially(this.dataRows);
  }

  private processSequentially<T>([current, ...remaining]: T[]): Observable<boolean[]> {
    return current
      ? this.validateSingleRow(current).pipe(
          switchMap((result) =>
            this.processSequentially(remaining).pipe(map((array) => [result, ...array]))
          )
        )
      : of([]);
  }

  // Mock function for validation
  private validateSingleRow(currentValue: any) {
    console.log(`Validating ${currentValue}...`);
    return of(Math.floor(Math.random() * 2) === 1).pipe(
      delay(1000),
      tap((result) => console.log(`Validation Result: ${result}`))
    );
  }
}

const instance = new ValidationClass();

instance.checkValidity().subscribe(console.log);

Check out the StackBlitz demo here

Answer 4

Would this code snippet serve your needs?

class ValidationClass {
  dataRows = [1, 2, 3];

  public checkValidity(): Observable<boolean[]> {
    return this.processSequentially(this.dataRows);
  }

  private processSequentially<T>([current, ...remaining]: T[]): Observable<boolean[]> {
    return current
      ? this.validateSingleRow(current).pipe(
          switchMap((result) =>
            this.processSequentially(remaining).pipe(map((array) => [result, ...array]))
          )
        )
      : of([]);
  }

  // Mock function for validation
  private validateSingleRow(currentValue: any) {
    console.log(`Validating ${currentValue}...`);
    return of(Math.floor(Math.random() * 2) === 1).pipe(
      delay(1000),
      tap((result) => console.log(`Validation Result: ${result}`))
    );
  }
}

const instance = new ValidationClass();

instance.checkValidity().subscribe(console.log);

Check out the StackBlitz demo here

Answer 5

Answer №3

The solution I came up with to meet my requirements may seem simpler than expected. By using the concat function along with toArray(), the code looks like this:

const rowDataValidation:Observable<boolean>[] = allRows.map(row=>defer(()=>this.validateSingleRow(row)));
return concat(...rowDataValidation).pipe(
  toArray(),
  map(results=>results.every(result=>result))
)

This allows each row to be validated one at a time and then transforms the boolean stream into an array of booleans using the toArray function.

Answer 6

The solution I came up with to meet my requirements may seem simpler than expected. By using the concat function along with toArray(), the code looks like this:

const rowDataValidation:Observable<boolean>[] = allRows.map(row=>defer(()=>this.validateSingleRow(row)));
return concat(...rowDataValidation).pipe(
  toArray(),
  map(results=>results.every(result=>result))
)

This allows each row to be validated one at a time and then transforms the boolean stream into an array of booleans using the toArray function.

Tips for serializing the execution of an array of observables

Answer №1

Answer №2

Answer №3

Similar questions

Preserving the background image on an html canvas along with the drawing on the canvas

Creating Stylish Popovers in Flask with Bootstrap 5 (Using Jinja Templating)

Is there a solution for the error message "Operator '+' cannot be used with types 'string | number' and 'string' | number'"?

The function Router.replace within Next Js is not recognized

Error message: Three JS is unable to locate the property 'add' due to its undefined status

The request to register at http://localhost:3000/api/auth/register returned a 404 error, indicating that the

Upgrading to Angular 18 Leads to Issue with Spread Operator: 'TypeError: res is not iterable' Error

Problems encountered while invoking a function using ng-click within an AngularJS controller?

Video background in webflow not currently randomizing

Position the div in the center of a container that is 100% width and has floating elements with dynamic widths

Commencing the invocation of actions from effects in NgRx

In ReactJs, you can use break tags to format a lengthy string and also apply color to certain words within the string

How to access an ejs variable within a Node.js environment using Javascript

What steps should I take to make the code in jsfiddle functional on my Visual Studio Code platform?

Adjust Scale to Less than 1 in JVectorMap

Can triggering an ngrx effect lead to memory leakage?

Searching for repeated values across disparate object fields?

Texture flagged for refresh, yet image remains undefined

What causes JavaScript parseFloat to add additional value in a for loop implementation?

Having trouble displaying dynamically generated texture from a fragment shader in ThreeJS