Implement Room with RxJava and Retrofit

I am trying to use Room with RxJava and Retrofit, Before You recommend use a component arch (In this opportunity is not possible, the project is in and 50% and Just need to continue with the arch clean).

So the problem is this. I have a web service that returns a POJO. Something like this:

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

POJO is more complex but for the example is ok with this. I need to do that since my view make query to invoke data from room, but if there is not data in my db call my web services,the reponse of my web services transform to entity and save in my db (room) and after return a list of data to my view.

I am using clean arch. I appreciate anyhelp with this. Again not trying to use

data layout

database
network
repository

domain

interactor
callbacks

presentation

presenter
view

POJO API response

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

My db entity

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

My dao for room

@Dao
public interface ClientsDao {

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

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

}

RxJava help class

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

        }
    });

}
}

My ClientsRepoFactory

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

My ClientsRepo

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

i dont work with room but familiar with rxjava you can design your repository like that

your room interfac

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

when use :
Maybe When there is no user in the database and the query returns no rows, Maybe will complete.

Flowable Every time the user data is updated, the Flowable object will emit automatically, allowing you to update the UI based on the latest dat

Single When there is no user in the database and the query returns no rows, Single will trigger onError(EmptyResultSetException.class)

read more about Room and RxJava this link

to achieve " if there is not data in db call web services " create your repository methode like that

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

}

finally call repository method from interactor to passed obsrevable to interactor then to presentation layout

more detail : you can inject Api and DB to your repository

update_Answer for reactive db if you want get last update on UI just do it :

your room interface:

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

repository :

   @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() {


                }
            });
}

update_Answer2 for reactive db and " if there is not data in db call web services " according this issue is better use return a Flowable <List<T>>

and check list size instead of Flowable<T> white swichIfEmpity because if don't any user in db Flowable<T> do'nt call onNext() and don't emite FlowableEmpity();

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

Kotlin way with retrofit , paging (pagingRX androidx) and room :

room Dao:

@Dao
abstract class UserDao   {

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

Repository:

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