Синхронизация данных данных в реальном времени

Question

У меня возникает проблема, когда Realm иногда возвращает мне разные данные каждый раз, когда я делаю тот же запрос. В настоящее время я использую SyncAdapter для загрузки. Идея состоит в том, что мы пытаемся реализовать автономный режим.

Итак, когда пользователь создает элемент, он добавляется в Realm db. Я генерирую идентификатор для этого элемента вручную, получая maxId и добавляя к нему 1000. После этого я отправляю itemID в UploadSyncAdapter, где я получаю itemById и отправляю его на бэкэнд, а бэкэнд возвращает мне элемент с реальным идентификатором. Поэтому после этого я удаляю старый элемент и просто вставляю новый элемент в Realm.

После того, как я вернусь и прочитаю данные, они возвращаются каждый второй раз, например массив данных размером 115, а в другой раз массив размером 116. Я даже ищу элемент с отладчиком по идентификатору, и он действительно один раз находит предмет, второй раз он этого не делает. Но похоже, что после того, как я перестрою проект, элемент там, и он работает нормально, или если я перезапущу приложение, используя Instant Run.

Любая помощь приветствуется ...

UPDATE Btw Я использую RXjava, чтобы получить данные с сервера, но в настоящее время подписаны и наблюдается на текущей нити (SyncAdapter нить).

Вот код:

@Override 
public void onNext(TaskResponse taskResponse) { 
    tasksDatabaseManager.deleteTaskById(taskId); 
    List<Task> tasks = taskResponse.getTaskDataList(); 
    tasksDatabaseManager.insertTasks(tasks); 
} 

public void deleteTaskById(int taskId){ 
    Realm realm = Realm.getDefaultInstance(); 
    realm.beginTransaction(); 
    RealmResults<Task> rows = realm.where(Task.class).equalTo(ID, taskId).findAll(); 
    rows.deleteAllFromRealm(); 
    realm.commitTransaction(); 
    realm.close(); 
} 

private void copyOrUpdateTasks(List<Task> tasksList){ 
    Realm realm = Realm.getDefaultInstance(); 
    ArrayList<Task> updatedTaskList; 
    //first initialize task permissions 
    updatedTaskList = filterTasksByPermission(tasksList); 
    //initialize custom task data 
    for (Task task : updatedTaskList) { 
     initializeTaskCustomFields(task); 
    } 
    //save new data 
    Log.d(TAG, "tasks number before update: " + getTasks().size()); 
    realm.beginTransaction(); 
    realm.copyToRealmOrUpdate(updatedTaskList); 
    realm.commitTransaction(); 
    realm.close(); 

    Log.d(TAG, "tasks number after update: " + getTasks().size()); 
} 

В filterTasksByPermission я просто вычислить некоторые разрешения для задач, но задача возвращается в список. И в initializeTaskCustomFields я также просто вычисляю 2 поля объекта перед сохранением в Realm (так что у меня есть эти значения, также сохраненные в Realm)

Answer 1

После некоторых исследований выяснилось, в чем проблема. Но, пожалуйста, поправьте меня, если я ошибаюсь. Я заменил свой Subscribers.io() на Subscribers.newThread() в моих вызовах Rx, и теперь он работает нормально. Итак, моя теория такова:

Перед вызовом моего UploadAdapter для загрузки измененных данных я использую вызов Rx с помощью Subscribers.io() для вставки моего элемента в базу данных. Subscribers.io() использует threadpool для повторного использования потоков или создает новые потоки, если это необходимо. Итак, скажем, это порождает поток под названием «А». Thread A get - это экземпляр Realm (скажем, что моментальный снимок Realm - «1») и вставляет в него созданный элемент. После этого вызывается вызов SyncAdapter и он также получает новый экземпляр Realm с тем же снимком Real «1». После того, как SyncAdapter завершает загрузку данных на сервер, он удаляет старый элемент и вставляет новый элемент, полученный с сервера. Таким образом, после того, как данные Realm изменились, новый снимок Realm теперь «2». SyncAdapter отправляет широковещательную передачу в Activity, что загрузка завершена, и она должна получать новые данные из базы данных Realm.

Для чтения данных из Realm Я также использую Rx с Subscribers.io(). Поэтому при запросе новых данных из Realm Subscribers.io() уже имеет поток в своем пуле, который ожидает повторного использования, и это поток «A». И поскольку этот поток является потоком не Looper, он не знает, что данные Realm изменились, и он по-прежнему использует снимок Realm «1», поэтому я получаю старые данные из Realm. А после обновления несколько раз Subscribers.io(), вероятно, создает новый поток, скажем, поток «B».

Так что нить «B» также получает снимок Realm, и на этот раз это самый новый моментальный снимок, поэтому снимок «2». И он возвращает правильные данные.

Таким образом, при использовании Subscribers.newThread() он всегда создает новый поток и всегда имеет новейший моментальный снимок Realm.

Вот ссылка относительно разницы между Subscribers.io() и Subscribers.newThread(): Retrofit with Rxjava Schedulers.newThread() vs Schedulers.io()

Надеется, что это помогает кто-то!