RxJava2 in CursorLoader’s onLoadFinished callback

Question

To get data from database I use CursorLoader in the app. Once onLoadFinished() callback method calls the logic of app converts Cursor object to List of objects within business model requirements. That conversion (heavy operation) takes some time if there is a lot of data. That slows UI thread. I tried to start conversion in non-UI Thread using RxJava2 passing Cursor object, but got Exception:

Caused by: android.database.StaleDataException: Attempting to access a closed CursorWindow.Most probable cause: cursor is deactivated prior to calling this method.

Here is the part of Fragment's code:

@Override
    public Loader<Cursor> onCreateLoader(int id, Bundle args) {
        QueryBuilder builder;
        switch (id) {
            case Constants.FIELDS_QUERY_TOKEN:
                builder = QueryBuilderFacade.getFieldsQB(activity);
                return new QueryCursorLoader(activity, builder);
            default:
                return null;
        }
    }

    @Override
    public void onLoadFinished(Loader<Cursor> loader, Cursor cursor) {
        if (cursor.getCount() > 0) {
            getFieldsObservable(cursor)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(this::showFields);
        } else {
            showNoData();
        }
    }

private static Observable<List<Field>> getFieldsObservable(Cursor cursor) {
            return Observable.defer(() -> Observable.just(getFields(cursor))); <-- Exception raised at this line

        }

private static List<Field> getFields(Cursor cursor) {
            List<Field> farmList = CursorUtil.cursorToList(cursor, Field.class);
            CursorUtil.closeSafely(cursor);
            return farmList;
        }

The purpose of using CursorLoader here is to get notifications from DB if there is data store updated.

Update As Tin Tran suggested, I removed CursorUtil.closeSafely(cursor); and now I get another exception:

Caused by: java.lang.IllegalStateException: attempt to re-open an already-closed object: /data/user/0/com.my.project/databases/db_file
                                                          at android.database.sqlite.SQLiteClosable.acquireReference(SQLiteClosable.java:55)
                                                          at android.database.CursorWindow.getNumRows(CursorWindow.java:225)
                                                          at android.database.sqlite.SQLiteCursor.onMove(SQLiteCursor.java:121)
                                                          at android.database.AbstractCursor.moveToPosition(AbstractCursor.java:236)
                                                          at android.database.AbstractCursor.moveToNext(AbstractCursor.java:274)
                                                          at android.database.CursorWrapper.moveToNext(CursorWrapper.java:202)
                                                          at com.db.util.CursorUtil.cursorToList(CursorUtil.java:44)
                                                          at com.my.project.MyFragment.getFields(MyFragment.java:230)

cursorToList() method of CursorUtil

public static <T> ArrayList<T> cursorToList(Cursor cursor, Class<T> modelClass) {
        ArrayList<T> items = new ArrayList<T>();
        if (!isCursorEmpty(cursor)) {
            while (cursor.moveToNext()) { <-- at this line (44) of the method raised that issue
                final T model = buildModel(modelClass, cursor);
                items.add(model);
            }
        }
        return items;
    }

Ljoka

Not an answer, but I would strongly recommend having a look at sqlbrite. – Luthuli

@Lukasz, yes, I know about sqlbrite and it would be a good solution for that, but I can't add it to the project as it already big and used another orm. – Ljoka

May this list be updated whilst the previous data is being processed? – Variolite

Answer 1

As you can see from my comment to your question, I was interested whether the data is being updated while getFieldsObservable() hasn't been yet returned. I received the info I was interested in your comment.

As I can judge, here's what happens in your case:

• onLoadFinished() is called with Cursor-1
• RxJava's method is being executed on another thread with Cursor-1 (hasn't yet been finished, here Cursor-1 is being used)
• onLoadFinished() is called with Cursor-2, LoaderManager API takes care of closing Cursor-1, which is still being queried by RxJava on another thread

Thus, an exception results.

So, you'd better stick with creating your custom AsyncTaskLoader (which CursorLoader extends from). This AsyncTaskLoader will incorporate all the logics that CursorLoader has (basically one-to-one copy), but would return already sorted/filter object in onLoadFinished(YourCustomObject). Thus, the operation, that you wished to perform using RxJava would actually be done by your loader in it's loadInBackground() method.

Here's the snapshot of the changes that MyCustomLoader will have in loadInBackground() method:

public class MyCustomLoader extends AsyncTaskLoader<PojoWrapper> {
  ...
  /* Runs on a worker thread */
  @Override
  public PojoWrapper loadInBackground() {
    ...
    try {
      Cursor cursor = getContext().getContentResolver().query(mUri, mProjection, mSelection,
          mSelectionArgs, mSortOrder, mCancellationSignal);
      ...

      // `CursorLoader` performs following:
      // return cursor;

      // We perform some operation here with `cursor`
      // and return PojoWrapper, that consists of `cursor` and `List<Pojo>`
      List<Pojo> list = CursorUtil.cursorToList(cursor, Field.class);
      return new PojoWrapper(cursor, list);
    } finally {
      ...
    }
  }
  ...
}

Where PojoWrapper is:

public class PojoWrapper {
  Cursor cursor;
  List<Pojo> list;

  public PojoWrapper(Cursor cursor, List<Pojo> list) {
    this.cursor = cursor;
    this.list = list;
  }
}

Thus, in onLoadFinished() you do not have to take care of delegating the job to another thread, because you already have done it in your Loader implementation:

@Override public void onLoadFinished(Loader<PojoWrapper> loader, PojoWrapper data) {
      List<Pojo> alreadySortedList = data.list;
}

Here's the entire code of MyCustomLoader.

Variolite

Answer 2

The loader will release the data once it knows the application is no longer using it. For example, if the data is a cursor from a CursorLoader, you should not call close() on it yourself. From: https://developer.android.com/guide/components/loaders.html

You should not close the cursor yourself which I think CursorUtil.closeSafely(cursor) does.

You can use switchMap operator to implement that. It does exactly what we want

private PublishSubject<Cursor> cursorSubject = PublishSubject.create()

public void onCreate(Bundle savedInstanceState) {
    cursorSubject
        .switchMap(new Func1<Cursor, Observable<List<Field>>>() {
             @Override public Observable<List<Field>> call(Cursor cursor) {
                  return getFieldsObservable(cursor);
             }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(this::showFields);
}

@Override
public void onLoadFinished(Loader<Cursor> loader, Cursor cursor) {
    cursorSubject.onNext(cursor)
}

You now need to modify showFields to and getFieldsObservable to account for empty Cursor

Adverbial

Yes, it does close the Cursor. But OP closes it when he no longer needs it, meaning that he has already "parsed" farmList from the Cursor and he no longer needs to leave that Cursor opened. I see no problems here. – Variolite

Yes, but after he closes the Cursor, the CursorLoader may still need to access the cursor. If it does, then an Exceptionis thrown – Adverbial

I see what you mean. I wonder whether LoaderManager is not smart enough to perform if check and load another time when Cursor is closed. – Variolite

I've tested without closing the cursor as Tin Tran suggested and got another exception, added trace as update to the question. – Ljoka

Is that the full stacktrace ? – Adverbial

@TinTran, it raised at this line List<Field> farmList = CursorUtil.cursorToList(cursor, Field.class); of getFields(). Trace is in updated. – Ljoka

The problem is Cursor is not thread-safe. While you are converting the cursor to List<Field>, a new Cursor arrive in onLoadFinished and the old Cursor is close. Therefore, the io thread accessing it will throw the exception. – Adverbial

looks like you are right, I see that onLoadFinished calls three times during 2 seconds as new objects of Cursor arrived. But, what do you suggest here is it possible somehow to move out cursorToList method from UI thread, maybe make cursor thread-safe or start CursorLoader from io thread from the beginning or maybe another solution? – Ljoka

What are u doing with the List<Field> after parsing?. Is it large ? – Adverbial

@TinTran, "Cursor is not thread-safe" I also was inclined that way, but as you can see in CursorLoader#loadInBackground, it calls getContext().getContentResolver().query(), which in turn creates the Cursor while being in background thread. Later, this Cursor would be dispatched to the UI thread, and we can safely use it. So, what's your clarification concerning Cursor's non-threadsafety? – Variolite

@Variolite I mean the usage of Cursor is not thread safe. You can close a Cursor while other thread is reading/accessing it. – Adverbial

Oh, I see what you meant. As far as I know, that has nothing to do with threadsafety. Normally, thread-safe means, that if some change is being done to the object on one thread, that change is visible for the second thread. Thus, Cursor seems to be thread-safe. – Variolite

@TinTran, getFieldsObservable(cursor) is a heavy operation, lets assume it takes 10 seconds. Now, while performing this job, another update comes from Loader, which closes the previous cursor, thus the heavy operation crashes. I cannot see how this actually will solve the problem, but yes, it is much more concise! – Variolite

We can modify the getFieldsObservable to check for subscriber.isUnsubscribe() before every accessing operation on the cursor. I think the Loader won't close the Cursor before making the new Cursor with the new result and deliver to us. If we have the new Cursor, the previous Observable on the old Cursor is unsubscribed. We can access the Subscriber using the Observable.create() to create Observable. – Adverbial

to check for subscriber.isUnsubscribe() before every accessing operation on the cursor - the problem is following: you are interacting with cursor on main thread, whereas CursorLoader creates another cursor on a background thread, so there is no synchronization between these two threads, you cannot tell "hey, CursorLoader, do not close this cursor for a while until I finish interacting with it". – Variolite