运行 Observable结果RxJava后的数据库更新方法

Question

你好

应用程序： 我有一个非常基本的应用程序，其中有一些机器 (name, id, total_income) 和一些收入 (id, money, note, machines_id)。在第一个屏幕上，我允许用户从 FAB 添加一台机器，然后我将其显示在 RecyclerView 中。当用户单击任何机器时，我将它们导航到第二个屏幕，用户可以在其中看到机器名称、总收入和 RecyclerView 及其相应的收入；有一个 FAB 使他们能够为该机器添加收入并刷新 RecyclerView。

问题：我一直未能从传统世界转换到 RxJava。

我已经设法使用 Room .allowMainThreadQueries() 和传统方法使其工作。

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

机器信息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("[=13=].0");
                     }
             }, throwable -> Log.e(TAG, "MachineInfo: ERROR", throwable )));

我知道 Room 需要在后台线程中实例化，这就是我使用 RxJava 的原因。但是当我尝试将上述方法转换为 RxJava World 时，就像下面的方法一样，我无法更新但不会崩溃，"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);

尝试#1：失败

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("[=16=].0");
                        }
                    }, throwable -> Log.d(TAG, "MachineInfo: ERROR")));

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

尝试#2：失败

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

机器信息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("[=19=].0");
                     }
             }, throwable -> Log.e(TAG, "MachineInfo: ERROR", throwable )));

Answer 1

失败是什么意思？数据库没有更新还是您遇到了一些异常？

无论如何，我看到的主要问题是您没有订阅您的 Completable 对象 - 没有这个，它们将不会被执行。

所以改变：

machineViewModel.updateByID(id, total_amount);

例如：

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

它会起作用（当然，您可能想将特定订阅者添加到 subscribe 方法）。

Answer 2

好的，所以 感谢 Michael 我设法找到了问题中的漏洞。

显然你可以 Add Disposables inside Disposables.

您需要初始化 Observable 并添加其 .observeOn(etc).subscribeOn(etc).subscribe(etc)"

像这样在一次性内添加一次性：

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

                        disposable.add(machineViewModel.updateByID(id, total_amount)
                                .subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(
                                        () -> Log.d(TAG, "MachineInfo: COMPLETED"),
                                        throwable -> Log.e(TAG, "MachineInfo: ERROR", throwable )));

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

我注意到这是否是一个干净的答案，但它有效。

运行 Observable结果RxJava后的数据库更新方法

Run DB Update method after Observable result RxJava

android

rx-java

rx-java2

android-room

你好