Rxjava: 1.x async Singles crashes for Exceptions thrown in onSuccess

[passsy]

Since 1.2.2 tested with 1.2.9

Single doesn't wrap the SingleSubscriber automatically in a SafeSubscriber. This leads to problems when an Exceptions is thrown on onSuccess. It will not be propagated to onError for async Singles.

From the changelog 1.2.2 I can see this has performance reasons. But the problem is that async Singles now behave differently than sync Singles, or Observables sync or async.

// sync (as expected)
Single.just("test").subscribe(
        result -> {
            System.out.println("got text: " + result);
            throw new IllegalStateException("single sync something is wrong");
            // calls on Error
        },
        e -> {
            System.out.println("caught error " + e.getMessage());
            e.printStackTrace();
        });

// async (behaves differently)
Single.just("test")
        .delay(1, TimeUnit.MILLISECONDS)
        .subscribe(
        result -> {
            System.out.println("got text: " + result);
            throw new IllegalStateException("single async something is wrong");
            // onError will not be called, hard crash
        },
        e -> {
            // not called
            System.out.println("caught error " + e.getMessage());
            e.printStackTrace();
        });

// sync (as expected)
Observable.just("test").subscribe(
        result -> {
            System.out.println("got text: " + result);
            throw new IllegalStateException("observable sync something is wrong");
            // calls on Error
        },
        e -> {
            System.out.println("caught error " + e.getMessage());
            e.printStackTrace();
        });


// async (as expected)
Observable.just("test")
        .delay(1, TimeUnit.MILLISECONDS)
        .subscribe(
                result -> {
                    System.out.println("got text: " + result);
                    throw new IllegalStateException("observable async something is wrong");
                    // calls on Error
                },
                e -> {
                    System.out.println("caught error " + e.getMessage());
                    e.printStackTrace();
                });

---

After thinking about this case 1 is the one which is buggy. The synchronous Single should crash like the async Single case. Analog to the implementation in rx2

[akarnokd]

Looks like the lambda-subscriber on Single.java:1777 needs some catch clauses. Would you like to submit a PR?

[passsy]

I just compared this to the rx2 implementation. Both Single cases (sync/async) crash there. I'd like to know what's the correct behaviour. The documentation states:

A Single is something like an Observable, but instead of emitting a series of values — anywhere from none at all to an infinite number — it always either emits one value or an error notification.

A Single will call only one of these methods, and will only call it once. Upon calling either method, the Single terminates and the subscription to it ends.

Is it ok to call onError when onSuccess was already called but fails? From an user perspective I want onError to be triggered but the documentation should be adjusted as well.

[akarnokd]

No. The protocol states onSuccess | onError. If your onSuccess fails, it means your stream requires a doOnSuccess whose lambda can fail and you get an onError in the final consumer.

[passsy]

When implementing the bugfix I think that a SafeSingleSubscriber wrapper would be the best solution. The same problem also happens when using subscribe(Observer).

Can you give me insights why a SafeSingleSubscriber doesn't exists and why unsubscribe() has to be called manually by the user as mentioned here? My implementation of SafeSingleSubscriber automatically calls unsubscribe() and fails the SingleTest#isUnsubscribedAfterSuccess test.

[akarnokd]

Can you give me insights why a SafeSingleSubscriber doesn't exists

Because nobody has written it.

why unsubscribe() has to be called manually

SingleSubscriber is an abstract class which preserves its abstractness over onSuccess and onError. In order to call unsubscribe somebody has to hijack these calls either via a wrapper or defining another set of abstract methods as replacements. Neither option was implemented.

[TWiStErRob]

@passsy @akarnokd so there's no followup for this yet? It's strange that not all onSuccess/onNext exceptions consistently go to onError.

[aballano]

@TWiStErRob

The protocol states onSuccess | onError. If your onSuccess fails, it means your stream requires a doOnSuccess whose lambda can fail and you get an onError in the final consumer.

Which kind of makes sense since onSuccess | onError are terminal events and should be treated equally, otherwise an exception in onSuccess would call onError but one in onError would crash.

[TWiStErRob]

Ah, I see onNext ... onComplete | onError vs onSuccess | onError, so onSucccess has the semantics of both onNext and onComplete, leading to tricky scenarios where the two semantics conflict, namely being called exclusively to onError or not.

[akarnokd]

Closing this as won't fix. We discussed in #5710 that changing this would have to prevent synchronous exceptions being thrown that could be relied upon by existing code.