Run DB Update method after Observable result RxJava

Question

Hello

The App: I have a very basic app which have some Machines (name, id, total_income) and some Incomes (id, money, note, machines_id). On the first screen I allow the users to add a machine from a FAB which then I display it in a RecyclerView. When the user clicks any machine I navigate them to the second screen, where the user can see the name of the machine, total income and a RecyclerView with its corresponding Income; There's a FAB which enables them to add the income for that machine and refreshes the RecyclerView.

The problem: I been failing to translate from the conventional world to RxJava.

I have managed to make it work using Room .allowMainThreadQueries() and conventional methods.

MachineViewModel

public long updateByID(long id, double total_income){
    return machinesDB.getMachineDAO().updateMachineByID(id, total_income);
}

MachineDAO

@Query("update machines set total_income = :total_income where id = :id")
int updateMachineByID(long id, double total_income);

MachineInfo Activity

disposable.add(incomeViewModel.getIncomeOfMachine(id)
             .defaultIfEmpty(0.0)
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
             .subscribe(total_amount -> {
                     if (total_amount != null) {
                        // This works correctly using .allowMainThreadQueries() and conventional methods
                        machineViewModel.updateByID(id, total_amount);

                        DecimalFormat formatter = new DecimalFormat("$#,##0.000");
                        String formatted = formatter.format(total_amount);
                        mMoney.setText(formatted);

                        Toast.makeText(this, "MachineInfo: " + String.valueOf(machineViewModel.updateByID(id, total_amount)), Toast.LENGTH_SHORT).show();

                         showMenu = true;
                     } else {
                         mMoney.setText("$0.0");
                     }
             }, throwable -> Log.e(TAG, "MachineInfo: ERROR", throwable )));

I know Room needs to be instantiated in a background thread and thats why I am using RxJava. But when I try to translate the above methods into RxJava World like the below methods I'm failing to update but not crashing, "Return Value of the method is never used".

MachineViewModel

public Completable updateByID(long id, double total_income){
    return Completable.fromAction(() -> machinesDB.getMachineDAO().updateMachineByID(id, total_income));
}

MachineDAO

@Query("update machines set total_income = :total_income where id = :id")
int updateMachineByID(long id, double total_income);

Try # 1: Failure

private PublishSubject<Double> publishSubject = PublishSubject.create();
private Observable<Double> clickEvent = publishSubject;

/*
/ other stuff in here
*/

disposable.add(incomeViewModel.getIncomeOfMachine(id)
            .defaultIfEmpty(0.0)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(total_amount -> {
                        if (total_amount != null) {

                            publishSubject.onNext(total_amount);

                            DecimalFormat formatter = new DecimalFormat("$#,##0.000");
                            String formatted = formatter.format(total_amount);
                            mMoney.setText(formatted);
                            showMenu = true;
                        } else {
                            mMoney.setText("$0.0");
                        }
                    }, throwable -> Log.d(TAG, "MachineInfo: ERROR")));

    disposable.add(clickEvent.subscribe(
            total_amount -> machineViewModel.updateByID(id, total_amount)));

Try # 2: Failure

MachineViewModel

public Completable updateByID(long id, double total_income){
    return Completable.fromCallable(() -> machinesDB.getMachineDAO().updateMachineByID(id, total_income));
}

MachineDAO

@Query("update machines set total_income = :total_income where id = :id")
int updateMachineByID(long id, double total_income);

MachineInfo Activity

disposable.add(incomeViewModel.getIncomeOfMachine(id)
             .defaultIfEmpty(0.0)
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
             .subscribe(total_amount -> {
                     if (total_amount != null) {
                        // Completable.fromCallable()
                        machineViewModel.updateByID(id, total_amount);

                        DecimalFormat formatter = new DecimalFormat("$#,##0.000");
                        String formatted = formatter.format(total_amount);
                        mMoney.setText(formatted);

                        Toast.makeText(this, "MachineInfo: " + String.valueOf(machineViewModel.updateByID(id, total_amount)), Toast.LENGTH_SHORT).show();

                         showMenu = true;
                     } else {
                         mMoney.setText("$0.0");
                     }
             }, throwable -> Log.e(TAG, "MachineInfo: ERROR", throwable )));

Michał Baran Michał Baran · Accepted Answer · 2018-04-10T22:33:35

What do you mean by failing? Is the database not getting updated or are you getting some exceptions?

Anyway, the main problem I see is that you are not subscribing to your Completable objects - without this, they won't be executed.

So change:

machineViewModel.updateByID(id, total_amount);

to for example:

machineViewModel.updateByID(id, total_amount).subscribeOn(Schedulers.io()).observeOn(Schedulers.io()).subscribe();

and it will work (of course, you might wanna add specific Subscriber to subscribe method).

Run DB Update method after Observable result RxJava

Hello

2 Answers