LiveData配合RxJava

Android JetPack现在已经在android开发中大行其道，以下是个人使用LiveData与数据库查询的一点总结。

LiveData + Room

官方已经出了rxRoom，DAO中已经可以直接使用RxJava。这里有一个使用LiveData+Room的例子 PDFReader

@Dao
public interface PdfDAO {

    @Query("SELECT * FROM Pdf WHERE read_time != 0 ORDER BY read_time DESC")
    Flowable<List<Pdf>> getRecentPdf();

    @Query("SELECT * FROM Pdf WHERE read_time != 0 ORDER BY read_time DESC")
    Flowable<List<Pdf>> getAllPdf();

    @Query("SELECT * FROM Pdf WHERE file_path = :path LIMIT 1")
    Single<Pdf> getPdf(String path);

    @Insert(onConflict = OnConflictStrategy.REPLACE)
    Completable insertPdf(Pdf pdf);

    @Delete
    int deletePdf(Pdf pdf);
}

返回结果可以是：Maybe，Flowable，Single，其中Completeable目前只有androidx支持

其中Flowable会一直监听数据库变化，具体如何实现的以后再分享

于是就可以使用Flowable来实时更新界面：

public class DocumentLiveData extends LiveData<List<Pdf>> {
    private boolean mRecentLivaData;
    private PdfDAO mPdfDao;
    private Disposable mDisposable;

    public DocumentLiveData(Context context, boolean recent) {
        mRecentLivaData = recent;
        mPdfDao = AppDatabase.getInstance(context).getPdfDAO();
    }

    @Override
    protected void onActive() {
        super.onActive();
        Flowable<List<Pdf>> flowable;
        if (mRecentLivaData) {
            flowable = mPdfDao.getRecentPdf();
        } else {
            flowable = mPdfDao.getAllPdf();
        }

        mDisposable = flowable.subscribeOn(Schedulers.io())
                .subscribe(this::postValue,
                        t -> Log.e("PdfRepository", "Query pdf items failed", t));
    }

    @Override
    protected void onInactive() {
        if (mDisposable != null) {
            mDisposable.dispose();
            mDisposable = null;
        }
        super.onInactive();
    }
}

LiveData + RxJava + ContentProvider

上面的方法适用与本地自建数据库，有时需要监听第三方数据库（多通过ContentProvider提供）
以下是一个例子，查询系统CalendarProvider中有事件的天：

public class EventDayLiveData extends MutableLiveData<List<Integer>> {

    private static final String[] PROJECTION = new String[]{
            Instances.START_DAY,
            Instances.END_DAY
    };
    private static final int INDEX_START_DAY = 0;
    private static final int INDEX_END_DAY = 1;

    public static final String SELECTION = Calendars.VISIBLE + "=1";

    private ContentResolver mContentResolver;
    private Handler mMainHandler;

    private ContentObserver mContentObserver;

    private PublishProcessor<Object> triggers;
    private Disposable mEventBus;
    private CompositeDisposable mQueries;

    // 事件加载的起始天, julianDay
    private int mStartDay;
    // 事件加载的结束天, julianDay
    private int mEndDay;

    private Runnable mTimeChangeUpdater = new Runnable() {
        @Override
        public void run() {
            // reload event day
            if (triggers != null) {
                triggers.onNext("TimeChange");
            }
        }
    };

    public EventDayLiveData(Context content) {
        mMainHandler = new Handler(Looper.getMainLooper());
        mContentResolver = content.getApplicationContext().getContentResolver();
        mQueries = new CompositeDisposable();
        triggers = PublishProcessor.create();

        mContentObserver = new ContentObserver(mMainHandler) {
            @Override
            public void onChange(boolean selfChange) {
                // reload event day
                if (triggers != null) {
                    triggers.onNext("EventChange");
                }
            }
        };

        TimeCalendar time = new TimeCalendar(1970, 0, 1);
        mStartDay = time.getJulianDay();
        time.set(2037, 11,31);
        mEndDay = time.getJulianDay();
    }

    @Override
    protected void onActive() {
        Timber.d("    onActive");
        // 过滤 200 毫秒内的多次加载请求
        mEventBus = triggers.throttleFirst(200, TimeUnit.MILLISECONDS)
                .doOnNext(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        Timber.d(" receive action : %s", o.toString());
                    }
                })
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        loadEventDays();
                    }
                });

        //Utils.setMidnightUpdater(mMainHandler, mTimeChangeUpdater, TimeZone.getDefault().getID());
        mContentResolver.registerContentObserver(CalendarContract.CONTENT_URI, true, mContentObserver);
        triggers.onNext("InitValue");
    }

    @Override
    protected void onInactive() {
        Timber.d("    onInactive");
        mContentResolver.unregisterContentObserver(mContentObserver);
        //Utils.resetMidnightUpdater(mMainHandler, mTimeChangeUpdater);

        if (mQueries != null) {
            mQueries.dispose();
        }

        if (mEventBus != null) {
            mEventBus.dispose();
        }
    }

    private void loadEventDays() {
        final Disposable disposable = Observable.create(
                new ObservableOnSubscribe<List<Integer>>() {
                    @Override
                    public void subscribe(ObservableEmitter<List<Integer>> emitter) {
                        if (emitter.isDisposed()) {
                            return;
                        }

                        try {
                            emitter.onNext(queryEventDay());
                            emitter.onComplete();
                        } catch (Exception e) {
                            emitter.onError(e);
                        }
                    }
                })
                .filter(new Predicate<List<Integer>>() {
                    @Override
                    public boolean test(List<Integer> list) {
                        // 判断数据是否变化
                        return eventDaysChanged(getValue(), list);
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> list) {
                        setValue(list);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) {
                        Timber.i(throwable,"Query event day failed");
                    }
                });

        mQueries.add(disposable);
    }

    private List<Integer> queryEventDay() {
        Uri.Builder builder = EventDays.CONTENT_URI.buildUpon();
        ContentUris.appendId(builder, mStartDay);
        ContentUris.appendId(builder, mEndDay);
        Uri uri = builder.build();


        Cursor c = mContentResolver.query(uri, PROJECTION, SELECTION, null, null);
        List<Integer> dayList = new ArrayList<>();

        if (c != null && c.moveToFirst()) {
            while (!c.isAfterLast()) {
                int startDay = c.getInt(INDEX_START_DAY);
                int endDay = c.getInt(INDEX_END_DAY);
                if (endDay != startDay) {
                    for (int i = startDay; i <= endDay; i++) {
                        if (!dayList.contains(i)) {
                            dayList.add(i);
                        }
                    }
                } else {
                    if (!dayList.contains(startDay)) {
                        dayList.add(startDay);
                    }
                }
                c.moveToNext();
            }
        }

        if (c != null) {
            c.close();
        }

        return dayList;
    }

    private boolean eventDaysChanged(List<Integer> oldList, List<Integer> newList) {
        if (oldList == null) {
            return true;
        }

        if (oldList.size() != newList.size()) {
            return true;
        }

        return !oldList.containsAll(newList);
    }
}

主要思路就是通过注册数据库变化来持续获取事件流丢给PublishProcessor下发到下游订阅者处理