javascript - 仅在完成时才通过链接多个可观察对象来防止金字塔

Question

我使用 rxjs，我的问题是，如果我只想在 observable 完成时调用 next 函数，有什么办法吗？所以不是这个:

this.Start(data).subscribe(
    (data) => {
        console.log('next');
    },
    (err) => {
        console.log('err');
    },
    () => {
        console.log('complete');

        this.nextFunction()   // this return with other observable
        .subscribe(
            (data) => {
                console.log('next');
            },
            (err) => {
                console.log('err');
            },
            () => {
                console.log('complete');
                this.nextFunction2(); // this returns with other observable
                .subscribe(
                    (data) => {
                        console.log('next');
                    },
                    (err) => {
                        console.log('err');
                    },
                    () => {
                        console.log('complete');
                        this.nextFunction3()
                        .subscribe(...)
                    }
                );
            }
        );
    }
);

像这样:

this.Start(data).subscribe(
    (data) => {
        console.log('next');
    },
    (err) => {
        console.log('err');
    },
    () => {
        console.log('complete');
        return other observable;
    }
)
.then(
    (data) => {
        console.log('next');
    },
    (err) => {
        console.log('err');
    },
    () => {
        console.log('complete');
        return other observable;
    }
)
.then(
    (data) => {
        console.log('next');
    },
    (err) => {
        console.log('err');
    },
    () => {
        console.log('complete');
        return other observable;
    }
);

所以我不想像金字塔一样设计我的代码。

Answer 1

您可以使用静态 concat运算符组成一个可观察对象，一个接一个地订阅可观察对象:

import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/concat';
import 'rxjs/add/observable/defer';

Observable.concat(
  this.Start(data),
  Observable.defer(() => this.nextFunction()),
  Observable.defer(() => this.nextFunction2()),
  Observable.defer(() => this.nextFunction3()),
)
.subscribe();

请注意 defer运算符用于延迟获取后续可观察对象的调用。

加入了一些日志记录:

import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/concat';
import 'rxjs/add/observable/defer';
import 'rxjs/add/operator/do';

Observable.concat(
  this.Start(data).do(
    (value) => { console.log("next: Start:", value); },
    (error) => { console.log("error: Start:", error); },
    () => { console.log("completed: Start"); }
  ),
  Observable.defer(() => this.nextFunction().do(
    (value) => { console.log("next: nextFunction:", value); },
    (error) => { console.log("error: nextFunction:", error); },
    () => { console.log("completed: nextFunction"); }
  )),
  Observable.defer(() => this.nextFunction2().do(
    (value) => { console.log("next: nextFunction2:", value); },
    (error) => { console.log("error: nextFunction2:", error); },
    () => { console.log("completed: nextFunction2"); }
  )),
  Observable.defer(() => this.nextFunction3().do(
    (value) => { console.log("next: nextFunction3:", value); },
    (error) => { console.log("error: nextFunction3:", error); },
    () => { console.log("completed: nextFunction3"); }
  )),
)
.subscribe();