javascript - Hur man serialiserar körning av en array av observables

Jag har en valideringsprocess som validerar data i tabellen rad för rad. Eftersom varje radvalidering använder en delad resurs måste åtkomst till den serialiseras.

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

Om jag förstår rätt kommer forkJoin inte att vänta på att varje observable ska bli klar innan den prenumererar på nästa, som concat skulle göra, så det kommer förmodligen att misslyckas. concat å andra sidan serialiserar alla observables till en enda ström.

Hur kan jag få en prenumerationsordning som med concat men ha en array med resultat från varje observable som med forkJoin och på så sätt effektivt synkronisera exekveringen av varje inre observable (som Javas synchronzied validateSingleRow)?

Svar #1

Faktiskt, om du vet att varje this.validateSingleRow(row) alltid bara kommer att sända en gång kan du använda toArray():

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

concat kommer att garantera korrekt ordning och toArray() kommer att samla alla utsläpp i en enda array och skicka ut den igen efter att käll-Observable har slutförts.

Annars, om validateSingleRow kan avge värden flera gånger och du alltid bara vill ha dess sista värde kan du använda scan:

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),
);

(Jag har inte testat det men jag tror du förstår poängen :).)

Svar #2

Skulle något sådant fungera för dig?

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

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

  private validateSequentially<T>([cur, ...obs]: T[]): Observable<boolean[]> {
    return cur
      ? this.validateSingleRow(cur).pipe(
          switchMap((x) =>
            this.validateSequentially(obs).pipe(map((arr) => [x, ...arr]))
          )
        )
      : of([]);
  }

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

const obj = new SomeClass();

obj.validate().subscribe(console.log);

StackBlitz-demonstration