Observable에 대한 Subscribe 콜백에 예외가 발생하면 어떻게해야합니까?

Question

나는 최신 Reactive Extensions을 사용하고 있는데 디자인 문제로 실행 한 : 내가 구독에 통과 대리자에서 예외가 발생하는 경우

무엇을 어떻게해야합니까? 소스 스테핑 통해

은, 내가 발견 :

• 주제는 예외를 무시합니다.
• Producer에서 파생 된 연산자 (예 : Where)는 예외가 예외를 통과 할 때 구독을 처리합니다.

당연히 나는 표준 RX 연산자를 통해 관찰 할 수있는 곳이면 모든 예외가 처분 때문에 이벤트가 바로 멈추는 것을 발견했습니다. 적어도 내가 다시 구독하지 않는 이상.

이것은 내 디자인에 의문을 제기하고 있습니다. 내 대의원으로부터 예외를 던지는 것은 나쁜 생각입니까? 확실히 RX 팀은 그렇게 생각합니다. (비록 "나쁜"구독을 조용하게 처분하는 것이 올바른 방법인지 의문을 제기하고있다.)

내 디자인을 보면, 왜 그것이 문제인지 알 수 없다. 몇 가지 보호 된 작업이 진행 중입니다. 일부 OnNext가 청취자에게 알림을 시작하기 시작합니다 (오래된 Skool .NET 이벤트를 Observables로 완전히 전환했습니다). 문제가 발생하면 핸들러에 충돌 할 때까지 스택을 던질 것입니다. . 필자의 경우 처리기는 처리중인 트랜잭션을 롤백하고 롤백을 청취자에게 알립니다. 그것은 모두 예외 안전, 잘 작동합니다. 적어도 Where 연산자의 생산자 기반에서 Dispose를 수행하지 않아도 올바르게 작동합니다.

조금 더 가면서 .. Subject와 peers가이 동작을하지 않는다는 것이 일치하지 않습니까? 그리고 우리가 여기에 쓴 적이있는 우리 자신의 IS 객체와 관측 할 수있는 연산자에 대해서도 우리는 예외 처리를 똑같이해야할까요?

나는 어떤 통찰력이라도 기대하고 있습니다!

Answer 1

문제가 발생하면 예외를 throw하십시오. 그들이 지적하는 문제를 다룰 수 있다고 생각할 때 잡으십시오. 예외 처리를위한 언어 지원에서는 함수 호출 스택이 있다고 가정하므로 예외는 스택을 따라가는 처리기를 찾습니다.

그러나 Rx를 사용할 때 처리 모델이 옆으로 돌았다는 것을 기억하십시오. 최상위에 소스 (호출 코드)가 있고 하단에 관찰자 (코드라고 함)가있는 깊은 함수 호출 스택이 없습니다. 따라서 Rx 스트림에서 옳은 일을하기 위해 언어에 의존 할 수는 없습니다.

콜백이 예외를 throw하는 경우 Rx는 예외를 캡처하여 관찰 가능 채널 OnError을 전달합니다. 구독 할 때 OnError 처리기를 제공하지 않으면 Rx가 백그라운드 스레드에서 예외를 발생시켜 응용 프로그램에서 처리되지 않은 예외를 생성하는 경향이 있습니다.

Rx는 이 아니며은 다운 스트림에 예외가 있음을 데이터 소스에 알립니다. 이는 데이터 소스가 데이터 소비자와 완전히 분리되어 있기 때문입니다. 동일한 프로세스 또는 동일한 시스템에 있거나 동일한 언어로 작성된 것은 아닙니다. 그렇기 때문에 Subject은 아무 것도하지 않습니다. 이 경우에는 Subject이 데이터 소스로 사용되며 실제로 관찰자가하는 일은 중요하지 않습니다.

앞서 언급했듯이 포착되지 않은 예외로 인해 Rx는 기본적으로 관측 대상에서 관찰자를 구독 취소하게됩니다. 이것은 빠른 속도의 철학이며 Rx가 만들 수있는 유일한 안전한 가정입니다. 관찰자가 예외를 발생시킨 경우 잘못된 것이 틀림 없으며 기본적으로 더 많은 데이터를 제공하면 안됩니다. Rx는 반드시 관찰 대상에서 탈퇴하지 않아도되는 방식으로 예외를 처리 할 수있는 메커니즘을 제공합니다.

한 가지 방법은 데이터로 예외를 설정하는 것입니다 :

// instead of: 
source.Where(foo => predicateThatMightThrowException(foo)).Subscribe(foo => ..., error => ...) 

// do: 
source.Select(foo => 
{ 
    try { return new { foo: foo, filter: predicateThatMightThrowException(foo), error: (Exception)null }; } 
    catch (Exception e) { return { foo: foo, filter: true, error: e } }; 
}) 
.Where(f => f.filter) 
.Subscribe(f => 
{ 
    if (f.error != null) { handle error } 
    else { handle f.foo } 
}); 

다른 방법 CatchException 또는 Retry를 사용하여 포함한다. Rxx 라이브러리에는 왼쪽/오른쪽 채널과 관찰 가능한 쌍이 있고 Either 유형이있어 "왼쪽"에서 좋은 데이터를 스트리밍하고 "오른쪽"에서 오류를 스트리밍하는 데 사용할 수 있습니다.

Answer 2

좋아요, 원본에서 내 대답을 발견했습니다. 후손을 위해 여기에 붙여 넣기 만하면됩니다.

// Safeguarding of the pipeline against rogue observers is required for proper 
// resource cleanup. Consider the following example: 
// 
// var xs = Observable.Interval(TimeSpan.FromSeconds(1)); 
// var ys = <some random sequence>; 
// var res = xs.CombineLatest(ys, (x, y) => x + y); 
// 
// The marble diagram of the query above looks as follows: 
// 
// xs -----0-----1-----2-----3-----4-----5-----6-----7-----8-----9---... 
//     |  |  |  |  |  |  |  |  | 
// ys --------4--+--5--+-----+--2--+--1--+-----+-----+--0--+-----+---... 
//    | | | |  | | | | |  |  | | |  | 
//    v v v v  v v v v v  v  v v v  v 
// res --------4--5--6--7-----8--5--6--5--6-----7-----8--7--8-----9---... 
//         | 
//        @#& 
// 
// Notice the free-threaded nature of Rx, where messages on the resulting sequence 
// are produced by either of the two input sequences to CombineLatest. 
// 
// Now assume an exception happens in the OnNext callback for the observer of res, 
// at the indicated point marked with @#& above. The callback runs in the context 
// of ys, so the exception will take down the scheduler thread of ys. This by 
// itself is a problem (that can be mitigated by a Catch operator on IScheduler), 
// but notice how the timer that produces xs is kept alive. 

따라서 대답은 예, OnNext에서 예외를 throw하는 것은 좋지 않습니다. 일반적으로. 내 구체적인 경우에는 그것이 괜찮다는 것을 알기 때문에 다른 길을 찾을 것입니다.