RxJS - 在 onNext 发生错误后继续发送通知

Question

我是 Rx 的新手，我真的很感激在错误处理方面的一些帮助。我有以下代码，它基本上是预先输入的:

var observable = Rx.Observable.fromEvent(searchField, 'keyup')
    .map(ev => ev.target.value)
    .filter(text => text.length > 2)
    .debounce(500 /* ms */)
    .distinctUntilChanged()
    .flatMapLatest(getAsyncSearchResults);

observable.subscribe(function(value) {
        console.log("onNext");
        throw "error"; // THIS TERMINATES THE OBSERVABLE
    },
    function(error) {
        console.log("Error");
    },
    function() {
        console.log("Completed");
    });

Observable 在 onNext 函数发生异常后终止。这对我来说似乎是一种不需要的行为，因为我不想将所有 onNext 代码包装在 try-catch 中。有什么方法可以告诉 observable 无论发生什么异常都继续发出通知？

Answer 1

您需要切换到一个新的一次性流，如果其中发生错误，它将被安全处理，并保持原始流处于事件状态。

我已将您的代码修改为更简单的 0 - 5 流以轻松演示。如果数字为 3，则会引发错误/异常。

Rx.Observable.from([0,1,2,3,4,5])
    .switchMap(value => {

        // This is the disposable stream!
        // Errors can safely occur in here without killing the original stream

        return Rx.Observable.of(value)
            .map(value => {
                if (value === 3) {
                    throw new Error('Value cannot be 3');
                }
                return value;
            })
            .catch(error => {
                // You can do some fancy stuff here with errors if you like
                // Below we are just returning the error object to the outer stream
                return Rx.Observable.of(error);
            });

    })
    .map(value => {
        if (value instanceof Error) {
            // Maybe do some error handling here
            return `Error: ${value.message}`;
        }
        return value;
    })
    .subscribe(
      (x => console.log('Success', x)),
      (x => console.log('Error', x)),
      (() => console.log('Complete'))
    );

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.1/Rx.min.js"></script>

这篇博文中有关此技术的更多信息: The Quest for Meatballs: Continue RxJS Streams When Errors Occur