RxJava onErrorResumeNext scheduler

Question

У меня есть наблюдаемое, которое может завершиться с особым исключением, и в этом случае я хочу показать диалог с кнопкой повтора. Я видел ответ this, но он не совсем делает то, что я хочу. Я не смог использовать retryWhen, чтобы решить мою проблему, поэтому вместо этого я использовал onErrorResumeNext. Если вы можете придумать способ сделать то же самое с retryWhen, пожалуйста, сообщите.

Сейчас у меня есть этот кусок кода:

public Observable<Order> proceedWithOrdering(Activity activity) { 
    return apiService.createOrder() 
      .subscribeOn(Schedulers.io()) 
      .compose(applyRetryLogic(activity)) 
      .subscribeOn(AndroidSchedulers.mainThread()) 
      .observeOn(AndroidSchedulers.mainThread()); 
} 

public <T extends ApiResponse> Observable.Transformer<T, T> applyRetryLogic(Activity activity) { 
    return observable -> observable 
      .onErrorResumeNext(retry(observable, activity)) 
      .subscribeOn(AndroidSchedulers.mainThread()); 
} 

public <T> Func1<Throwable, ? extends Observable<? extends T>> retry(Observable toRetry, Activity activity) { 
    return throwable -> { 
     if (throwable instanceof NetworkException) { 
      MaterialDialog dialog = retryDialog(activity); 
      View retry = dialog.getActionButton(DialogAction.POSITIVE); 
      View cancel = dialog.getActionButton(DialogAction.NEGATIVE); 
      Observable<Object> retryClick = RxView.clicks(retry).map(o -> { 
       dialog.dismiss(); 
       return o; 
      }); 
      Observable<Object> cancelClick = RxView.clicks(cancel).flatMap(o -> { 
       dialog.dismiss(); 
       return Observable.error(throwable); 
      }); 

      dialog.show(); 

      return Observable.amb(retryClick, cancelClick) 
        .flatMap(o -> toRetry.compose(applyRetryLogic(activity))); 
     } else { 
      return Observable.error(throwable); 
     } 
    }; 
} 

Проблема заключается в том, что call внутри retry запускается на выполнение не в главном потоке, и это поднимает Can't create handler inside thread that has not called Looper.prepare() исключение.

Вопрос в том, как заставить его выполнять основную нить? Как вы можете видеть, я уже пробовал делать subscribeOn(AndroidSchedulers.mainThread()) сразу после того, как оба compose и onErrorResumeNext не повезло.

Я проверил свой код с помощью простых наблюдаемых, которые не работают с отдельными потоками, и он отлично работает.

Answer 1

Вы можете сделать это flatMap ping a PublishSubject, который затем обновляется после нажатия соответствующей кнопки. Вот классический пример Java Swing.

public class RetryWhenEnter { 
    public static void main(String[] args) { 
     AtomicInteger d = new AtomicInteger(); 
     Observable<Integer> source = Observable.just(1); 

     source.flatMap(v -> { 
      if (d.incrementAndGet() < 3) { 
       return Observable.error(new RuntimeException()); 
      } 
      return Observable.just(v); 
     }) 
     .retryWhen(err -> { 
      return err.flatMap(e -> { 
       System.out.println(Thread.currentThread() + " Error!"); 
       PublishSubject<Integer> choice = PublishSubject.create(); 
       SwingUtilities.invokeLater(() -> { 
        int c = JOptionPane.showConfirmDialog(null, 
         e.toString() + "\r\nRetry?", "Error", 
         JOptionPane.YES_NO_OPTION); 
        if (c == JOptionPane.YES_OPTION) { 
         choice.onNext(1); 
        } else { 
         choice.onCompleted(); 
        } 
       }); 
       return choice; 
      }); 
     }).subscribe(System.out::println, 
       Throwable::printStackTrace); 
    } 
} 

Edit:

Или использовать observeOn(AndroidSchedulers.mainThread()) непосредственно перед onErrorResumeNext или при использовании retryWhen: retryWhen(o -> o.observeOn(AndroidSchedulers.mainThread())...).

Редактировать 2 Я отбросил изменение, чтобы ответ был значимым снова.

Answer 2

Существует способ решения моей проблемы, используя retryWhen (спасибо @akarnokd):

public <T extends ApiResponse> Observable.Transformer<T, T> applyRetryLogic(Activity activity) { 
    return observable -> observable 
      .retryWhen(err -> err.flatMap(throwable -> { 
       L.d(Thread.currentThread() + " Error!"); 
       if (throwable instanceof NetworkException) { 
        PublishSubject<Integer> choice = PublishSubject.create(); 
        activity.runOnUiThread(() -> { 
         MaterialDialog dialog = retryDialog(activity); 
         View retry = dialog.getActionButton(DialogAction.POSITIVE); 
         View cancel = dialog.getActionButton(DialogAction.NEGATIVE); 
         RxView.clicks(retry).subscribe(o -> { 
          dialog.dismiss(); 
          choice.onNext(1); 
         }); 
         RxView.clicks(cancel).subscribe(o -> { 
          dialog.dismiss(); 
          choice.onError(throwable); 
         }); 

         dialog.show(); 
        }); 
        return choice; 
       } else { 
        return Observable.error(throwable); 
       } 
      })); 
}