使用 RxJava 和 Retrofit 实施 Room

Question

我正在尝试将 Room 与 RxJava 和 Retrofit 结合使用，在您建议使用组件拱门之前（在这种情况下不可能，该项目已完成 50%，只需继续清理拱门）。

所以问题是这样的。我有一个返回POJO 的网络服务。像这样的东西：

{
 "success":"true",
 "message":"message",
 "data":{[
   "id":"id",
   "name":"name",
   "lname":"lname",
 ]} 
}

POJO 更复杂，但对于这个例子来说这是可以的。我需要这样做，因为我的视图进行查询以从房间调用数据，但如果我的数据库中没有数据，则调用我的 Web 服务，我的 Web 服务的响应将转换为实体并保存在我的数据库（房间）中并在返回后我的观点的数据列表。

我正在使用干净的拱门。我很感激任何帮助。再次不尝试使用

数据布局

• 数据库
• 网络
• 存储库

领域

• 互动者
• 回调

推介会

• 主持人
• 看法

POJO API 响应

{
 "success":"true",
 "message":"message",
 "data":{[
   "id":"id",
   "name":"name",
   "address":"address",
   "phone":"phone",
 ]} 
}

我的数据库实体

@Entity(tableName = "clients")
    public class clients {

    String id;
    String name;
    String address;
    String phone;
    String status;


    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public String getPhone() {
        return phone;
    }

    public void setPhone(String phone) {
        this.phone = phone;
    }

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }
}

我的房间之道

@Dao
public interface ClientsDao {

     @Insert(onConflict = OnConflictStrategy.REPLACE)
     void saveAll(List<Clients> clients);

     @Query("SELECT * FROM Clients")
     Flowable<List<Clients>> listClients();

}

RxJava帮助类

public class RxHelper {
private static final String TAG = RxHelper.class.getName();

@NonNull
public static <T>Observable<T> getObserbable(@NonNull final Call<T> reponse){

    return Observable.create(new ObservableOnSubscribe<T>() {
        @Override
        public void subscribe(final ObservableEmitter<T> emitter) throws Exception {

            reponse.enqueue(new Callback<T>() {
                @Override
                public void onResponse(Call<T> call, Response<T> response) {

                    if (!emitter.isDisposed()) {
                        emitter.onNext(response.body());
                    }
                }

                @Override
                public void onFailure(Call<T> call, Throwable t) {
                    if (!emitter.isDisposed()) {
                        emitter.onError(t);
                    }
                }
            });

        }
    });

}
}

我的客户RepoFactory

public Observable<ResponseClients> getApiClients(){
        String token = preferences.getValue(SDConstants.token);
        return RxHelper.getObserbable(apiNetwork.getClients(token));
}

我的客户回购

@Override
public Observable<ResponseClients> listClients() {
    return factory.listClients();
}

Answer 1

我不使用 room 但熟悉 rxjava 你可以这样设计你的存储库

\n\n

你的房间界面

\n\n

@Query(\xe2\x80\x9cSELECT * FROM Users WHERE id = :userId\xe2\x80\x9d)\nSingle<User> getUserById(String userId);\n

\n\n

当使用 :
\n Maybe 当数据库中没有用户并且查询没有返回行时， Maybe 将完成。

\n\n

Flowable每次更新用户数据时，Flowable对象都会自动发出，允许您根据最新数据更新UI

\n\n

Single当数据库中没有用户且查询没有返回行时，Single会触发onError(EmptyResultSetException.class)

\n\n

阅读有关 Room 和 RxJava 的更多信息此链接

\n\n

要实现“如果数据库中没有数据，则调用Web服务”创建您的存储库方法

\n\n

public Single<User> getUserById(String userId){\n return  db.getUserById(userId)\n              /// if there is no user in the database get data from api\n             .onErrorResumeNext(api.getUserById(userId)\n              .subscribeOn(Schedulers.io())\n              //check your request\n              .filter(statusPojo::getStatus)\n               // save data to room\n              .switchMap(data -> {\n              //sava data to db\n              return Observable.just(data)\n              })\n           );\n\n}\n

\n\n

最后从交互器调用存储库方法，将可观察到的交互器传递给交互器，然后传递给演示布局

\n\n

更详细：您可以将 Api 和 DB 注入您的存储库

\n\n

反应式数据库的update_Answer \n如果你想获得 UI 上的最新更新，只需执行以下操作：

\n\n

你的房间界面：

\n\n

@Query(\xe2\x80\x9cSELECT * FROM Users WHERE id = :userId\xe2\x80\x9d)\nFlowable<User> getUserById(String userId);\n

\n\n

存储库：

\n\n

   @Override\npublic Flowable<User> getUser(int id) {\n    getUserFromNet(id);\n         //first emit cache data in db and after request complete   emit last update from net \n        return db.getUserById(id);\n\n }\n\n\n private Flowable<User> getUserFromNet(int id){\n      api.getUserById(userId)\n          .subscribeOn(Schedulers.io())\n          .observeOn(Schedulers.io())\n          //check your request\n          .filter(statusPojo::getStatus)\n           // save data to room\n          .subscribe(new DisposableObserver<User>() {\n                @Override\n                public void onNext(User user) {\n                     // save data to room\n                }\n\n                @Override\n                public void onError(Throwable e) {\n                    Timber.e(e);\n                }\n\n                @Override\n                public void onComplete() {\n\n\n                }\n            });\n}\n

\n\n

update_Answer2对于反应式数据库和“如果数据库中没有数据调用 Web 服务”\n根据这个问题最好使用 return aFlowable <List<T>>

\n\n

并检查列表大小而不是Flowable<T>白色swichIfEmpity，因为如果数据库中没有任何用户Flowable<T>不调用onNext()也不发出FlowableEmpity();

\n\n

private Flowable<List<User>>  getUser(int id){\n       return db.getUserById(id).\n         /// if there is no user in the database get data from \n           .flatMp(userList-> \n           if(userList.size==0)\n          api.getUserById(userId)\n          .subscribeOn(Schedulers.io())\n          //check your request\n          .filter(statusPojo::getStatus)\n           // save data to room\n          .subscribe(new DisposableObserver<User>() {\n                @Override\n                public void onNext(User user) {\n                     // save data to room\n                }\n\n                @Override\n                public void onError(Throwable e) {\n                    Timber.e(e);\n                }\n\n                @Override\n                public void onComplete() {\n\n\n                }\n            });\n                return Flowable.just(data)\n                );\n}\n

\n\n

Kotlin 方式与改造，分页（pagingRX androidx）和房间：

\n\n

房道：

\n\n

@Dao\nabstract class UserDao   {\n\n@Query("SELECT * FROM users ")\nabstract fun findAll(): DataSource.Factory<Int, User>\n}\n

\n\n

存储库：

\n\n

private fun getFromDB(pageSize:Int): Flowable<PagedList<User>> {\n    return RxPagedListBuilder(userDao.findAll(), pageSize)\n        .buildFlowable(BackpressureStrategy.LATEST)\n}\n\n\nprivate fun getApi(page: Int,pageSize: Int): Disposable {\n    return api.getUserList("token", page = page,perPage = pageSize)\n        .subscribeOn(Schedulers.io())\n        .observeOn(Schedulers.io())\n        .subscribe { t1: List<User>?, t2: Throwable? ->\n            t1?.let {\n                if (it.isNotEmpty())\n                    userDao.insert(it)\n            }\n        }\n}\n\noverride fun  findAll(page: Int ,pageSize:Int ): \nFlowable<PagedList<User>> {\n    return getFromDB(pageSize).doOnSubscribe { getApi(page,pageSize) }\n}\n

\n