RxJava Flowable 在 Room db 中的数据发生变化时不调用 Subscriber onNext

Question

我正在构建一个 Android 应用程序，其中 Android Room 作为持久层和 RxJava 2。

在我的 UI 中，我订阅了一个由 Room dao 返回的 Flowable：

MainActivity.java

mTaskViewModel.getTasksForDay(SessionManager.getInstance(this).getUser().getId(), CalendarManager.getInstance().getDayString(CalendarManager.getInstance().getSelectedDate()))
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(taskModelsSubscriber);

MainActivity 中的订阅者

private Subscriber<List<Task>> taskModelsSubscriber = new Subscriber<List<Task>>() {
        @Override
        public void onSubscribe(Subscription s) {
            s.request(Long.MAX_VALUE);
            runOnUiThread(new Runnable() {
                @Override
                public void run() {
                    mRefreshLayout.setRefreshing(true);
                }
            });
        }

        @Override
        public void onNext(List<Task> tasks) {
            mTasksList = tasks;
            scheduleNotifications();
            Collections.sort(mTasksList);

            runOnUiThread(new Runnable() {
                @Override
                public void run() {
                    TaskArrayAdapter taskArrayAdapter = new TaskArrayAdapter(MainActivity.this, mTasksList, SessionManager.getInstance(getApplicationContext()).getUser());
                    tasksListView.setAdapter(taskArrayAdapter);
                    mRefreshLayout.setRefreshing(false);
                }
            });
        }

        @Override
        public void onError(Throwable t) {
            Log.d(TAG, t.getMessage());
            runOnUiThread(new Runnable() {
                @Override
                public void run() {
                    mRefreshLayout.setRefreshing(false);
                    Toast.makeText(context, R.string.calendar_download_failed, Toast.LENGTH_SHORT).show();
                }
            });
        }

        @Override
        public void onComplete() {
            mRefreshLayout.setRefreshing(false);
        }
    };

这似乎有效。但是，现在我想从另一个 class （重复作业）更新基础数据。

一些class:

TaskDataSource dataSource = Injection.provideTaskDataSource(context);

                        List<String> uuids = new ArrayList<>();
                        for (Task task : tasks){
                            try {
                                task.setUserId(userId);
                                Task found = dataSource.getTask(task.getUuid(), userId, today, task.getType()).blockingGet();

                                if (found != null){
                                    dataSource.updateTask(task);
                                } else {
                                     //Task doesn't exist, insert
                                    dataSource.insertTask(task);
                                }
                                uuids.add(task.getUuid());
                            } catch(NoSuchElementException ex) {
                                Log.d(TAG, ex.getMessage());
                            }
                        }

因为我在 Whosebug 上的某个地方读到要使它起作用，所以两个 classes 都需要使用相同的 DAO 实例，所以我制作了 TaskDataSource class，它是一个 Singleton 并被提供给与我在 MainActivity (mTaskViewModel) 中使用的 ViewModel 相同的 TaskDao 实例。

但是还是不行。但是 classes 可以访问数据，但是插入、删除或更新不会在 MainActivity 的 Subscriber 中触发对 onNext 的调用。

供参考，这里是缺少的 classes:

Injection.java

public class Injection {
    public static UserDataSource provideUserDataSource(Context context) {
        DeliciousDatabase database = DeliciousDatabase.getInstance(context);
        return new LocalUserDataSource(database.userDao());
    }

    public static TaskDataSource provideTaskDataSource(Context context) {
        //DeliciousDatabase database = DeliciousDatabase.getInstance(context);
        return LocalTaskDataSource.getInstance(context);
    }

    public static ViewModelFactory provideViewModelFactory(Context context) {
        UserDataSource userDataSource = provideUserDataSource(context);
        TaskDataSource taskDataSource = provideTaskDataSource(context);

        return new ViewModelFactory(userDataSource, taskDataSource);
    }
}

LocalTaskDataSource（实现 TaskDataSource）

public class LocalTaskDataSource implements TaskDataSource {

    private final TaskDao mTaskDao;

    private static LocalTaskDataSource mInstance;

    private LocalTaskDataSource(TaskDao taskDao) {
        mTaskDao = taskDao;
    }

    public static LocalTaskDataSource getInstance(Context context) {
        if(mInstance == null) mInstance = new LocalTaskDataSource(DeliciousDatabase.getInstance(context).taskDao());
        return mInstance;
    }

    @Override
    public Flowable<List<Task>> getTasksForDay(String userId, String day) {
        return Maybe.zip(
                mTaskDao.getQuestionnairesForDay(userId, day),
                mTaskDao.getPhysicalTestsForDay(userId, day),
                mTaskDao.getSpeechTestsForDay(userId, day),
                mTaskDao.getStressTestsForDay(userId, day),
                mTaskDao.getPlateControlsForDay(userId, day),
                mTaskDao.getWeeklyFeedbacksForDay(userId, day),
                (questionnaires, physicalTests, speechTests, stressTests, plateControls, weeklyFeedbacks) -> {
                    List<Task> combined = new ArrayList<>();
                    combined.addAll(questionnaires);
                    combined.addAll(physicalTests);
                    combined.addAll(speechTests);
                    combined.addAll(stressTests);
                    combined.addAll(plateControls);
                    combined.addAll(weeklyFeedbacks);
                    return combined;
                }
        ).toFlowable();
    }
}

和任务视图模型：

public class TaskViewModel extends ViewModel {

    private final TaskDataSource mDataSource;

    private List<Task> mTasks;

    public TaskViewModel(TaskDataSource dataSource) {
        mDataSource = dataSource;
    }

    public Flowable<List<Task>> getTasksForDay(String userId, String day) {
        return mDataSource.getTasksForDay(userId, day);
    }

    public Maybe<? extends Task> getTask(int id, Task.Type type) {
        return mDataSource.getTaskById(id, type);
    }

    public void updateTask(Task task) {
        mDataSource.updateTask(task);
    }

    public Maybe<List<TaskStats>> getWeeklyStatistics(long start, long end) {
        return mDataSource.getWeeklyStatistics(start, end);
    }
}

Answer 1

似乎是继承的罪魁祸首。如您所见，我正在执行的 getTasksForDay 查询 returns a Flowable<List<Task>>。但是，它实际上是 Task 个子类的列表。当更改的基础数据类型不完全相同时，Room 查询似乎不会发出。因此，如果您调用 updateTask((Questionnaire) task)，Flowable<List<Task>> 不会发出，但是当您调用 updateTask((Task) task) 时它会发出。

RxJava Flowable 在 Room db 中的数据发生变化时不调用 Subscriber onNext

RxJava Flowable not calling Subscriber onNext when data changes in Room db

android

rx-java2

android-room