如何使用 rxjava 处理多个数据源? [英] How to handle multiple data sources with rxjava?

问题描述

情况是这样的:

我有领域层为业务逻辑提供数据获取接口，我有2个数据源:本地数据库和远程网络.

I have the domain layer to provide data fetching interface for business logics, and I have 2 data sources: local database and remote network.

它是这样工作的:

1. 请求所有用户:DataRepository.getInstance().getUsers();
2. 在 DataRepository 中，有 2 个来源:
   • LocalDataSource.getUsers() 从本地数据库获取所有用户，如果没有数据则忽略此请求.
   • RemoteDataSource.getUsers() 从我们的服务器请求最新的用户列表(即使本地数据库中有数据，为了保持数据更新的目的)，当请求数据时然后保存或更新它到本地数据库并将结果发回.

1. Request all users: DataRepository.getInstance().getUsers();
2. In DataRepository, there are 2 sources:
   • LocalDataSource.getUsers() which fetches all users from local database, if there is no data then ignore this request.
   • RemoteDataSource.getUsers() which requests latest user list from our server(Even if there is data in local database, for the purpose of keeping data updated), when data requested then save or update it to local database and send back the results.

我知道我可以通过在 DataRepository 中执行此操作来实现我的目标:

I know that I may achieve my goal by doing this in DataRepository:

public Observable<List<User>> getUsers() {
    return Observable.create(new Observable.OnSubscribe<List<User>>() {
        @Override
        public void call(Subscriber<? super List<User>> subscriber) {
            // 1. Request users from local database 
            List<User> localUsers = mLocalDataSource.getUsers();
            if (!localUsers.isEmpty()) {
                subscriber.onNext(localUsers);
            }
            // 2. Request the latest user list from server
            // Send a retrofit2 request
            Call<List<User>> call = mRemoteDataSource.getUsers();
            try {
                List<User> networkUsers = call.execute().body();
                mLocalDataSource.saveUsers(networkUsers);
                subscriber.onNext(networkUsers);
                subscriber.onCompleted();
            } catch (IOException e) {
                subscriber.onError(e);
            }
        }
    });
}

现在想到我已经在项目中使用了rxjava，为什么不使用SqlBrite和Retrofit2 RxAdapters来做这个更方便呢?所以 LocalDataSource.getUsers() 现在返回 Observable>，RemoteDataSource.getUsers() 也是如此.

Now thinking that I already used rxjava in the project, why not using SqlBrite and Retrofit2 RxAdapters to do this for more convenience? So LocalDataSource.getUsers() now returns Observable<List<User>> and so does RemoteDataSource.getUsers().

LocalDataSource.java

LocalDataSource.java

public Observable<User> getUsers() {
    final String sqlQuery = String.format("SELECT * FROM %s", UserTable.TABLE_NAME);
    return mDatabaseHelper.createQuery(UserTable.TABLE_NAME, sqlQuery)
            .mapToList(new Func1<Cursor, User>() {
                @Override
                public User call(Cursor c) {
                    return UserTable.parseCursor(c);
                }
            });
}

RemoteDataSource.java

RemoteDataSource.java

public Observable<List<User>> getUsers() {
    return mRetrofitApi.users();
}

问题:

我应该在 DataRepository.getUsers() 中做什么来实现我用旧技巧所做的同样的事情?

What should I do in DataRepository.getUsers() to achieve the same thing I did with the old tricks?

public Observable<List<User>> getUsers() {
    Observable<List<User>> localUsers = mLocalDataSource.getUsers();
    Observable<List<User>> remoteUsers = mRemoteDataSource.getUsers()
            .flatMap(new Func1<List<User>, Observable<User>>() {
                @Override
                public Observable<User> call(List<User> users) {
                    return Observable.from(users);
                }
            })
            .doOnNext(new Action1<User>() {
                @Override
                public void call(User user) {
                    mLocalDataSource.saveUser(user);
                }
            })
            .toList();
    // What should I return to make two observables both able to emit results to the Subscriber
    return Observable.concat(localUsers, remoteUsers); // ???
}

推荐答案

我应该在 DataRepository.getUsers() 中做什么来实现我用旧技巧所做的同样的事情?

What should I do in DataRepository.getUsers() to achieve the same thing I did with the old tricks?

在这种情况下，您可以使用concat:

In such case, you can use concat:

public Observable<List<User>> getUsers() {
    return Observable.concat(localUsers.first(), remoteUsers);
}

但是如果不管是本地结果还是远程结果先出现，你可以使用merge:

But if it doesn't matter whether local or remote result comes first, you can use merge:

public Observable<List<User>> getUsers() {
    return Observable.merge(localUsers.first(), remoteUsers);
}

另外，如果你只想要一个结果(快一个赢)，你可以使用amb:

In addition, if you only just want one result (quicker one wins), you can use amb:

public Observable<List<User>> getUsers() {
    return Observable.amb(localUsers.first(), remoteUsers);
}

这篇关于如何使用 rxjava 处理多个数据源?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！