串行执行动态创建的可观察数组(Execute dynamically created array of observables in series)

我正在开发一个项目（Angular2），我正在动态创建Observable并将它们放在一个数组中

var ObservableArray : Observable<any>[] = []; //filling up Observable array dynamically for (var i = 0; i < this.mainPerson.children.length; i++) { ObservableArray.push(Observable.fromPromise(this.determineFate(this.mainPerson.children[i]))); } } var finalObservable: Observable<any> = Observable.concat(ObservableArray); finalObservable .subscribe( data => { //here I expected to execute determineFate() for all observables inside array console.log("determine fate resolved data returned [" + data + "]"); }, error => { console.error("error on Age Year for Characters") },() => { //Here I expect this gets executed only when all Observables inside my array finishes console.log("determine fate resolved data returned COMPLETED"); //DB call }); determineFate(..): Promise<boolean> { ... return either true / false if success or error; }

我想在一系列中执行所有observable（ forkJoin似乎并行运行 - 所以使用concat）。一旦执行了所有可观察对象，就要执行一些与DB相关的代码。但似乎我在'Completed'块中的代码并没有等待所有Observables完成。我怎样才能做到这一点？

提前致谢

I am working on a project (Angular2) where I am creating Observables dynamically and putting them in an array

var ObservableArray : Observable<any>[] = []; //filling up Observable array dynamically for (var i = 0; i < this.mainPerson.children.length; i++) { ObservableArray.push(Observable.fromPromise(this.determineFate(this.mainPerson.children[i]))); } } var finalObservable: Observable<any> = Observable.concat(ObservableArray); finalObservable .subscribe( data => { //here I expected to execute determineFate() for all observables inside array console.log("determine fate resolved data returned [" + data + "]"); }, error => { console.error("error on Age Year for Characters") },() => { //Here I expect this gets executed only when all Observables inside my array finishes console.log("determine fate resolved data returned COMPLETED"); //DB call }); determineFate(..): Promise<boolean> { ... return either true / false if success or error; }

I want to execute all observables in a series (forkJoin seems to run in parallel - so used concat). Once all observables are executed, want to execute some DB related code. But it seems my code inside 'Completed' block does not wait for all Observables to finish. How can I achieve this?

Thanks in advance

最满意答案

使用Observable.concat(ObservableArray)将展平数组并逐个从ObservableArray发出每个Observable。顺便说一句，使用concat的静态版本只有两个或更多参数才有意义（参见http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#static-method-concat ）。

相反，您可以迭代Observables数组，并等待它们使用concatAll()运算符concatAll()完成。

此示例模拟您的用例：

var observableArray = []; // filling up Observable array dynamically for (var i = 0; i < 10; i++) { observableArray.push(Observable.of('Value ' + i)); } Observable.from(observableArray) .concatAll() .subscribe(console.log, null, () => console.log('completed'));

Observable.from()发出每个Observable， concatAll()按照它们发出的顺序订阅它们中的每一个。

此演示打印到控制台以下输出：

Value 0 Value 1 Value 2 Value 3 Value 4 Value 5 Value 6 Value 7 Value 8 Value 9 completed

Using Observable.concat(ObservableArray) will just flatten the array and emit each Observable from ObservableArray one by one. Btw, using the static version of concat makes sense only with two or more parameters (see http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#static-method-concat).

Instead you can iterate the array of Observables and wait until they complete one by one with the concatAll() operator.

This example simulates your use-case:

var observableArray = []; // filling up Observable array dynamically for (var i = 0; i < 10; i++) { observableArray.push(Observable.of('Value ' + i)); } Observable.from(observableArray) .concatAll() .subscribe(console.log, null, () => console.log('completed'));

The Observable.from() emits each Observable separately and concatAll() subscribes to each one of them in the order they were emitted.

This demo prints to console the following output:

Value 0 Value 1 Value 2 Value 3 Value 4 Value 5 Value 6 Value 7 Value 8 Value 9 completed

更多推荐