android - 使用 RxJava 的存储库网络感知

Question

我很难找到合适的解决方案来让 View / View 模型了解网络状态，特别是使用 RxJava。

我试着关注 Google's NetworkBoundResource和 Iammert's networkBoundResouce (也试过GithubBrowserSample但我真的不明白代码或它应该如何封装。

我正在寻找如何使用 RxJava2 实现 NetworkBoundResource，使数据通过数据库自动刷新。

我会展示我目前拥有的东西:

带有 arch.lifecycle.ViewModel 的 fragment

public class LoginViewModel extends ViewModel {

    static final int SCREEN_JUST_BUTTONS = 0;
    static final int SCREEN_LOGIN_INPUT = 1;
    static final int SCREEN_PASS_INPUT = 2;

    // using a PublishSubject because we are not interested in the last object that was emitted
    // before subscribing. Like this we avoid displaying the snackbar multiple times
    @NonNull
    private final PublishSubject<Integer> snackbarText;

    private int SCREEN = 0;

    private LoginUiModel loginUiModel;
    @NonNull
    private BaseSchedulerProvider schedulerProvider;
    @NonNull
    private UserRepository userRepository;

    @Inject
    LoginViewModel(@NonNull final BaseSchedulerProvider schedulerProvider, @NonNull final UserRepository userRepository) {
        this.schedulerProvider = schedulerProvider;
        this.userRepository = userRepository;
        snackbarText = PublishSubject.create();
        disposables = new CompositeDisposable();
        loginUiModel = new LoginUiModel();
    }

    int getScreen() {
        return SCREEN;
    }

    void setScreen(final int screen) {
        this.SCREEN = screen;
    }

    @NonNull
    Observable<Integer> getSnackbarText() {
        return snackbarText;
    }

    LoginUiModel getLoginUiModel() {
        return loginUiModel;
    }


    public Single<User> login(final String login, final String password) {
        return userRepository.login(login, password)
                             .subscribeOn(schedulerProvider.io())
                             .observeOn(schedulerProvider.ui())
                             .doOnError(this::handleErrorLogin);
    }

    private void handleErrorLogin(final Throwable t) {
        if (t instanceof HttpException) {
            showSnackbar(R.string.Login_invalid_input);
        } else if (t instanceof IOException) {
            showSnackbar(R.string.Common_no_connection);
        } else {
            showSnackbar(R.string.Common_error_loading);
        }
    }

    private void showSnackbar(@StringRes int textId) {
        snackbarText.onNext(textId);
    }

    void forgotPassword() {
        if (isValidEmail(loginUiModel.getEmail())) {
            showSnackbar(R.string.Login_password_sent);
        } else {
            showSnackbar(R.string.Login_invalid_login);
        }
        //TODO;
    }

    boolean isValidEmail(@Nullable final String text) {
        return text != null && text.trim().length() > 3 && text.contains("@");
    }
}

在我的 UserRepository 中:

@Singleton
public class UserRepository {

    @NonNull
    private final UserDataSource userRemoteDataSource;

    @NonNull
    private final UserDataSource userLocalDataSource;

    @NonNull
    private final RxSharedPreferences preferences;

    @Inject
    UserRepository(@NonNull RxSharedPreferences rxSharedPreferences, @NonNull @Remote UserDataSource userRemoteDataSource, @NonNull @Local UserDataSource
            userLocalDataSource) {
        this.userRemoteDataSource = userRemoteDataSource;
        this.userLocalDataSource = userLocalDataSource;
        preferences = rxSharedPreferences;
    }

    @NonNull
    public Single<User> login(final String email, final String password) {
        return userRemoteDataSource
                .login(email, password) // busca do webservice
                .flatMap(userLocalDataSource::saveUser) // salva no banco local
                .flatMap(user -> userLocalDataSource.login(user.getEmail(), user.getPassword())) // busca usuário no banco local como SSOT
                .doOnSuccess(this::saveUserId); // salva o ID do usuário nas preferências.
    }

    public Single<User> saveUser(String email, String password, String name, String phone, String company) {
        User user = new User(0, name, email, password, phone, company);

        return saveUser(user);
    }

    private Single<User> saveUser(final User user) {
        return userRemoteDataSource.saveUser(user)
                                   .flatMap(userLocalDataSource::saveUser)
                                   .doOnSuccess(this::saveUserId); // salva o ID do usuário nas preferências.
    }

    private void saveUserId(final User user) {
        preferences.getLong(Consts.KEY_USER_ID__long).set(user.getUserId());
    }

    public Flowable<User> getUserUsingSystemPreferences() {
        Preference<Long> userIdPref = preferences.getLong(Consts.KEY_USER_ID__long, -1L);
        if (userIdPref.get() <= 0) {
            //nao deveria acontecer, mas se acontecer, reseta:
            userIdPref.delete();
            return Flowable.error(new RuntimeException("Not logged in."));
        } else if (!userIdPref.isSet()) {
            return Flowable.error(new RuntimeException("Not logged in."));
        } else {
            return userLocalDataSource.getUser(String.valueOf(userIdPref.get()));
            }
        }
    }

和我的 UserLocalDataSource:

@Singleton
public class UserLocalDataSource implements UserDataSource {

    private final UserDao userDao;

    @Inject
    UserLocalDataSource(UserDao userDao) {
        this.userDao = get(userDao);
    }

    @NonNull
    @Override
    public Single<User> login(final String email, final String password) {
        return userDao.login(email, password);
    }

    @Override
    public Flowable<User> getUser(final String id) {
        return userDao.getUserById(Long.valueOf(id));
    }

    @Override
    public Single<User> saveUser(final User user) {
        return Single.create(e -> {
            long id = userDao.insert(user);
            user.setUserId(id);
            e.onSuccess(user);
        });
    }

    @Override
    public Completable deleteUser(final String id) {
        return getUser(id).flatMapCompletable(user -> Completable.create(e -> {
            int numberOfDeletedRows = userDao.delete(user);
            if (numberOfDeletedRows > 0) {
                e.onComplete();
            } else {
                e.onError(new Exception("Tried to delete a non existing user"));
            }
        }));
    }
}

和用户远程数据源:

public class UserRemoteDataSource implements UserDataSource {

    private final ISApiService apiService;

    @Inject
    UserRemoteDataSource(ISApiService apiService) {
        this.apiService = get(apiService);
    }

    @NonNull
    @Override
    public Single<User> login(final String email, final String password) {
        return apiService.login(email, password);
    }

    @Override
    public Flowable<User> getUser(final String id) {
        throw new RuntimeException("NO getUser endpoint");
    }

    @Override
    public Single<User> saveUser(final User user) {
        Map<String, String> params = new HashMap<>();

        params.put(ISApiConsts.EMAIL, user.getEmail());
        params.put(ISApiConsts.NAME, user.getName());
        params.put(ISApiConsts.PASSWORD, user.getPassword());
        params.put(ISApiConsts.PHONE, user.getPhone());
        params.put(ISApiConsts.COMPANY, user.getCompany());

        return apiService.signUp(params);
    }

    @Override
    public Completable deleteUser(final String id) {
        //can't delete user.
        return Completable.complete();
    }
}

UserDao 接口(interface)(它正在使用 Room)

@Dao
public interface UserDao extends BaseDao<User> {

    @Query("SELECT * FROM user WHERE user_id = :id")
    Flowable<User> getUserById(long id);

    @Query("SELECT * FROM user WHERE email = :login AND password = :password")
    Single<User> login(String login, String password);

}

我想确切地知道的是:如何封装我的域，同时仍然可以访问网络等可能出现的问题，并在 MVVM 模式中使用 RxJava 将其显示给用户？

Answer 1

首先让我复制过去，从 Github 浏览器示例中获取网络方法。

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 -> 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 -> setValue(Resource.success(newData)))
                );
            });
        } else {
            onFetchFailed();
            result.addSource(dbSource,
                    newData -> setValue(Resource.error(response.errorMessage, newData)));
        }
    });
}

我从这个 fragment 中了解到，你真的不监控网络状态，但应用程序只是尝试访问服务器，当成功响应时它以“成功”状态发布到 UI，如果它无法从服务器获取数据，它会触发 onFetchFailed() 方法，并将数据库数据设置为状态为“失败”的检索数据。

如果您决定显示缓存数据并显示获取新数据的失败消息，我认为您需要状态。

NetworkBoundResource 类中还有许多其他抽象方法，因此您可以在没有获得新数据时选择正确的操作。

说到 Rxjava，我正在使用 iammert 存储库的一个分支来使用 Rxjava，完成后我会在这里分享链接。