RxJS 取消嵌套回调

Question

我想做这样的事情：

Rx.Observable.of(userToken)
    .flatMap(verifyToken)
    .flatMap(getUserInformation)
    .flatMap(createUser)
    .flatMap(signNewToken)
    .subcribe({
        next: result => useResult(result),
        error: error => handleError(error)
    })

我要避免的是回调混乱。在我的代码中，verifyToken 之类的函数是 Observables，我想将它们链接起来。这种模式在正确的轨道上吗？因为现在只要其中一个在内部执行observer.error(new Error('problem'))，链就会崩溃并且我的错误处理程序不会被调用。我该如何改进？

Answer 1

我认为该错误确实会传播到您的订阅。这应该模拟您的情况：

const Rx = require('rxjs/Rx');
const Observable = Rx.Observable;

let userToken = Observable.of(42);
let verifyToken = Observable.range(1,3);
let getUserInformation = Observable.of('whatever');
let createUser = Observable.create(observer => {
    observer.error(new Error('problem'));
});

Observable.of(userToken)
    .flatMap(val => verifyToken)
    .flatMap(val => getUserInformation)
    .flatMap(val => createUser)
    .subscribe({
        next: result => console.log("Next:", result),
        error: error => console.log("Error subscriber:", error.message)
    });

控制台输出如预期：

Error subscriber: problem