用RxJava实现Room并进行改造

Question

我试图使用RxJava和Rome的改型，在你建议使用组件拱之前（在这个机会是不可能的，项目是在和50%和只需要继续与拱清理）。

所以问题是这个。我有一个返回POJO的web服务。类似于这样：

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

null

• 相互作用者
• 回调

演示文稿

• 演示者
• 查看

{
 "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();

}

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);
                    }
                }
            });

        }
    });

}
}

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，您可以这样设计存储库

您的房间interfac

@Query(“SELECT * FROM Users WHERE id = :userId”)
Single<User> getUserById(String userId);

当使用时:
可能当数据库中没有用户并且查询没有返回行时，可能会完成。

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

}

更多细节：您可以向存储库注入Api和DB

reactive db的update_Answer如果您想要获得UI上的最新更新，只需执行以下操作：

您的房间界面：

@Query(“SELECT * FROM Users WHERE id = :userId”)
Flowable<User> getUserById(String userId);

   @Override
public Flowable<User> getUser(int id) {
    getUserFromNet(id);
         //first emit cache data in db and after request complete   emit last update from net 
        return db.getUserById(id);

 }


 private Flowable<User> getUserFromNet(int id){
      api.getUserById(userId)
          .subscribeOn(Schedulers.io())
          .observeOn(Schedulers.io())
          //check your request
          .filter(statusPojo::getStatus)
           // save data to room
          .subscribe(new DisposableObserver<User>() {
                @Override
                public void onNext(User user) {
                     // save data to room
                }

                @Override
                public void onError(Throwable e) {
                    Timber.e(e);
                }

                @Override
                public void onComplete() {


                }
            });
}

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

                @Override
                public void onError(Throwable e) {
                    Timber.e(e);
                }

                @Override
                public void onComplete() {


                }
            });
                return Flowable.just(data)
                );
}

@Dao
abstract class UserDao   {

@Query("SELECT * FROM users ")
abstract fun findAll(): DataSource.Factory<Int, User>
}

存储库：

private fun getFromDB(pageSize:Int): Flowable<PagedList<User>> {
    return RxPagedListBuilder(userDao.findAll(), pageSize)
        .buildFlowable(BackpressureStrategy.LATEST)
}


private fun getApi(page: Int,pageSize: Int): Disposable {
    return api.getUserList("token", page = page,perPage = pageSize)
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.io())
        .subscribe { t1: List<User>?, t2: Throwable? ->
            t1?.let {
                if (it.isNotEmpty())
                    userDao.insert(it)
            }
        }
}

override fun  findAll(page: Int ,pageSize:Int ): 
Flowable<PagedList<User>> {
    return getFromDB(pageSize).doOnSubscribe { getApi(page,pageSize) }
}