mapNotNull operators for Rx Observable/Single/Maybe.

Rx+mapNotNull.kt

fun <T, R> Observable<T>.mapNotNull(mapper: (T) -> R?): Observable<R> {
    return lift { observer ->
        object: Observer<T> {
            override fun onNext(t: T) {
                mapper(t)?.let { observer.onNext(it) }
            }
            override fun onComplete() {
                observer.onComplete()
            }
            override fun onError(e: Throwable) {
                observer.onError(e)
            }
            override fun onSubscribe(d: Disposable) {
                observer.onSubscribe(d)
            }
        }
    }
}

fun <T, R> Single<T>.mapNotNull(mapper: (T) -> R?): Maybe<R> {
    return toMaybe().mapNotNull(mapper)
}

fun <T, R> Maybe<T>.mapNotNull(mapper: (T) -> R?): Maybe<R> {
    return lift { observer ->
        object: MaybeObserver<T> {
            override fun onSuccess(t: T) {
                mapper(t)?.let { observer.onSuccess(it) }
                        ?: run { observer.onComplete() }
            }
            override fun onComplete() {
                observer.onComplete()
            }
            override fun onError(e: Throwable) {
                observer.onError(e)
            }
            override fun onSubscribe(d: Disposable) {
                observer.onSubscribe(d)
            }
        }
    }
}