Split SubscribeOn into SubscribeOn/UnsubscribeOn

[benjchristensen]

Working with @headinthebox based on discussions at #869 and #880 (comment) we determined that there are times when unsubscribeOn behavior is needed.

The subscribeOn operator can not mix subscribe and unsubscribe scheduling behavior without breaking the lift/Subscriber behavior that allows unsubscribing synchronous sources. The newly added unsubscribeOn operator will not work with synchronous unsubscribes, but it will work for the targeted use cases such as UI event handlers.

[zsxwing]

I think this still can not guarantee that unsubscribe is always called in the scheduler. For example,

    public static void main(String[] args) throws InterruptedException {
        Observable.create(new OnSubscribe<Integer>() {

            @Override
            public void call(final Subscriber<? super Integer> t1) {
                final Subscription s = Subscriptions.create(new Action0() {

                    @Override
                    public void call() {
                        System.out.println(Thread.currentThread().getName());
                    }
                });
                t1.add(s);
                t1.onNext(1);
                t1.onNext(2);
                t1.onCompleted();
            }
        }).take(1).observeOn(Schedulers.io()).unsubscribeOn(Schedulers.newThread()).subscribe();
    }

will output main in my PC.

[zsxwing]

As you said, it can not work for synchronous unsubscribes, so we still need to write a thread-safe Subscription. I just feel unsubscribeOn will confuse users that when to use unsubscribeOn and when it's necessary to write a thread-safe Subscription.