Tengo un proceso de validación que valida los datos en la tabla fila por fila. Debido a que cada validación de fila usa un recurso compartido, el acceso a él debe serializarse.
public validate():Observable<boolean>{ const rowValidations:Observable<boolean>[] = dataRows.map(row=>this.validateSingleRow(row); return forkJoin(...rowValidations).pipe( map(results=>results.every(r=>r)) ) } Si entiendo correctamente, forkJoin no esperará a que finalice cada observable antes de suscribirse al siguiente como lo haría concat , por lo que probablemente fallará. concat , por otro lado, serializa todos los observables en una sola secuencia.
¿Cómo puedo obtener una orden de suscripción como con concat pero tener una matriz de resultados de cada observable como con forkJoin sincronizando efectivamente la ejecución de cada observable interno (como la validación synchronzied validateSingleRow de Java)?
¿Te serviría algo como esto?
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);En realidad, si sabe que cada this.validateSingleRow(row) siempre se emitirá solo una vez, puede usar toArray() :
concat(...rowValidations).pipe( toArray(), ); concat garantizará el orden correcto y toArray() recopilará todas las emisiones en una sola matriz y la reemitirá después de que se complete la fuente Observable.
De lo contrario, si validateSingleRow puede emitir varias veces y siempre desea solo su último valor, puede usar 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), );(No lo probé, pero creo que entiendes la idea :)).
La solución que cumple con mi requisito es más simple de lo que uno podría pensar. He usado concat con toArray() como este
const rowValidations:Observable<boolean>[] = dataRows.map(row=>defer(()=>this.validateSingleRow(row)); return concat(...rowValidations).pipe( toArray(), map(results=>results.every(r=>r)) ) por lo tanto, validateSingleRow se ejecuta uno por uno y toArray transforma el flujo booleano en una matriz de booleanos.