java - 重构谷歌的 NetworkBoundResource 类以使用 RxJava 而不是 LiveData-6ren

java - 重构谷歌的 NetworkBoundResource 类以使用 RxJava 而不是 LiveData

转载作者：塔克拉玛干更新时间：2023-11-03 00:33:56

Google的android架构组件教程here有一部分解释了如何抽象化通过网络获取数据的逻辑。在其中，他们创建了一个名为 NetworkBoundResource 的抽象类，使用 LiveData 创建一个 react 流作为所有反应性网络请求的基础。

public abstract class NetworkBoundResource<ResultType, RequestType> {
private final AppExecutors appExecutors;

private final MediatorLiveData<Resource<ResultType>> result = new MediatorLiveData<>();

@MainThread
NetworkBoundResource(AppExecutors appExecutors) {
    this.appExecutors = appExecutors;
    result.setValue(Resource.loading(null));
    LiveData<ResultType> dbSource = loadFromDb();
    result.addSource(dbSource, data -> {
        result.removeSource(dbSource);
        if (shouldFetch()) {
            fetchFromNetwork(dbSource);
        } else {
            result.addSource(dbSource, newData -> result.setValue(Resource.success(newData)));
        }
    });
}

private void fetchFromNetwork(final LiveData<ResultType> dbSource) {
    LiveData<ApiResponse<RequestType>> apiResponse = createCall();
    // we re-attach dbSource as a new source, it will dispatch its latest value quickly
    result.addSource(dbSource, newData -> result.setValue(Resource.loading(newData)));
    result.addSource(apiResponse, response -> {
        result.removeSource(apiResponse);
        result.removeSource(dbSource);
        //noinspection ConstantConditions
        if (response.isSuccessful()) {
            appExecutors.diskIO().execute(() -> {
                saveCallResult(processResponse(response));
                appExecutors.mainThread().execute(() ->
                        // we specially request a new live data,
                        // otherwise we will get immediately last cached value,
                        // which may not be updated with latest results received from network.
                        result.addSource(loadFromDb(),
                                newData -> result.setValue(Resource.success(newData)))
                );
            });
        } else {
            onFetchFailed();
            result.addSource(dbSource,
                    newData -> result.setValue(Resource.error(response.errorMessage, newData)));
        }
    });
}

protected void onFetchFailed() {
}

public LiveData<Resource<ResultType>> asLiveData() {
    return result;
}

@WorkerThread
protected RequestType processResponse(ApiResponse<RequestType> response) {
    return response.body;
}

@WorkerThread
protected abstract void saveCallResult(@NonNull RequestType item);

@MainThread
protected abstract boolean shouldFetch();

@NonNull
@MainThread
protected abstract LiveData<ResultType> loadFromDb();

@NonNull
@MainThread
protected abstract LiveData<ApiResponse<RequestType>> createCall();
}

据我了解，这个类的逻辑是:

a) 创建一个名为“result”的MediatorLiveData作为主要返回对象，并设置其初始值为Resource.loading(null)

b) 从 Android Room db 获取数据作为 dbSource LiveData 并将其作为源 LiveData 添加到“结果”

c) 在 dbSource LiveData 的第一次发射中，从“结果”中删除 dbSource LiveData 并调用“shouldFetchFromNetwork()”，这将

如果为 TRUE，则调用“fetchDataFromNetwork(dbSource)”，它通过“createCall()”创建网络调用，返回封装为 ApiResponse 对象的响应的 LiveData
将 dbSource LiveData 添加回“result”并将发出的值设置为 Resource.loading(data)
将 apiResponce LiveData 添加到“result”，并在第一次发射时删除 dbSource 和 apiResponce LiveDatas
如果 apiResponse 成功，调用“saveCallResult(processResponse(response))”并将 dbSource LiveData 添加回“result”并将发出的值设置为 Resource.success(newData)
如果 apiResponse 失败，调用“onFetchFailed()”并将 dbSource LiveData 添加回“result”并将发出的值设置为 Resource.error(response.errorMessage, newData))
如果为 FALSE，只需将 dbSource LiveData 添加到“result”并将发出的值设置为 Resource.success(newData)

鉴于此逻辑是正确的解释，我尝试重构此类以使用 RxJava Observables 而不是 LiveData。这是我成功重构的尝试(我删除了初始的 Resource.loading(null)，因为我认为这是多余的)。

public abstract class NetworkBoundResource<ResultType, RequestType> {

private Observable<Resource<ResultType>> result;

@MainThread
NetworkBoundResource() {
    Observable<Resource<ResultType>> source;
    if (shouldFetch()) {
        source = createCall()
                .subscribeOn(Schedulers.io())
                .doOnNext(apiResponse -> saveCallResult(processResponse(apiResponse)))
                .flatMap(apiResponse -> loadFromDb().toObservable().map(Resource::success))
                .doOnError(t -> onFetchFailed())
                .onErrorResumeNext(t -> {
                    return loadFromDb()
                            .toObservable()
                            .map(data -> Resource.error(t.getMessage(), data))

                })
                .observeOn(AndroidSchedulers.mainThread());
    } else {
        source = loadFromDb()
                .toObservable()
                .map(Resource::success);
    }

    result = Observable.concat(
            loadFromDb()
                    .toObservable()
                    .map(Resource::loading)
                    .take(1),
            source
    );
}

public Observable<Resource<ResultType>> asObservable() {return result;}

protected void onFetchFailed() {}

@WorkerThread
protected RequestType processResponse(ApiResponse<RequestType> response) {return response.body;}

@WorkerThread
protected abstract void saveCallResult(@NonNull RequestType item);

@MainThread
protected abstract boolean shouldFetch();

@NonNull
@MainThread
protected abstract Flowable<ResultType> loadFromDb();

@NonNull
@MainThread
protected abstract Observable<ApiResponse<RequestType>> createCall();
}

由于我是 RxJava 的新手，我的问题是我是否正确重构为 RxJava 并保持与此类的 LiveData 版本相同的逻辑？

最佳答案

public abstract class ApiRepositorySource<RawResponse extends BaseResponse, ResultType> {

    // result is a Flowable because Room Database only returns Flowables
    // Retrofit response will also be folded into the stream as a Flowable
    private Flowable<ApiResource<ResultType>> result; 
    private AppDatabase appDatabase;

    @MainThread
    ApiRepositorySource(AppDatabase appDatabase) {
        this.appDatabase = appDatabase;
        Flowable<ApiResource<ResultType>> source;
        if (shouldFetch()) {
            source = createCall()
                .doOnNext(this::saveCallResult)
                .flatMap(apiResponse -> loadFromDb().toObservable().map(ApiResource::success))
                .doOnError(this::onFetchFailed)
                .onErrorResumeNext(t -> {
                    return loadFromDb()
                            .toObservable()
                            .map(data -> {
                                ApiResource apiResource;

                                if (t instanceof HttpException && ((HttpException) t).code() >= 400 && ((HttpException) t).code() < 500) {
                                    apiResource = ApiResource.invalid(t.getMessage(), data);
                                } else {
                                    apiResource = ApiResource.error(t.getMessage(), data);
                                }

                                return apiResource;
                            });
                })
                .toFlowable(BackpressureStrategy.LATEST);
        } else {
            source = loadFromDb()
                    .subscribeOn(Schedulers.io())
                    .map(ApiResource::success);
        }

        result = Flowable.concat(initLoadDb()
                            .map(ApiResource::loading)
                            .take(1),
                            source)
                .subscribeOn(Schedulers.io());
    }

    public Observable<ApiResource<ResultType>> asObservable() {
        return result.toObservable();
    }

    @SuppressWarnings("WeakerAccess")
    protected void onFetchFailed(Throwable t) {
        Timber.e(t);
    }

    @WorkerThread
    protected void saveCallResult(@NonNull RawResult resultType) {
        resultType.saveResponseToDb(appDatabase);
    }

    @MainThread
    protected abstract boolean shouldFetch();

    @NonNull
    @MainThread
    protected abstract Flowable<ResultType> loadFromDb();

    @NonNull
    @MainThread
    protected abstract Observable<RawResult> createCall();

    @NonNull
    @MainThread
    protected Flowable<ResultType> initLoadDb() {
        return loadFromDb();
    }
}

所以这是我在多次迭代后决定使用的。这目前正在生产中，并且适用于我的应用程序。以下是一些外带笔记:

创建一个BaseResponse接口(interface)
```
    public interface BaseResponse {
         void saveResponseToDb(AppDatabase appDatabase);
    }
```
并在您所有的 api 响应对象类中实现它。这样做意味着您不必在每个 ApiResource 中实现 save_to_database 逻辑，如果需要，您可以将其默认为响应的实现。
为简单起见，我选择在 onErrorResumeNext block 中处理 Retrofit 错误响应，但我建议您创建一个可以容纳所有这些逻辑的 Transformer 类。在这种情况下，我为 ApiResources 添加了一个额外的 Status 枚举值，称为 INVALID 用于 400 级响应。
您可能想使用 Reactive Streams 架构组件库来处理 LiveData
implementation "android.arch.lifecycle:reactivestreams:$lifecycle_version" 并在这个类中添加一个方法
```
    public LiveData<ApiResource<ResultType>> asLiveData {
         return LiveDataReactiveStreams.fromPublisher(result);
    }
```
理论上，这会完美地工作，因为我们的 ViewModel 不必将 Observable 发射转换为 LiveData 发射或为 View 中的 Observable 实现生命周期逻辑。不幸的是，这个流在每次配置更改时都会重建，因为它会在任何调用的 onDestroy 中处理 LiveData(无论 isFinishing 是真还是假)。因此，我们必须管理这个流的生命周期，这首先违背了使用它的目的，或者每次设备旋转时都会重复调用。

这是一个 UserRepository 创建 ApiNetworkResource 实例的例子

@Singleton
public class UserRepository {

    private final RetrofitApi retrofitApi;
    private final AppDatabase appDatabase;

    @Inject
    UserRepository(RetrofitApi retrofitApi, AppDatabase appDatabase) {
        this.retrofitApi = retrofitApi;
        this.appDatabase = appDatabase;
    }

    public Observable<ApiResource<User>> getUser(long userId) {
        return new ApiRepositorySource<UserResponse, User>(appDatabase) {

            @Override
            protected boolean shouldFetch() {
                return true;
            }

            @NonNull
            @Override
            protected Flowable<User> loadFromDb() {
                return appDatabase.userDao().getUserFlowable(userId);
            }

            @NonNull
            @Override
            protected Observable<UserResponse> createCall() {
                return retrofitApi.getUserById(userId);
            }
        }.asObservable();
    }

}

关于java - 重构谷歌的 NetworkBoundResource 类以使用 RxJava 而不是 LiveData，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/49845532/

文章推荐： r - 如何告诉Anaconda使用系统的RStudio？

文章推荐： linux - 在循环中捕获 bash 函数的返回值和输出

文章推荐： c++ - 使用不完整类型显式实例化函数模板

android - 如何在 Retrofit 中使用具有暂停乐趣的 NetworkBoundResource
我正在使用 Retrofit、LiveData、Room (Android AAC)。 NetworkBoundResource是googlesimple提供的网络资源和房间结合的好 helper .
java - NetworkBoundResource 类中 MediatorLiveData 的使用
在 Google NetworkBoundResource 类中，我无法理解 MediatorLiveData 的 addSource() 和 removeSource() 的顺序被使用。例如--在
android - 带 Kotlin 协程的 NetworkBoundResource
您对如何使用 NetworkBoundResource 实现存储库模式有任何想法吗？和 Kotlin 协程？我知道我们可以使用 GlobalScope 启动协程，但这可能会导致协程泄漏。我想将 vie
java - RxJava/Retrofit2/Java - NetworkBoundResource 未按预期工作
我遇到了一些有线问题。当我第一次订阅时，它会进行网络调用并将数据保存到数据库，但是 loadFromDb() 永远不会执行，并且不会引发任何错误。为什么会发生这种情况？ Flowable>>
android - 如何将 Android 分页库与 NetworkBoundResource 集成
我的应用使用 Android 的 Architecture components库并显示从具有无限滚动效果的分页 REST api 获取的项目列表。我想做的是使用 Paging Library结合
android - 没有 Room 的 NetworkBoundResource 帮助程序类
当我尝试为 Room Db 和 Retrofit 实现 NetworkBoundResource 和 Resource 帮助程序类时，它工作得很好。但是，我需要使用 Retrofit only wit
java - 重构谷歌的 NetworkBoundResource 类以使用 RxJava 而不是 LiveData
Google的android架构组件教程here有一部分解释了如何抽象化通过网络获取数据的逻辑。在其中，他们创建了一个名为 NetworkBoundResource 的抽象类，使用 LiveData

塔克拉玛干

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

java - 重构谷歌的 NetworkBoundResource 类以使用 RxJava 而不是 LiveData