android - 如何在 rxjava 和改造中每小时重试请求 5 次

Question

我想弄清楚当我们从服务器收到错误 5 次尝试时如何每小时重新订阅相同的可观察对象。我知道 retryWhen 但真的无法理解如何在我的案例中使用它。我正在使用 retrofit 进行服务器调用和 rxjava 订阅。

这是我使用改造进行调用的方法。请帮助解决这个问题。

@Override
public Observable<Integer> uploadFileToServer(FileUploadData fileUploadData, File file) {
    //  log.i(TAG, "uploadFileToServer");
    FileUploadEndpoint fileUploadEndpoint = null;
    try {
        fileUploadEndpoint = retrofitServiceFactory.getService(FileUploadEndpoint.class);
    } catch (BaseUrlNotFoundException e) {
        e.printStackTrace();
        // log.i(TAG, "uploadFileToServer" + e.getMessage());
        return Observable.just(FileUploadConstants.EXCEPTION_FILE_UPLOAD);
    }

    // create RequestBody instance from file
    RequestBody requestFile =
            RequestBody.create(okhttp3.MultipartBody.FORM, file);

    // MultipartBody.Part is used to send also the actual file name
    MultipartBody.Part body =
            MultipartBody.Part.createFormData("uploadfile", file.getName(), requestFile);

    // add another part within the multipart request
    String descriptionString = "file upload";
    RequestBody description =
            RequestBody.create(
                    okhttp3.MultipartBody.FORM, descriptionString);

    Map<String, String> queryMap = new HashMap<>();
    queryMap.put("SENDER", fileUploadData.getSender());
    queryMap.put("SOURCE", fileUploadData.getSource());
    queryMap.put("SCHEMEID", fileUploadData.getSchemeId());
    queryMap.put("ISPROCESSINGREQ", "false");
    queryMap.put("ISENCRYPTED", "true");
    queryMap.put("UID", fileUploadData.getSchemeId());
    queryMap.put("METADATA", fileUploadData.getMetaData());


    final Observable<FileUploadResponse> requestObservable = fileUploadEndpoint.upload(queryMap, description, body);
    return requestObservable.map(new Function<FileUploadResponse, Integer>() {
        @Override
        public Integer apply(FileUploadResponse fileUploadResponse) throws Exception {
            if (fileUploadResponse != null) {
                int code = fileUploadResponse.getStatusCode();
                switch (code) {
                    case 100:
                        return FileUploadConstants.FILE_UPLOAD_SUCCESSFUL;
                }
            }
            return FileUploadConstants.EXCEPTION_FILE_UPLOAD;
        }
    }).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
        @Override
        public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
            return throwableObservable.zipWith(Observable.range(1, 5), new BiFunction<Throwable, Integer, FileUploadResponse>() {
                @Override
                public FileUploadResponse apply(Throwable throwable, Integer integer) throws Exception {
                    return null;//not able to write the logic :(
                }
            });
        }
    });
}

@Override
public void setBaseUrl(String baseUrl) {
    retrofitServiceFactory.setBaseUrl(baseUrl);
}

private interface FileUploadEndpoint {

    @Multipart
    @POST("da/appupload/file")
    Observable<FileUploadResponse> upload(@QueryMap Map<String, String> additionValues,
                                          @Part("description") RequestBody description,
                                          @Part MultipartBody.Part file);
}

Answer 1

这是给你的食谱。

public class RetryWithDelay implements
        Func1<Observable<? extends Throwable>, Observable<?>> {
    private static final String TAG = "RetryWithDelay";

    private static final int DEFAULT_RETRY_COUNT = 5;
    private static final int DEFAULT_RETRY_DELAY = 1000 * 60;

    private final int maxRetries;
    private final int retryDelayMillis;
    private int retryCount;

    public RetryWithDelay() {
        this.maxRetries = DEFAULT_RETRY_COUNT;
        this.retryDelayMillis = DEFAULT_RETRY_DELAY;
        this.retryCount = 0;
    }

    public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
        this.maxRetries = maxRetries;
        this.retryDelayMillis = retryDelayMillis;
        this.retryCount = 0;
    }

    @Override
    public Observable<?> call(Observable<? extends Throwable> attempts) {
        return attempts.flatMap(new Func1<Throwable, Observable<?>>() {
                    @Override
                    public Observable<?> call(Throwable throwable) {
                        if (throwable instanceof HttpException) {
                            LOGD(TAG, "Caught http exception.");
                        }

                        if (throwable instanceof IOException) {
                            LOGD(TAG, "Network error");
                        }

                        if (++retryCount < maxRetries) {
                            // When this Observable calls onNext, the original
                            // Observable will be retried (i.e. re-subscribed).
                            return Observable.timer(retryDelayMillis, TimeUnit.MILLISECONDS);
                        }

                        // Max retries hit. Just pass the error along.
                        return Observable.error(throwable);
                    }
                });
    }
}

然后在你的代码中这样使用它

// Leave constructor empty for default values
.retryWhen(new RetryWithDelay());

// Or setup different values
// In this case retry 3 times, with 5s delay
.retryWhen(new RetyryWithDelay(3, 5000));