Cách xử lý ngoại lệ từ các phương thức không đồng bộ trong câu lệnh SelectMany

Question

Tôi đang cố gắng xử lý một số tác vụ không đồng bộ bằng Rx, ví dụ:

var list = Enumerable.Range(0, 100) 
    .ToObservable() 
    .SelectMany(x => Observable.Start(() => { 
     Console.WriteLine("Processing {0} ...", x); 

     Thread.Sleep(100 * x % 3); 

     if (x > 90) { 
      Console.WriteLine("Procesing exception {0} > 90", x); 
      throw new Exception("Value too large"); 
     } 
     Console.WriteLine("Processing {0} completed.", x); 
     return x; 
    })) 
    .Subscribe(
     x => { Console.WriteLine("Next [{0}]", x); }, 
     e => { 
      Console.WriteLine("Exception:"); 
      Console.WriteLine(e.Message); 
     }, 
     () => { Console.WriteLine("Complete"); } 
    ); 

Vấn đề tôi có với mã này là ngoại lệ không được chuyển cho người đăng ký. Vì vậy, sau khi rất nhiều cố gắng tôi đã từ bỏ và quyết định đặt câu hỏi đơn giản này:

Làm thế nào để bạn xử lý các trường hợp ngoại lệ được nêu ra từ bên trong các phương thức không đồng bộ trong tuyên bố SelectMany?

Chỉ cần làm rõ, thực hiện cuối cùng là cuộc gọi hàm đồng bộ có thể hoặc không thể ném ngoại lệ. Mục tiêu là chuyển nó cho người đăng ký để nó có thể được xử lý thêm (trong trường hợp cụ thể, một thông báo sẽ được hiển thị cho người dùng).

Sửa

tôi chuyển những phát hiện của tôi xuống đến một câu trả lời, vì vậy mà tôi có thể đánh dấu câu hỏi này như đã trả lời. Cá nhân, tôi không đồng ý với tự trả lời ... nhưng đôi khi không có cách nào khác, vì vậy xin lỗi vì nó.

Answer 1

Câu trả lời

Trên thực tế mã là hoạt động đúng. Tuy nhiên, trình gỡ lỗi phá vỡ các ngoại lệ khi các hoạt động không đồng bộ vẫn được thực thi trong nền - ít nhất là các trình gỡ lỗi đã bắt đầu khi ngoại lệ đầu tiên xảy ra. Ném tôi đi! Nếu bạn chạy mã mà không có trình gỡ lỗi thì các ngoại lệ sẽ bị nuốt. Do đó, tôi đoán vấn đề thực sự ở phía trước máy tính :-)

Vẫn còn một số giải thích rõ ràng - một điều này đúng - mà việc triển khai phải có thực sự một số xử lý lỗi được triển khai ... xem Bối cảnh.

nền

Observable.Start là một phương pháp thuận tiện sử dụng phương thức Observable.ToAsync để biến một hàm/acion thành một hoạt động async. Nếu bạn nhìn vào việc thực hiện các phương pháp bạn sẽ thấy rằng nó đã làm việc xử lý/chuyển tiếp ngoại lệ.

public static Func<IObservable<TResult>> ToAsync<TResult>(this Func<TResult> function, IScheduler scheduler) { 
    if (function != null) { 
     if (scheduler != null) { 
      return() => { 
       AsyncSubject<TResult> asyncSubject = new AsyncSubject<TResult>(); 
       scheduler.Schedule(() => { 
        TResult result = default(TResult); 
        try { 
         result = function(); 
        } catch (Exception exception1) { 
         Exception exception = exception1; 
         asyncSubject.OnError(exception); 
         return; 
        } 
        asyncSubject.OnNext(result); 
        asyncSubject.OnCompleted(); 
       }); 
       return asyncSubject.AsObservable<TResult>(); 
      }; 
     } else { 
      throw new ArgumentNullException("scheduler"); 
     } 
    } else { 
     throw new ArgumentNullException("function"); 
    } 
} 

Answer 2

Sử dụng Materialize để chuyển đổi tin nhắn OnError/OnCompleted thành thông báo.

Ví dụ,

observable.SelectMany(x => Observable.Start(fn).Materialize())

sẽ giúp bạn có được lỗi/hoàn bọc trong một thông báo được xử lý tại điểm thuê bao đường thực tế của bạn xuống dòng, như trái ngược với các lỗi bị chấm dứt trong SelectMany .

Điều này rất hữu ích đối với hầu hết các hoạt động cuộc gọi Async vì phương pháp không thành công hoặc hoàn tất.