Subscriber.onStart is called twice on nested unsafeSubscription

[ibaca]

This test shows the onStart called twice problem. It is a clone of SubscriberTest.testOnStartCalledOnceViaUnsafeSubscribe but adding the defer operator which calls unsafeSubscribe resulting in a nested call to unsafeSubscribe, the one from the test and the second one from OnSubscribeDefer.

@Test
public void testOnStartCalledOnceViaNestedUnsafeSubscribe() {
    final AtomicInteger c = new AtomicInteger();
    Observable.defer(new Func0<Observable<Integer>>() {
        @Override public Observable<Integer> call() {
            return Observable.just(1, 2, 3, 4).take(2);
        }
    }).unsafeSubscribe(new Subscriber<Integer>() {
        @Override public void onStart() {
            c.incrementAndGet();
            request(1);
        }

        @Override public void onCompleted() { }

        @Override public void onError(Throwable e) { }

        @Override public void onNext(Integer t) {
            request(1);
        }
    });

    assertEquals(1, c.get());
}

Two proposed solutions (first one looks ugly, second one adds more code to Subscriber)

1. Add a parameter to subscription calls to transfer the 'onStart called' state.
2. Modify Subscriber so .subscribe() .unsafeSubscribe() calls Subscriber.start() which delegates to the actual .onStart() protecting for duplicate calls.

[akarnokd]

Thanks for discovering this.

With a little overhead, it is possible to suppress the second onStart call in defer by wrapping the subscriber with another subscriber that does nothing in its onStart. Would you like to submit a PR?

[akarnokd]

In fact, many other operators suffer from the same issue. You can discover them by altering Subscriber and running the standard tests:

final AtomicInteger once = new AtomicInteger();
// ...
public void onStart() {
    if (once.getAndIncrement() > 0) {
        throw new IllegalStateException();
    }
}

[ibaca]

Closing because has been solved by #2983. Thanks!