Android RxJava2應用程序在創建observable時調用emitter.onError（）時崩潰

Question

我在create()的幫助下手動創建Observable。現在裏面，我檢查一些條件&基於此，我想通知用戶有關錯誤。以下是我正在創建觀察到：

public Observable<User> loginUser(String email, String password) { 
    return Observable.create(
     emitter -> { 
      myAsynchronousWork.onCompleteListener(
       result -> { 
        if(!result.isSuccess()) { 
         // This causes the crash. 
         emitter.onError(new Throwable(result.getError())); 
        } else { 
         // Process result & create User object & return it. This works as expected. 
         emitter.onNext(user); 
         emitter.onComplete(); 
        } 
       } 
      ); 

     } 
    ); 

} 

&然後我訂閱loginUser()像：

loginUser("", "") 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(user -> { 
        Log.d("TAG", "logged in user => " + user.getUuid()); 
        Log.d("TAG", "user name => " + user.getUserName()); 
        Log.d("TAG", "user email => " + user.getEmailId()); 
       }, throwable -> { 
        Log.e("TAG", "error in login => " + throwable.getMessage()); 
       },() -> { 

       }); 

我希望呼籲emitter.onError()應該往裏走的loginUser()subscribe()在那裏我已經登錄異常的onError，但相反，應用程序崩潰的異常返回emitter.onError() logcat像沒有人來處理它！

我通過調試&檢查發現雖然它在線emitter.onError(),emitter是"null"。但onNext & onComplete不會導致任何問題。 請讓我知道我在做什麼錯了？

Answer 1

之所以我的應用程序是越來越崩潰的是，我竟然有這樣的事情：

myAsynchronousWork.onCompleteListener(
       result -> { 
        if(!result.isSuccess()) { 
         // This causes the crash. 
         emitter.onError(new Throwable(result.getError())); 
        } else { 
         // Process result & create User object & return it. This works as expected. 
         emitter.onNext(user); 
         emitter.onComplete(); 
        } 
       }, 
       exception -> { 
        emitter.onError(exception); // This was the reason of problem! 
       } 
      ); 

我沒貼完整部分，因爲我沒有得到這導致異常。

這裏發生了什麼事情，我的代碼第一次進入exception->部分，該部分通知觀察者關於觀察者正在終止的錯誤&。現在我的result->部分正在執行&這裏，當我試圖再次調用emitter.onError()時，它崩潰了，因爲沒有emitter來處理這個& RxJava2正在全球範圍內拋出它。

所以對我來說，我刪除了exception ->部分，因爲它是任何如何將是對result ->在那裏我可以檢查結果&我還包裹emitter.onError與

if(!emitter.isDisposed()) emitter.onError(); 

，因爲在我的情況，如果發射器被丟棄，忽略錯誤是很好的。

Answer 2

當您創建可觀察性時，您必須自己處理所有可能的錯誤。所以你的代碼看起來應該象下面這樣：

public Observable<User> loginUser(String email, String password) { 
    return Observable.create(
     emitter -> { 
      try{ 
      myAsynchronousWork.onCompleteListener(
       result -> { 
       try{ 
        if(!result.isSuccess()) { 
         // This causes the crash. 
         emitter.onError(new Throwable(result.getError())); 
        } else { 
         // Process result & create User object & return it. This works as expected. 
         emitter.onNext(user); 
         emitter.onComplete(); 
        } 
       }catch(Exception e){ 
        emitter.onError(e); 
       } 
       } 
      ); 
     }catch(Exception e){ 
      emitter.onError(e); 
     } 

     } 
    ); 

} 

PS：確保創建觀察的和一次性的，當過錯誤檢查。