Added the switch operator to Observable

[michaeldejong]

I've implemented the switch operator for Observable sequences (see issue #13). Because "switch" is a keyword in Java I have opted for switchDo. This matches to the finallyDo method which represents the finally operator and has the same naming problem.

Please let me know if you have any feedback.

[cloudbees-pull-request-builder]

RxJava-pull-requests #126 SUCCESS
This pull request looks good

[benjchristensen]

We don't need to synchronize as we expect the sequence to comply with the Rx contract. This adds performance overhead that we don't want to add.

See Rx Design Guidelines 6.8 for more background on this.

[michaeldejong]

Thanks for the guidelines. That does make more sense in terms of performance.

[cloudbees-pull-request-builder]

RxJava-pull-requests #128 SUCCESS
This pull request looks good

[benjchristensen]

Since the onError discussion is now part of an outdated diff ... moving it here:

---

I believe it needs to be this:

the error should be propagated and all future sequences produced by the parent sequence should be ignored (unsubscribed)

It's the same way with merge, zip, etc. If an error occurs on any sequence it immediately kills the entire sequence and passes the error.

I could see an overload of switch existing that could add behavior such as onErrorResumeNext so that the user can choose whether to ignore the error and just switch to the next sequence, or handle it with a default or something else before switching, etc.

We can't just swallow errors though so if an onError is received from a child sequence it must behave as if the parent received an onError and unsubscribe and pass on the error.

[michaeldejong]

I've modified the OperationSwitch class to handle errors in subsequences. However I'm not entirely convinced yet that unsubscribing from the parent sequence will always work. Example: If right after subscribing to the parent sequence, an error is thrown in a subsequence I'll try to unsubscribe from the parent sequence, but it's not guaranteed that the parent variable containing the Subscription to the parent sequence has already been set.

Ideas, thoughts?

[cloudbees-pull-request-builder]

RxJava-pull-requests #146 SUCCESS
This pull request looks good

[cloudbees-pull-request-builder]

RxJava-pull-requests #147 SUCCESS
This pull request looks good

[benjchristensen]

That's a good point ... there could be a race condition (very unlikely but possible).

The only way I can think of checking it is keeping a boolean flag that we're supposed to be finished and if we receive more onNext/onError notifications we continue to unsubscribe each time (and not send those events through).

[benjchristensen]

I'll make that change as I do some other cleanup before 0.9 is released.

[benjchristensen]

Thank you @michaeldejong for this submission.

[benjchristensen]

Actually AtomicObservableSubscription already handles this case in the wrap method:

    public AtomicObservableSubscription wrap(Subscription actualSubscription) {
        if (!this.actualSubscription.compareAndSet(null, actualSubscription)) {
            if (this.actualSubscription.get() == UNSUBSCRIBED) {
                actualSubscription.unsubscribe();
                return this;
            }
            throw new IllegalStateException("Can not set subscription more than once.");
        }
        return this;
    }

If the wrap occurs after an unsubscribe occurs it will immediately unsubscribe.