NetworkBoundSource.java

NetworkBoundSource.java

public abstract class NetworkBoundSource<LocalType, RemoteType> {

    public NetworkBoundSource(FlowableEmitter<Resource<LocalType>> emitter) {
        Disposable firstDataDisposable = getLocal()
                .map(Resource::loading)
                .subscribe(emitter::onNext);

        getRemote().map(mapper())
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                .subscribe(localTypeData -> {
                    firstDataDisposable.dispose();
                    saveCallResult(localTypeData);
                    getLocal().map(Resource::success).subscribe(emitter::onNext);
                });
    }

    public abstract Single<RemoteType> getRemote();

    public abstract Flowable<LocalType> getLocal();

    public abstract void saveCallResult(LocalType data);

    public abstract Function<RemoteType, LocalType> mapper();

}

Was asking where did you import "Resource" object from

At Line:14 Flowable returned from getLocal() is never disposed and will emit any changed that happens to the Room, but there will be no exception, because no one is subscribed to that Flowable when user navigates back from activity / fragment. I changed that line of code to this:

val secondDataDisposable = getLocal().map { Resource.success(it) }.subscribe { 
                                                                     Log.d("Flowable", "Is this called?")
                                                                    emitter.onNext(it) 
                                                               }
emitter.setDisposable(secondDataDisposable)

And it works, I do not see "Is this called?" string more then once... However I am not sure is this correct solution and will cause some additional troubles...

At line 11 ditn't you have to have an "OnError' handler?

from:

.subscribe(localTypeData -> {
                    firstDataDisposable.dispose();
                    saveCallResult(localTypeData);
                    getLocal().map(Resource::success).subscribe(emitter::onNext);
});

to:

.subscribe(localTypeData -> {
                    firstDataDisposable.dispose();
                    saveCallResult(localTypeData);
                    getLocal().map(Resource::success).subscribe(emitter::onNext);
},e -> {
                    e.printStackTrace();
});