Why am I getting "inferred type is Observable<T?>! but Observable<T> was expected" with Observable Kotlin extension function?

Question

I have the following extension function:

import io.reactivex.Notification
import io.reactivex.Observable
import io.reactivex.subjects.Subject

fun <T> Subject<Notification<T>>.fromNotifications(): Observable<T> {
    return compose {
        it.takeUntil { event ->
            !event.isOnNext
        }.flatMap { event ->
            when {
                event.isOnError -> Observable.error(event.error)
                event.isOnNext -> Observable.just(event.value)
                else -> Observable.empty()
            }
        }
    }
}

I use it to convert a Subject with Notification values to ordinary Observable.

I don't quite understand why I'm getting the error

Type inference failed. Expected type mismatch: inferred type is Observable<T?>! but Observable<T> was expected

Where does inferred type is Observable<T?> come from? I guess it's related to the fact that in Java there's no difference between nullable and non-nullable types.

What are the ways to deal with this problem?

Answer 1

It's happening because event.value is marked as @Nullable, so Observable.just(event.value) is an Observable<T?>. You can make this compile by changing the return type to be Observable<T?>, or by using the !! operator to assert that event.value is not null when creating the Observable (or by otherwise dealing with a null value explicitly):

fun <T> Subject<Notification<T>>.fromNotifications(): Observable<T> {
    return compose {
        it.takeUntil { event ->
            !event.isOnNext
        }.flatMap { event ->
            when {
                event.isOnError -> Observable.error(event.error)
                event.isOnNext -> Observable.just(event.value!!)
                else -> Observable.empty()
            }
        }
    }
}