This sample shows replayed response (that could have a param) from an Observable Subject used like a proxy between the client and original Observable

main.kt

import io.reactivex.Observable
import io.reactivex.subjects.PublishSubject
import java.util.*

fun main(args: Array<String>) {
    val vm = VM(Repo())
    vm.observableData.subscribe {
        println("from sub1 $it")
    }
    Thread.sleep(3333)

    vm.observableData.subscribe {
        println("from sub2 $it")
    }

    vm.requestData("JayZeeBra")

    Thread.sleep(3333)

    vm.observableData.subscribe {
        println("from sub3 $it")
    }

}

class VM(repo: Repo) {
    private val subject: PublishSubject<String> = PublishSubject.create()

    var observableData: Observable<String>

    init {
        observableData = subject.flatMap {
            repo.fetchDataFromInfrastructure(it)
        }.replay().autoConnect()
    }

    fun requestData(nameOfRequester: String) {
        subject.onNext(nameOfRequester)
    }
}

class Repo {
    fun fetchDataFromInfrastructure(caller: String? = null): Observable<String> =
            Observable.fromIterable(listOf(1, 2, 3, 4, 5))
                    .map { number ->
                        "from caller: $caller ===> $number " + (0..100).random()
                    }
}


fun ClosedRange<Int>.random() =
        Random().nextInt((endInclusive + 1) - start) + start

sub1 ,sub2 and sub3 get the same result as it is cached using the replay() , helpful in some cases

autoConnect() maybe that 's not the best choice , I'd prefere to use autoConnect(1, addTo(disposible))