当房间数据库中的数据发生更改时,RxJava Flowable不呼叫订阅者onNext

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

在我的UI中,我订阅了由房间dao返回的Flowable:

主要活动。Java语言

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

主活动中的订阅服务器

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

这似乎奏效了。但是,现在我想从另一个类(定期作业)更新基础数据。

某些类别:

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

由于我在Stackoverflow上的某个地方读到,要使其工作,两个类都需要使用相同的DAO实例,因此我创建了TaskDataSource类,它是一个单例,并与我在MainActivity中使用的ViewModel(mTaskViewModel)提供了相同的TaskDao实例。

但它仍然不起作用。但类可以访问数据,但插入、删除或更新类不会触发对MainActivity中订阅服务器中onNext的调用。

以下是缺少的类供参考:

注射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();
    }
}

和TaskViewModel:

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