Как правильно обрабатывать onError внутри RxJava (Android)?

Я получаю список установленных приложений на устройстве. Это дорогостоящая операция, поэтому я использую Rx для этого:

    Observable<List> observable = Observable.create(subscriber -> {
        List result = getUserApps();

        subscriber.onNext(result);
        subscriber.onError(new Throwable());
        subscriber.onCompleted();
    });

    observable
            .map(s -> {
                ArrayList<String> list = new ArrayList<>();
                ArrayList<Application> applist = new ArrayList<>();
                for (Application p : (ArrayList<Application>) s) {
                    list.add(p.getAppName());
                    applist.add(p);
                }
                return applist;
            })
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> L.e(TAG, "Throwable " + throwable.getMessage()))
            .subscribe(s -> createListView(s, view));

Однако моя проблема связана с ошибками обработки. Обычно пользователь запускает этот экран, ждет загрузки приложений, выбирает лучшее и переходит на следующую страницу. Однако, когда пользователь быстро меняет приложение UI - приложение с NullPointer.

Хорошо, поэтому я реализовал это onError. Однако он все еще не работает, и с помощью выше этого он меня бросает:

    04-15 18:12:42.530  22388-22388/pl.digitalvirgo.safemob E/AndroidRuntime﹕ FATAL EXCEPTION: main
        java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.
                at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:52)
                at android.os.Handler.handleCallback(Handler.java:730)
                at android.os.Handler.dispatchMessage(Handler.java:92)
                at android.os.Looper.loop(Looper.java:176)
                at android.app.ActivityThread.main(ActivityThread.java:5419)
                at java.lang.reflect.Method.invokeNative(Native Method)
                at java.lang.reflect.Method.invoke(Method.java:525)
                at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1046)
                at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:862)
                at dalvik.system.NativeStart.main(Native Method)
         Caused by: rx.exceptions.OnErrorNotImplementedException
                at rx.Observable$31.onError(Observable.java:7134)
                at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:154)
                at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:111)
                at rx.internal.operators.OperatorDoOnEach$1.onError(OperatorDoOnEach.java:70)
                at rx.internal.operators.NotificationLite.accept(NotificationLite.java:147)
                at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.pollQueue(OperatorObserveOn.java:177)
                at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.access$000(OperatorObserveOn.java:65)
                at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber$2.call(OperatorObserveOn.java:153)
                at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:47)
                at android.os.Handler.handleCallback(Handler.java:730)
                at android.os.Handler.dispatchMessage(Handler.java:92)
                at android.os.Looper.loop(Looper.java:176)
                at android.app.ActivityThread.main(ActivityThread.java:5419)
                at java.lang.reflect.Method.invokeNative(Native Method)
                at java.lang.reflect.Method.invoke(Method.java:525)
                at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1046)
                at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:862)
                at dalvik.system.NativeStart.main(Native Method)
         Caused by: java.lang.Throwable
                at pl.digitalvirgo.safemob.fragments.wizard.ApplicationsFragment.lambda$getAppList$25(ApplicationsFragment.java:267)
                at pl.digitalvirgo.safemob.fragments.wizard.ApplicationsFragment.access$lambda$2(ApplicationsFragment.java)
                at pl.digitalvirgo.safemob.fragments.wizard.ApplicationsFragment$$Lambda$3.call(Unknown Source)
                at rx.Observable$1.call(Observable.java:145)
                at rx.Observable$1.call(Observable.java:137)
                at rx.Observable.unsafeSubscribe(Observable.java:7304)
                at rx.internal.operators.OperatorSubscribeOn$1$1.call(OperatorSubscribeOn.java:62)
                at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:47)
                at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:390)
                at java.util.concurrent.FutureTask.run(FutureTask.java:234)
                at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:153)
                at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:267)
                at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1080)
                at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:573)
                at java.lang.Thread.run(Thread.java:841)

Как правильно справиться с этой проблемой?

Ответ 1

Мой прием: вы, вероятно, используете Action1 в

.subscribe(s -> createListView(s, view));

Вам нужно будет заменить его Абонентом или Наблюдателем, который имеет абстрактный метод onError. Этот метод будет вызываться из subscriber.onError(new Throwable());

EDIT: Вот как я это сделаю. При ближайшем рассмотрении я думаю, что основная проблема в вашем коде - это ранняя часть, на которую вы вызываете subscriber.onError, даже если нет ошибки. Вам, вероятно, не нужно map либо потому, что вы технически передаете данные как есть без манипуляций. Но я оставил его на случай, если это понадобится позже.

     Observable.create(new Observable.OnSubscribe<Application>() {
        @Override
        public void call(Subscriber<? super Application> subscriber) {
            List result = getUserApps();
            if (result != null){
                for (Application app : result){
                     subscriber.onNext(app);
                }
                subscriber.onComplete();
            }else{
                subscriber.onError(new IOException("no permission / no internet / etc"));
               //or if this is a try catch event you can pass the exception
            }     
        }
     })
    .subscribeOn(Schedulers.io())//the thread *observer* runs in
    .observeOn(AndroidSchedulers.mainThread())//the thread *subscriber* runs in
    .map(new Func1<Application, String>() {

        // Mapping methods are where data are manipulated. 
        // You can simply skip this and 
        //do the same thing in Subscriber implementation
        @Override
        public String call(Application application) {
            return application.getName();
        }
    }).subscribe(new Subscriber<String>() {
        @Override
        public void onCompleted() {
           Toast.makeText(context, "completed", Toast.LENGTH_SHORT).show();
           //because subscriber runs in main UI thread it ok to do UI stuff
           //raise Toast, play sound, etc
        }

        @Override
        public void onError(Throwable e) {
           Log.e("getAppsError", e.getMessage());
           //raise Toast, play sound, etc
        }

        @Override
        public void onNext(String s) {
            listAdapter.add(s);
        }
    });

Ответ 2

.doOnError() является оператором и не является такой частью Subscriber.

Следовательно, наличие .doOnError() не считается реализованным onError().

О вопросе в одном из комментариев, конечно, можно использовать lambdas.

В этом случае просто замените

.doOnError(throwable -> L.e(TAG, "Throwable " + throwable.getMessage()))
.subscribe(s -> createListView(s, view))

с

.subscribe(s -> createListView(s, view),
    throwable -> L.e(TAG, "Throwable " + throwable.getMessage()))

Ответ 3

Вот ответ новичка (потому что я новичок в javarx и, наконец, исправил эту проблему):

Вот ваша реализация:

    Observable.create(new Observable.OnSubscribe<RegionItem>() {
                @Override
                public void call(Subscriber<? super RegionItem> subscriber) {
                    subscriber.onError(new Exception("TADA !"));
                }
            })
            .doOnNext(actionNext)
            .doOnError(actionError)
            .doOnCompleted(actionCompleted)
            .subscribe();

В этой предыдущей реализации, когда я подписываюсь, я запускаю поток ошибок... и я получаю сбой приложения.

Проблема заключается в том, что вам нужно управлять ошибкой из вызова subscribe(). "DoOnError (...)" - это всего лишь помощник, который клонирует ошибку и дает вам новое место для выполнения некоторых действий после ошибки. Но он не обрабатывает ошибку.

Итак, вам нужно сменить код следующим образом:

    Observable.create(new Observable.OnSubscribe<RegionItem>() {
                @Override
                public void call(Subscriber<? super RegionItem> subscriber) {
                    subscriber.onError(new Exception("TADA !"));
                }
            })
            .subscribe(actionNext, actionError, actionCompleted);

Не уверен в настоящем объяснении, но я это исправить. Надеюсь, это поможет.