Best way to cache data in Android with ViewModel and RxJava2

Question

In my application I'm using RxJava2 and new class from Architecture Components ViewModel. In my case, I need to push SQL clause to ViewModel, which will do some magic and return Observable that will give me the data I need. Everything works fine, but I am not sure if I am using RX in the best way.

My data flow: ViewModel has PublishSubject on which I am pushing SQL's. ViewModel has also Observable which is created by mapping subject. Also, I used distinctUntilChanged on Subject, to prevent from executing the same query again. To cache data I used replay(1).autoconnect(1) on Observable, but that approach had a flaw. Sometimes my Subject pushed Sql when Observable wasn't yet connect, and my data never arrived to me. Should I use BehaviourSubject? Or maybe I shouldn't use replay(1).autoconnect(1) in the first place? Or maybe my whole flow is wrong? Example:

val listSubject: Subject<RawSql> = PublishSubject.create()
val sqlListEmitter: Observable<List<T>> =
        listSubject
                .subscribeOn(Schedulers.computation())
                .map { // SOME MAGIC HERE }
                .replay(1).autoConnect(1, { compositeDisposable.add(it) })

Answer 1

In your case autoConnect() just waits for the first subscription to connect() to your stream. Since your subject and your stream build an inherent entity, you might not want to wait for it at all and instead connect it directly.

val listSubject: Subject<RawSql> = PublishSubject.create()
val sqlListEmitter: Observable<List<T>> =
        listSubject
                .observeOn(Schedulers.computation())
                .map { // SOME MAGIC HERE }
                .replay(1)
                .let {
                    it.connect(compositeDisposable::add)
                    it.publish()
                }

Also you might need to change subscribeOn() to observeOn(). The subject emits on the same thread as the data is pushed to it and does not consider the thread it's subscribed on.