1
\$\begingroup\$

I have used RxJava to Observe multiple sources and then perform action depending upon values of Observables. I want to know if it can be further simplified.

Goal: Complete implementation of RxJava.

What I have:

  1. Room (Database ORM) for storing local database using RxJava.
  2. WebService API with Retrofit using RxJava.

Conditions:

  1. Run Webservice call at regular interval.
  2. Webservice call execute depending on result from Data retrieved from Database.

Code:

Dao class

@Query("select * from Login")
Flowable<List<Login>> getLoginRx();

API class

@Multipart
@POST("..")
Flowable<NotificationResponse> notification();

Java class

private Subscriber<List<Login>> loginSubscriber;
private Subscriber<NotificationResponse> notifSubscriber;
private Observer<Long> timerObserver;
private RoomDb roomDb;

@Override
protected void onCreate(Bundle b){
    roomDb = RoomDb.getInstance(this);
    notifSubscriber = new Subscriber<NotificationResponse>() {
        @Override
        public void onSubscribe(Subscription s) {
            s.request(1);
            notifSubscription = s;
        }

        @Override
        public void onNext(NotificationResponse nr) {
            if (nr.getSuccess()) {
                if (NetworkClient.STATUS_OK.equals(nr.getData().getStatus())) {
                    updateCounts(nr.getData());
                }
            }
        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onComplete() {

        }
    };

    timerObserver = new Observer<Long>() {
        @Override
        public void onSubscribe(Disposable d) {
            networkCallRepeat = d;
        }

        @Override
        public void onNext(Long aLong) {
            NetworkClient.getInstance(MainActivity.this).getNetworkAPI().notification(NetworkClient.stringPart(NetworkAPI.STAFF_TOKEN))
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribeOn(Schedulers.io())
                    .subscribe(notifSubscriber);
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    };
    loginSubscriber = new Subscriber<List<Login>>() {
        @Override
        public void onSubscribe(Subscription s) {
            s.request(1);
            loginSubscription = s;
        }

        @Override
        public void onNext(List<Login> logins) {
            if (logins != null && logins.size() > 0) {
                updateLoginState(logins.get(0));
                Observable
                        .interval(0L, 1L, TimeUnit.MINUTES)
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(timerObserver);
            } else {
                if (networkCallRepeat != null && !networkCallRepeat.isDisposed()) {
                    networkCallRepeat.dispose();
                }
                updateLoginState(null);
            }
        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onComplete() {

        }
    };
}

@Override
protected void onResume() {
    super.onResume();
    roomDb.loginDao().getLoginRx()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(loginSubscriber);
}
\$\endgroup\$
  • \$\begingroup\$ I believe extracting those subscribers to separate classes could help readability (but it's not related to rxJava at all :) ). Have you altered your code since then? \$\endgroup\$ – Antek Aug 13 '18 at 8:26

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.