Implemented the Merge overloads

[zsxwing]

Hi, this PR implemented the overloads of merge in #62. Please take a look.

[cloudbees-pull-request-builder]

RxJava-pull-requests #635 FAILURE
Looks like there's a problem with this pull request

[cloudbees-pull-request-builder]

RxJava-pull-requests #636 FAILURE
Looks like there's a problem with this pull request

[zsxwing]

I checked the failure in the groovy unit tests. Here is the problem.

In the groovy unit test, there is an assumption. The following Observable

Observable.merge(
    Observable.from(6),                                                                                                                        
    Observable.error(new NullPointerException()),                                                                                              
    Observable.from(7)
)

should emit 6, and emit a NullPointerException later.

However, my implementation uses from(Iterable<? extends T> iterable) to create an Observable<? extends Observable<? extends T>>. The default scheduler is Schedulers.currentThread(), which make the previous assumption failed. As emitting 6 will be delayed, so the Observable will only emit a NullPointerException.

[zsxwing]

I suppose merge does not need to provide such order guarantee.

Thoughts?

[benjchristensen]

Why would 6 be delayed? CurrentThreadScheduler doesn't change the order. It should subscribe to them in order, particularly since there is no concurrency and thus no race conditions here?

[zsxwing]

I wanna use the following codes to discuss:

        Observable<Integer> o1 = Observable.from(6);
        Observable<Integer> o2 = Observable.<Integer> error(new NullPointerException());
        Observable<Integer> o3 = Observable.from(7);
        Observable<Integer> o = Observable.merge(o1, o2, o3);

In this example, when we subscribe o, the order is:

subscribe o (schedule (emit o1, emit o2, emit o3))
emit o1
subscribe o1 (schedule(emit 6))
emit o2
subscribe o2 (NullPointerException)
emit o3
subscribe o3 (schedule(emit 7))
emit 6
emit 7

The problem is that Observable.error(new NullPointerException()) throws the exception directly when it's subscribed. The behavior is like using Schedulers.immediate(). So actually, we can say that there are two types of Scheduler involved here.
If we wanna guarantee the order, Observable.error(new NullPointerException()) need to be changed to Observable.error(new NullPointerException(), Schedulers.currentThread()).

[zsxwing]

Just found using Schedulers.currentThread() in error still can't fix this problem.
For the nested merge, for example,

123        Observable.merge(                                                                                                                                  
124                Observable.from(1, 2, 3),                                                                                                                  
125                Observable.merge(                                                                                                                          
126                    Observable.from(6),                                                                                                                        
127                    Observable.error(new NullPointerException(), Schedulers.currentThread()),                                                                                              
128                    Observable.from(7)),                                                                                                                       
129                Observable.from(4, 5))                                                                                                                     
130                .subscribe({ result -> a.received(result)}, { exception -> a.error(exception)});  

The order is,

1
2
3
4
5
6
NullPointerException
7

Now I can only find two solutions,

• Do not promise any order.
• Use Schedulers.immediate() when using from(Iterable<? extends T> iterable) to create an Observable<? extends Observable<? extends T>>, just like the previous merge implementation.

[cloudbees-pull-request-builder]

RxJava-pull-requests #643 SUCCESS
This pull request looks good

[zsxwing]

Rebased. #724 helped pass the groovy tests.

[benjchristensen]

I'm holding off on this for now as I'm stabilizing 0.16.0 for release and am already in the midst of testing on Netflix production canaries. I'll review this for 0.16.1 once 0.16.0 is released.