RxObservable that only emits to first subscriber

ColdObservableSingleSubscriberProxy.kt

fun <T> Observable<T>.toColdSingleSubscribeable(): Observable<T> {
    return ColdObservableSingleSubscriberProxy(this)
}

class ColdObservableSingleSubscriberProxy<T>(private val observable: Observable<T>) : Observable<T>() {

    private val hasBeenSubscribed = AtomicBoolean(false)

    override fun subscribeActual(observer: Observer<in T>) {
        if (hasBeenSubscribed.compareAndSet(false, true)) {
            observable.subscribe(observer)
        } else {
            never<T>().subscribe(observer)
        }
    }

}

@Test
fun `assert observable fires only once ever`() {
    val testObservable = Observable.just(true).toColdSingleSubscribeable()

    assertFalse(testObservable.test().assertValue(true).isDisposed)

    assertFalse(testObservable.test().assertNoValues().assertEmpty().isDisposed)
}