TasksRepository implements TasksDataSource { @Override public Flowable<Optional<Task>> getTask(@NonNull final String taskId) { checkNotNull(taskId); final Task cachedTask = getTaskWithId(taskId); if (cachedTask != null) { return Flowable.just(Optional.of(cachedTask)); } if (mCachedTasks == null) { mCachedTasks = new LinkedHashMap<>(); } Flowable<Optional<Task>> localTask = getTaskWithIdFromLocalRepository(taskId); Flowable<Optional<Task>> remoteTask = mTasksRemoteDataSource .getTask(taskId) .doOnNext(taskOptional -> { if (taskOptional.isPresent()) { Task task = taskOptional.get(); mTasksLocalDataSource.saveTask(task); mCachedTasks.put(task.getId(), task); } }); return Flowable.concat(localTask, remoteTask) .firstElement() .toFlowable(); } private  TasksRepository(@NonNull TasksDataSource tasksRemoteDataSource,
                            @NonNull TasksDataSource tasksLocalDataSource); static TasksRepository getInstance(@NonNull TasksDataSource tasksRemoteDataSource,
                                              @NonNull TasksDataSource tasksLocalDataSource); static void destroyInstance(); @Override Flowable<List<Task>> getTasks(); @Override void saveTask(@NonNull Task task); @Override void completeTask(@NonNull Task task); @Override void completeTask(@NonNull String taskId); @Override void activateTask(@NonNull Task task); @Override void activateTask(@NonNull String taskId); @Override void clearCompletedTasks(); @Override Flowable<Optional<Task>> getTask(@NonNull final String taskId); @Override void refreshTasks(); @Override void deleteAllTasks(); @Override void deleteTask(@NonNull String taskId);  }
@Test public void getTask_requestsSingleTaskFromLocalDataSource() { Task task = new Task(TASK_TITLE, "Some Task Description", true); Optional<Task> taskOptional = Optional.of(task); setTaskAvailable(mTasksLocalDataSource, taskOptional); setTaskNotAvailable(mTasksRemoteDataSource, taskOptional.get().getId()); TestSubscriber<Optional<Task>> testSubscriber = new TestSubscriber<>(); mTasksRepository.getTask(task.getId()).subscribe(testSubscriber); Mockito.verify(mTasksLocalDataSource).getTask(Matchers.eq(task.getId())); testSubscriber.assertValue(taskOptional); }