Hangs from Unhandled Errors when Scheduling

[benjchristensen]

I found a use case where the unsafeSubscribe combined with Scheduler since 0.17.5 can cause a hang in certain cases because a terminal event (onError in this case) never gets emitted.

Here is an example:

                inner.schedule(new Action0() {

                    @Override
                    public void call() {
                        o.unsafeSubscribe(subscriber);
                    }
                });

In this case, if the Observable being subscribed to throws an exception (breaks the contract) instead of using onError the Exception is completely swallowed because it's inside a Scheduler on another thread.

Here's a unit test demonstrating it:

    @Test
    public void testThrownErrorHandling() {
        TestSubscriber<String> ts = new TestSubscriber<String>();
        Observable.create(new OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> s) {
                throw new RuntimeException("fail");
            }

        }).subscribeOn(Schedulers.computation()).subscribe(ts);
        ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
        ts.assertTerminalEvent();
    }

Because it's inside a Scheduler the catch-all inside lift does not help this case.

There are two options as I see it:

1. Make every use of unsafeSubscribe inside a Scheduler try/catch/onError
2. Make unsafeSubscribe just a tad safer for the broken contract case

Option 1 is not great as all it takes is one mistaken place and we end up with code that hangs without any clue as to why.

Option 2 is simple, no cost and solves it everywhere. It does not defeat the reason for unsafeSubscribe as this will not involve SafeSubscriber, plugin invocation or anything else that should only happen with subscribe.

The code will change from:

    public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
        onSubscribe.call(subscriber);
        return subscriber;
    }

to

    public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
        try {
            onSubscribe.call(subscriber);
        } catch (Throwable e) {
            // handle broken contracts
            subscriber.onError(e);
        }
        return subscriber;
    }