Как решить подписки PublishSubject, MutableLiveData, RxJava и Disposable в kotlin и java?

Котлайн раздел

1. Я написал BaseViewModel.kt, используя приведенную ниже конфигурацию.

val ioDisposables: CompositeDisposable
val inputs: PublishSubject<Response<List<X>>>
val outputs: MutableLiveData<Response<List<X>>>

init {
    outputs = rx.toLiveData(inputs, ioDisposables)
}

override fun onCleared() {
    ioDisposables.clear()
    super.onCleared()
}

fun observeOutputs(owner: LifecycleOwner, observer: Observer<Response<List<X>>>) {
    multipleOwners.add(owner)
    outputs.reObserve(owner, observer)
}

2. RxMapper содержит динамические привязки для toLiveData () api, как показано ниже:

fun <T> toLiveData(subject: PublishSubject<T>, disposables: CompositeDisposable): MutableLiveData<T> {
    val data = MutableLiveData<T>()
    val disposable = subject.subscribe({ data.setValue(it) })
    disposables.add(disposable)
    return data
} 

3. LiveDataMapper предназначен для reObserve () api

fun <T> LiveData<T>.reObserve(owner: LifecycleOwner, observer: Observer<T>) {
    removeObserver(observer)
    observe(owner, observer)
} 

Раздел Java - из класса Fragments

4. Подписка на выходы в первый раз

    vm = ViewModelProviders.of(this, factory).get(FlagViewModel.class);
    vm.observeOutputs(this, this::processResponse);

Этот код работает нормально в первый раз, но он заполняет повторяющийся набор данных во второй раз. :( :(