Suggestions for Episode 3

[artem-zinnatullin]

Suggestions for Episode 3

[vanniktech]

First of all nice Topic, I really like that one. 👍

It still did not click on my side about those things even though I read some articles about it already so I'd appreciate it if he could try to explain it.

• What's the in depth and technical difference between subscribeOn() and observeOn()?
• How does calling them in different orders or even multiple times with different Schedulers affect the Observable?

And just for curiosity:

• What is his opinion about Single?

[artem-zinnatullin]

Thanks, @vanniktech, added!

[Lakedaemon]

With RxJava, what is the best way to model a task that returns a result with progress report (0-100%) ?
a) say in the case of one download/heavy computation ?
b) what about many parallel tasks (0-100% being the sum of those progress reports) ?

[akarnokd]

@Lakedaemon Your questions are best answered in written form:

a) If you are in control of the download/computation chunks and if you can determine the progress, then you may create an OnSubscribe implementation that emits a Pair<Integer, Data>. While the operation is in progress, you emit a Pair.of(x, null) and once it completes, you emit Pair.of(100, data). Since the progress tracking not really backpressureable, you can warn the users of your operator to apply onBackpressureXXX.

b) This is a job for combineLatest() where you can tuple-wise gather the progress amounts of multiple ongoing downloads and you can apply any mathematical transformation to a row of progress values: min, avg, max, sum (although 800% download progress looks strange). Since individual download Observables may singal its first value at any time, you may lose progress information because combineLatest() can't gather a full row of events. To avoid this, you can add startWith(Pair.of(0, null)) to begin with 0%:

Observable<Pair<Integer, Object>> download(String what) { ... }

Pair<Integer, Object> zero = Pair.of(0, null);

Observable.combineLatest(
    download("a").startWith(zero),
    download("b").startWith(zero),
    download("c").startWith(zero),
    (a, b, c) -> (a.first + b.first + c.first) / 3);

[akarnokd]

My only request is to ask questions that you think can be answered without the need to write code verbally. If you have such questions, please consider asking it on StackOverflow. There is a high chance, depending on the complexity, that I'll answer it there too, and you may not need to wait till the podcast goes out.

[dbacinski]

Testing:
How to test asynchronous code that includes Observables with jUnit?
How to use TestSubscriber?

[Lakedaemon]

Thanks for the answer and also, you are right, I should have asked this on stack overflow instead, this question wasn't a good one for The Context.

[artem-zinnatullin]

@dbacinski thanks, added (even though it's more about using a specific API there is a theoretical part in your question which we can discuss)

@Lakedaemon looks like David answered your question, my suggestion will be pretty much the same though I'd probably just emit progress as number and payload will be usually persisted on disk to prevent OOM, though Pair is ok approach too for many cases.

[sockeqwe]

I have a question (probably it's too advanced, so I wouldn't mind if you could not answer that during the podcast because it doesn't fit in)

What is the reason that I can only specify subscribeOn() one time for the whole observable stream?
Is there a technical reason? Is the implementation of multiple subscribeOn too hard? Or doesn't multiple subscribeOn() make sense at all because we can use observeOn() downwards (from my understanding the difference between subscribeOn and observeOn is that subscribeOn() is used when creating subscribers (internally) and observeOn() is used for propagating onNext() etc., right?)

I also think that the API is "inconsistent" here, since you can call .subscribeOn() multiple times, but only the first subscribeOn() has an effect. Wouldn't it make more sense to add to the Observable factories an additional (optional) parameter scheduler like this Observable.fromCallable( () -> ... , Schedulers.io() ) and removing .subscribeOn() from Observable. Are there any plans to do so in RxJava 2?

[artem-zinnatullin]

@sockeqwe good one, and I believe it's worth to discuss it in the podcast, I'll try to ask this during subscribeOn/observeOn discussion, thanks!

About overriding subscribeOn(): I'd say that at the moment it's technical reason due to historical reasons :)

[sockeqwe]

👍
maybe also one or two sentences about backpressure would be nice in subscribeOn / observeOn discussion, very high level like what is backpressure, how can it occure and what kind of strategies to resolve it are already there (the most common 1 or 2, just name them so people can google for them). Not more than two minutes ...

[Dmitry-Borodin]

What is the best way to test currect (exact) order of emitted results in JUnit?

[vanniktech]

@sockeqwe 👍 I'm also interested in the backpressure topic regarding subscribeOn / observeOn

@artem-zinnatullin thanks for also adding Completable to the curiosity question. I forgot about that one.

[artem-zinnatullin]

We've recorded the episode, answered most of these questions + dive into RxJava history, hope to release it in a few days!