我有一個驗證程序，它逐行驗證表中的資料。因為每一行驗證都使用一個共享資源，所以對它的訪問必須被序列化。
public validate():Observable<boolean>{
const rowValidations:Observable<boolean>[] = dataRows.map(row=>this.validateSingleRow(row);
return forkJoin(...rowValidations).pipe(
map(results=>results.every(r=>r))
)
}
如果我理解正確，forkJoin在訂閱下一個之前不會等待每個 observable 完成，concat這樣可能會失敗。concat另一方面，將所有 observable 序列化為單個流。
我怎樣才能獲得像 with 一樣的訂閱訂單，concat但有一個每個 observable 的結果陣列，例如forkJoin有效地同步每個內部 observable 的執行（如 Javas synchronzied validateSingleRow）？
uj5u.com熱心網友回復：
實際上，如果您知道每個this.validateSingleRow(row)總是只發出一次，您可以使用toArray()：
concat(...rowValidations).pipe(
toArray(),
);
concat將保證正確的順序，toArray()并將所有的排放收集到一個單一的陣列中，并在源 Observable 完成后重新發送它。
否則， ifvalidateSingleRow可能會發出多次，而您總是只想要它的最后一個值，您可以使用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),
);
（我沒有測驗它，但我相信你明白了:)）。
uj5u.com熱心網友回復：
這樣的事情對你有用嗎？
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 演示
uj5u.com熱心網友回復：
滿足我的要求的解決方案比人們想象的要簡單。我已經使用concat具有toArray()這樣的
const rowValidations:Observable<boolean>[] = dataRows.map(row=>defer(()=>this.validateSingleRow(row));
return concat(...rowValidations).pipe(
toArray(),
map(results=>results.every(r=>r))
)
sovalidateSingleRow被一一執行toArray并將布爾流轉換為布爾陣列。
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/375620.html
標籤：javascript 打字稿 rxjs
下一篇：打字稿函式回傳型別