typescript 如何在行为主题上抛出错误并继续流？

How do I throw an error on a behaviour subject and continue the stream?

提问by David

On one end, I have a stream which may occasionally throw an error:

一方面，我有一个可能偶尔会抛出错误的流：

this.behaviorSubject.error(error)

Later on, however, I want to continue the stream:

但是，稍后我想继续直播：

this.behaviorSubject.next(anotherValue)

on the other end, I have a subscriber subscribed to behaviorSubject.asObservable().

另一方面，我有一个订阅者订阅了behaviorSubject.asObservable().

In the subscribtion, I'm handling the value and the error:

在订阅中，我正在处理值和错误：

.subscribe( 
   ( value ) =>  {  /* ok */ },
   ( error ) =>  {  /* some error */ }
);

I want the effect to be the same as a simple onSuccessand onErrorcallback, where onErroris called every time an error occurs and doesn't prevent future onSuccesscalls from being made. How do I do this with RXJS?

我想要的效果是一样的简单onSuccess和onError回调，其中onError被称为每次出现错误并不会阻止未来的时间onSuccess呼叫被制成。我如何用 RXJS 做到这一点？

I've looked into catch but it seems to just prevent error from being called on subscribers.

我已经研究过 catch ，但它似乎只是防止在订阅者上调用错误。

采纳答案by olsn

Short answer:It's not possible.

简短的回答：这是不可能的。

How to work with this:The basic concept of RxJS is that any erroror complete-call will basically "kill"a stream. This concept forces you not "just to throw around errors here and there as you please"but to handle errors and the flow of data within your application properly. A BehaviorSubjectfor example is typically meant to hold data, however it should notbe used to also include the process of retrieving/creating that data and handle possible errors that might occur during the retrieval of the data.

如何处理：RxJS 的基本概念是 anyerror或-callcomplete基本上会“杀死”一个流。这个概念迫使您不要“只是随心所欲地到处抛出错误”，而是要正确处理应用程序中的错误和数据流。BehaviorSubject例如，一个example 通常用于保存数据，但是它不应该用于还包括检索/创建该数据的过程以及处理在检索数据期间可能发生的可能的错误。

So if you want to go by the book, you should split up your flow into two parts:

因此，如果您想按照本书进行操作，则应将流程分为两部分：

1. Retrieval/creation of the data: A stream, that will run once then then completes and/or throws an error whenever one occurs. When the data is retrieved it will be sent to the store.
2. The store(e.g. as in your case: a bunch of BehaviorSubjects): Only valid data arrives in the store, this means that no error-handling is done here and all parts relying on the store can trust in the store that it holds the correct data.

1. 数据的检索/创建：一个流，将运行一次，然后在发生时完成和/或抛出错误。检索到数据后，会将其发送到商店。
2. 商店（例如在你的情况下：一堆 BehaviorSubjects）：只有有效数据到达商店，这意味着这里没有进行错误处理，依赖商店的所有部分都可以信任它持有正确的商店数据。

---

---

As an example your data flow could look as follows (as a rough sketch):

例如，您的数据流可能如下所示（作为一个粗略的草图）：

store.ts

store.ts

dataStore: BehaviorSubject<IData> = new BehaviorSubject<IData>();
errorMessage: BehaviorSubject<IErrorMsg> = new BehaviorSubject<IErrorMsg>();

data-retrieval.ts

数据检索.ts

fetchDataById(id: string) {
    httpService.get(`some/rest/endpoint/${id}`)
        .subscribe(handleData, handleError);
}

handleData(data: IData) {
    errorMessage.next(null);
    dataStore.next(data);
}

handleError(error: Error) {
    errorMessage.next(error.message);
    dataStore.next(null);
}

---

---

"But this looks like a lot of overhead..."- True, however it ensures a clean and easy-to-understand flow of data within your application, that is easy to test and maintain. Also there are ready-to-use store-concepts like ngrxor reduxthat could be used.

“但这看起来像很多开销......”- 没错，但是它确保了应用程序中干净且易于理解的数据流，并且易于测试和维护。也有现成的存储概念，如ngrx或redux可以使用。

回答by Mark van Straten

Rx is fundamentally built upon the concept that an observable is either active or finalized (onComplete or onError). When an Observable is finalizing it will unSubscribe from its upstream Observable. No .catchcan fix that behaviour, it only gives you the option to map the error to something else.

Rx 从根本上建立在 observable 要么是活动的要么是最终的（onComplete 或 onError）的概念之上。当一个 Observable 完成时，它将取消订阅它的上游 Observable。没有.catch可以修复该行为，它只能让您选择将错误映射到其他内容。

Rx.Observable.interval(500)
  .mergeMap(i => i % 3 == 2 ? Rx.Observable.throw(new Error('kboom')) : Rx.Observable.of(i))
  .catch(err => Rx.Observable.of(err.message))
  .subscribe(
    val => console.log('val: ' + val),
    err => console.log('err: ' + err),
    () => console.log('stream completed')
  )

Note that this example completes after 3 emissions instead of 5

请注意，此示例在 3 次而不是 5 次排放后完成

When you invoke this.behaviorSubject.error(error)it wil internally finalize the Observable contained in your Subject. If you want to somehow emit errors then you need to make your errors non-error values:

当你调用this.behaviorSubject.error(error)它时，它会在内部完成你的主题中包含的 Observable。如果您想以某种方式发出错误，那么您需要将错误设为非错误值：

this.behaviorSubject.next({ value: 'somevalue' });
this.behaviorSubject.next({ error: error });
this.behaviorSubject.next({ value: 'somevalue' });

Then you are able to distinguish based on the properties on your emitted value what action you should take.

然后，您可以根据发出的值的属性来区分您应该采取什么行动。

回答by KickerKeeper

This may not work for your situation, but I ran into this same issue when work with Angular 2 because we would navigate across screens and would want the service to retry an API and not just call the error function again. It would actually cause bigger issues because the function was called in our constructor and the error function would try to update the UI which was not yet ready.

这可能不适用于您的情况，但我在使用 Angular 2 时遇到了同样的问题，因为我们会跨屏幕导航并希望服务重试 API 而不仅仅是再次调用错误函数。它实际上会导致更大的问题，因为在我们的构造函数中调用了该函数，并且错误函数会尝试更新尚未准备好的 UI。

What I did seems to work fine and be pretty clean. I created a reset the Subject in the error handler.

我所做的似乎工作正常并且非常干净。我在错误处理程序中创建了一个重置​​主题。

subject.subscribe( 
   ( value ) =>  {  /* ok */ },
   ( error ) =>  {  
      //handle error
      //reset subject
      this.subject = new Subject<any>();
   }
);

This works in our case because every time you navigate to the screen new subscriptions are getting torn down from the old screen then set up in the new, so the new subject won't hurt anything.

这在我们的情况下有效，因为每次导航到屏幕时，新订阅都会从旧屏幕上拆除，然后在新屏幕中设置，因此新主题不会伤害任何东西。