sample_of_returning_complete_in_flatmap.kt

val upstream = PublishSubject.create<Int>()
upstream.doOnDispose { Timber.d("Uh, the upstream gets disposed") } // This log will never print.

val tester = upstream
    .flatMap { v ->
        when (v) {
            0 -> {
                Timber.d("flatMap produces an empty")
                Observable.empty<Int>()
            }
            else -> Observable.just(v)
        }
    }
    .test()

upstream.onNext(0) // Will return a complete in flatMap.
upstream.onNext(1) // If the stream dies, we'll never emit this value successfully.

tester.assertValueCount(1) // Actually, the stream is still alive because flatMap ignores the complete!