RxJava Observable.create not emitting events for switchIfEmpty(observable)

Question

I was experimenting with RxJava operators and came across this issue and don't know why it's behaving the way it is. Let me walk through an example and hopefully the problem will be clear.

I have a fruits Observable which I created using Observable.just

Observable<String> fruits = Observable.just("Apple", "Banana", "Mango", "Strawberry", "Raspberry", "Blood Oranges");

I have another languages observable which I created using Observable.create

Observable<String> pLanguages = Observable.create(emitter -> {
            emitter.onNext("Kotlin");
            emitter.onNext("Java");
            emitter.onNext("Python");
            emitter.onNext("Javascript");
            emitter.onNext("Go");
            emitter.onNext("C");
            emitter.onNext("Rust");
        });

The switchIfEmpty() operator will subscribe to the operator which we pass to it if the source operator doesn't emit any observables.

Using this filter, I should see the fruits being emitted based on the definition of switchIfEmpty().

pLanguages.filter(s -> s.startsWith("Z"))
                .switchIfEmpty(fruits)
                .subscribe(System.out::println, System.out::println, () -> System.out.println("Complete"));

But I don't see any emissions. However, if I switch the observables like so

fruits.filter(s -> s.startsWith("Z"))
                .switchIfEmpty(pLanguages)
                .subscribe(System.out::println, System.out::println, () -> System.out.println("Complete"));

I do see the pLanguages observer emitting events. I've tried to debug but not sure why the Observable.create() is not emitting an empty observable for the switchIfEmpty() operator while Observable.just() does. I've spent couple of hours on this and hoping someone here can answer the question.

Joshua de Guzman Joshua de Guzman · Accepted Answer · 2019-01-21T06:45:47

Summary

You need to signal a completion on the languages source

Example (based on the provided source code)

Observable<String> pLanguages = Observable.create(emitter -> {
    emitter.onNext("Kotlin");
    emitter.onNext("Java");
    emitter.onNext("Python");
    emitter.onNext("Javascript");
    emitter.onNext("Go");
    emitter.onNext("C");
    emitter.onNext("Rust");
    emitter.onComplete();
});

Example in Kotlin (addt'l)

fun main(args: Array<String>) {
    val numbers = Observable.just("one", "two", "three")
    val languages = Observable.create<String> { emitter ->
        emitter.onNext("java")
        emitter.onNext("kotlin")
        emitter.onNext("swift")
        emitter.onComplete() // <-- Invoke this
    }

    languages
            .filter { s -> s.toLowerCase().startsWith("z") }
            .switchIfEmpty(numbers)
            .subscribe({
                System.out.println(it)
            }, {
                System.out.println(it)
            })

    numbers
            .filter { s -> s.toLowerCase().startsWith("z") }
            .switchIfEmpty(languages)
            .subscribe({
                System.out.println(it)
            }, {
                System.out.println(it)
            })


}

Output

one
two
three
java
kotlin
swift

Read more on

http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Emitter.html

RxJava Observable.create not emitting events for switchIfEmpty(observable)

1 Answers