Retrofit2+RxJava2, неверный токен, как обновить поток при повторной подписке retryWhen()

Здесь есть несколько проблем:

• вам нужно передать токен доступа методу postSomethingWithAccessToken при повторной попытке, иначе вы просто повторите попытку с тем же старым недействительным токеном доступа.
• ваша повторная попытка, когда логика неверна, вы должны реагировать на ошибки Observable, которые вы получаете, и помещать туда свою логику повторной попытки. как вы сказали, этот метод выполняется первым, а не когда возникает ошибка, throwableObservable - это ответ на ошибку, он будет отражать ошибки как выбросы (onNext()), вы можете flatMap() каждую ошибку и ответ либо с ошибкой (для доставки ошибки источнику stream) complete или с помощью onNext() с каким-либо объектом, чтобы сигнализировать о повторной попытке.
  Отличный запрет сообщений в блоге Дэна Лью на эту тему.

Итак, вам нужно:
1) хранить токен доступа где-то, где вы можете изменить его с помощью обновления токена доступа.
2) исправить повторную попытку, когда логика правильно реагирует на ошибки.

Вот код предложения:

postSomethingWithAccessToken(request, accessToken)
        .subscribeOn(Schedulers.io())
        .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
                   @Override
                   public ObservableSource<?> apply(
                           @NonNull Observable<Throwable> throwableObservable) throws Exception {
                       return throwableObservable.flatMap(
                               new Function<Throwable, ObservableSource<? extends R>>() {
                                   @Override
                                   public ObservableSource<? extends R> apply(
                                           @NonNull Throwable throwable) throws Exception {
                                       if (throwable.code == 401) { //or 404/403, just a pseudo-code, put your real error comparing logic here
                                           return getAccessToken()
                                                           .doOnNext(refreshedToken -> accessToken.updateToken(refreshedToken));
                                                   //or keep accessToken on some field, the point to have mutable
                                                   //var that you can change and postSomethingWithAccessToken can see
                                       }
                                       return Observable.error(throwable);
                                   }
                               });
                       }
                   }
        )
        .subscribeOn(Schedulers.io())
        .subscribe(new Consumer<Result>() {
                       @Override
                       public void accept(@NonNull Result result) throws Exception {
                           //handle result
                       }
                   }
        );

БОЛЬШОЕ Спасибо yosriz за то, что он указал мне правильное направление для решения моей проблемы со скрежетом зубов, я должен использовать defer. Итак, я столкнулся с этой проблемой в GitHub, Зачем повторно подписывать исходный наблюдаемый вывод, когда я использую retryWhen оператор?

Это точно такая же проблема, с которой я сталкиваюсь сейчас, для тех, кто испытывает ту же проблему, вот мое решение.

Observable
    .defer(new Callable<ObservableSource<?>>() {
        @Override
        public ObservableSource<?> call() throws Exception {
            // return an observable source here, the observable that will be the source of the entire stream;
        }
    })
    .subscribeOn( /*target thread to run*/ )
    .retryWhen( {
        // return a throwable observable here that will perform the logic when an error occurred
    })
    .subscribe( /*subscription here*/ )

или вот полная нелямбда моего решения

Observable
    .defer(new Callable<ObservableSource<?>>() {
        @Override
        public ObservableSource<?> call() throws Exception {
            return mApiService.api().postSomethingWithAccessToken(
                request, preferences.getString("access_token", ""));
        }
    })
    .subscribeOn(Schedulers.io())
    .retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
        @Override
        public ObservableSource<?> apply(final Observable<Throwable> throwableObservable) throws Exception {
            return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                @Override
                public ObservableSource<?> apply(Throwable throwable) throws Exception {
                    if (throwable instanceof HttpException) {
                        HttpException httpException = (HttpException) throwable;
                        if (httpException.code() == 401) {
                            return mApiService.api().getAccessToken().doOnNext(new Consumer<Authentication>() {
                                    @Override
                                    public void accept(Authentication authentication) throws Exception {
                                        update(authentication);
                                    }
                                });
                        }
                    }
                    return Observable.error(throwable);
                }
            });
        }
    })
    .subscribe(new Observer<Void>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.e("subscribe", "TOKEN : " + preferences.getString("access_token", ""));
        }

        @Override
        public void onNext(Void value) {
            Log.e("onNext", "TOKEN : " + preferences.getString("access_token", ""));
        }

        @Override
        public void onError(Throwable e) {
            e.printStackTrace();
        }

        @Override
        public void onComplete() {
            Log.e("Complete", "____ COMPLETE");
        }
    });

Ключевым моментом здесь является «как изменить/обновить существующий наблюдаемый источник, когда оператор .retryWhen() повторно подписывается на наблюдаемый источник»