RxJava2 Convert Flowable to Single

11
votes

How do I convert a Flowable to a Single? Or if there's another way to make it stop emitting after the first response that is also of interest.

I've tried this but it doesn't seem to work:

  disposables.add(
        repository.getAllSomethings()
           .subscribeOn(SchedulerProvider.getInstance().computation())
           .observeOn(SchedulerProvider.getInstance().ui())
           .toSingle()
           .subscribeWith(object : DisposableSingleObserver<List<Something>>() {
                override fun onSuccess(t: List<Something>) {
                }

                override fun onError(e: Throwable) {
                }
            })

getAllSomethings() returns a Flowable<List<Something>>

In the above code .subscribeWith() is underlined in red complaining that: 

Type parameter bound for E in 
fun <E : SingleObserver<in Flowable<List<Something>!>!>!> subscribeWith(observer: E!): E!
is not satisfied: inferred type  ! is not a subtype of SingleObserver<in Flowable<List<Something>!>!>!
2
androidrx-java2
I suggest you familiarize yourself with the available operators - which will save you a lot of time. - akarnokd
ok so I guess you're saying I should use single(T defaultItem)? But I don't want to return a defaultItem. It should emit a single value or timeout and call the onError. - Michael Vescovo
Keep looking at the available operators. - akarnokd
ok I got it. or at least it seems to work with firstOrError(). thanks for your advice! do you want to answer it or should I do it? I actually remember you from a previous post and you wanted me to do the answer as you didn't care about points. - Michael Vescovo
yes I found it but with your help (and you knew the answer). I felt obliged to ask, however I'll take this to mean that if you ever answer like that again that this is what you want. Thanks again. - Michael Vescovo

2 Answers

27
votes

Ok so thanks to @akarnokd I found the answer as you can see in the comments.

Teaching me how to fish rather than giving me the straight answer he suggested looking here: http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html (which really I should have done in the first place!).

Looking there I found firstOrError() which solves my problem.

Even though the object I was calling toSingle() from was a Flowable, the IDE didn't complain. Yet looking at the link above, toSingle() isn't even a valid option!

0
votes

In my case, I actually had a Flowable and wanted the benefits of that (i.e. backpressure), but I still wanted to return a Single, e.g. Single<List<String>>. In my case, I was using Android WorkManager's RxWorker which expects my function to return Single<Result>, not a Flowable.

In this case,

val flowable = Flowable.just("my flowable")
return flowable.toList() // type: Single<List<String>>

or if you wanted to return a specific value because you don't care about the flowable output:

return flowable.toList().map {} // returns unit
return flowable.toList().map { "Hello" } // returns String
return flowable.toList().map { Result.success() } // returns Result