安卓RxJava2应用程序从创建的可观察的调用emitter.onError()崩溃

Question

我在Observable的帮助下手动创建create()。现在，在内部，我检查一些条件&基于此，我想将错误通知订阅者。下面是我如何创建可观察的：

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()应该在我记录异常的onError of subscribe() of loginUser()中进行，但是应用程序会崩溃，而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 ->上检查结果&我还用

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

因为在我的例子中，如果处理发射器，忽略错误是很好的。

Answer 2

使用emitter.tryOnError对我有用，它也会调用subscribe()内部的错误，而不需要运行UndeliverableException或应用程序。

Answer 3

当你创造可观察的时候，你必须自己处理所有可能的错误。因此，您的代码应该如下所示：

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:在创建可观察的和一次性的时，始终确保错误检查。