Skip to content

Commit a4d56c8

Browse files
committed
Merge pull request #3765 from akarnokd/SwitchOnNextDelayError1
1.x: switchOnNextDelayError and switchMapDelayError
2 parents 044620e + 92a255c commit a4d56c8

File tree

3 files changed

+401
-166
lines changed

3 files changed

+401
-166
lines changed

src/main/java/rx/Observable.java

Lines changed: 54 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2881,7 +2881,36 @@ public static <T> Observable<Boolean> sequenceEqual(Observable<? extends T> firs
28812881
* @see <a href="http://reactivex.io/documentation/operators/switch.html">ReactiveX operators documentation: Switch</a>
28822882
*/
28832883
public static <T> Observable<T> switchOnNext(Observable<? extends Observable<? extends T>> sequenceOfSequences) {
2884-
return sequenceOfSequences.lift(OperatorSwitch.<T>instance());
2884+
return sequenceOfSequences.lift(OperatorSwitch.<T>instance(false));
2885+
}
2886+
2887+
/**
2888+
* Converts an Observable that emits Observables into an Observable that emits the items emitted by the
2889+
* most recently emitted of those Observables and delays any exception until all Observables terminate.
2890+
* <p>
2891+
* <img width="640" height="370" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchDo.png" alt="">
2892+
* <p>
2893+
* {@code switchOnNext} subscribes to an Observable that emits Observables. Each time it observes one of
2894+
* these emitted Observables, the Observable returned by {@code switchOnNext} begins emitting the items
2895+
* emitted by that Observable. When a new Observable is emitted, {@code switchOnNext} stops emitting items
2896+
* from the earlier-emitted Observable and begins emitting items from the new one.
2897+
* <dl>
2898+
* <dt><b>Scheduler:</b></dt>
2899+
* <dd>{@code switchOnNext} does not operate by default on a particular {@link Scheduler}.</dd>
2900+
* </dl>
2901+
*
2902+
* @param <T> the item type
2903+
* @param sequenceOfSequences
2904+
* the source Observable that emits Observables
2905+
* @return an Observable that emits the items emitted by the Observable most recently emitted by the source
2906+
* Observable
2907+
* @see <a href="http://reactivex.io/documentation/operators/switch.html">ReactiveX operators documentation: Switch</a>
2908+
* @Experimental The behavior of this can change at any time.
2909+
* @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
2910+
*/
2911+
@Experimental
2912+
public static <T> Observable<T> switchOnNextDelayError(Observable<? extends Observable<? extends T>> sequenceOfSequences) {
2913+
return sequenceOfSequences.lift(OperatorSwitch.<T>instance(true));
28852914
}
28862915

28872916
/**
@@ -8734,6 +8763,30 @@ public final <R> Observable<R> switchMap(Func1<? super T, ? extends Observable<?
87348763
return switchOnNext(map(func));
87358764
}
87368765

8766+
/**
8767+
* Returns a new Observable by applying a function that you supply to each item emitted by the source
8768+
* Observable that returns an Observable, and then emitting the items emitted by the most recently emitted
8769+
* of these Observables and delays any error until all Observables terminate.
8770+
* <p>
8771+
* <img width="640" height="350" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchMap.png" alt="">
8772+
* <dl>
8773+
* <dt><b>Scheduler:</b></dt>
8774+
* <dd>{@code switchMap} does not operate by default on a particular {@link Scheduler}.</dd>
8775+
* </dl>
8776+
*
8777+
* @param func
8778+
* a function that, when applied to an item emitted by the source Observable, returns an
8779+
* Observable
8780+
* @return an Observable that emits the items emitted by the Observable returned from applying {@code func} to the most recently emitted item emitted by the source Observable
8781+
* @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
8782+
* @Experimental The behavior of this can change at any time.
8783+
* @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
8784+
*/
8785+
@Experimental
8786+
public final <R> Observable<R> switchMapDelayError(Func1<? super T, ? extends Observable<? extends R>> func) {
8787+
return switchOnNextDelayError(map(func));
8788+
}
8789+
87378790
/**
87388791
* Returns an Observable that emits only the first {@code count} items emitted by the source Observable. If the source emits fewer than
87398792
* {@code count} items then all of its items are emitted.

0 commit comments

Comments
 (0)