OperatorGroupBy does not accept multiple subscriptions when source async

[davidmoten]

The test below fails in 0.16.1. I have not tested against new versions but I suggest we should if the test is valid. Please have a look, thanks.

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;

import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.mockito.Matchers;

import rx.Observable;
import rx.Observer;
import rx.observables.GroupedObservable;
import rx.util.functions.Func1;

public class OperatorGroupByTest2 {

    @Test
    public void testGroupByOnAsynchronousSourceAcceptsMultipleSubscriptions()
            throws InterruptedException {

        // choose an asynchronous source
        Observable<Long> source = Observable
                .interval(10, TimeUnit.MILLISECONDS).take(1);

        // apply groupBy to the source
        Observable<GroupedObservable<Boolean, Long>> stream = source
                .groupBy(IS_EVEN);

        // create two observers
        Observer<GroupedObservable<Boolean, Long>> o1 = mock(Observer.class);
        Observer<GroupedObservable<Boolean, Long>> o2 = mock(Observer.class);

        // subscribe with the observers
        stream.subscribe(o1);
        stream.subscribe(o2);

        // check that subscriptions were successful
        verify(o1, never()).onError(Matchers.<Throwable> any());
        verify(o2, never()).onError(Matchers.<Throwable> any());
    }

    private static Func1<Long, Boolean> IS_EVEN = new Func1<Long, Boolean>() {

        @Override
        public Boolean call(Long n) {
            return n % 2 == 0;
        }
    };

}

[benjchristensen]

I merged the unit test into master branch and it is passing.

[benjchristensen]

Thank you by the way for submitting a unit test along with the report. That makes it far easier.