Skip to content

Add Single.onErrorResumeNext(Func1<Throwable, Single>) #3772

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions src/main/java/rx/Single.java
Original file line number Diff line number Diff line change
Expand Up @@ -1435,6 +1435,39 @@ public final Single<T> onErrorResumeNext(Single<? extends T> resumeSingleInCaseO
return new Single<T>(new SingleOperatorOnErrorResumeNextViaSingle<T>(this, resumeSingleInCaseOfError));
}

/**
* Instructs a Single to pass control to another Single rather than invoking {@link Observer#onError}
* if it encounters an error.
* <p/>
* <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/onErrorResumeNext.png" alt="">
* <p/>
* By default, when a Single encounters an error that prevents it from emitting the expected item to its
* {@link Observer}, the Single invokes its Observer's {@code onError} method, and then quits without
* invoking any more of its Observer's methods. The {@code onErrorResumeNext} method changes this
* behavior. If you pass a function that returns a Single ({@code resumeFunction}) to a Single's
* {@code onErrorResumeNext} method, if the original Single encounters an error, instead of invoking its
* Observer's {@code onError} method, it will instead relinquish control to the Single returned from
* {@code resumeFunction}, which will invoke the Observer's {@link Observer#onNext onNext} method if it
* is able to do so. In such a case, because no Single necessarily invokes {@code onError}, the Observer
* may never know that an error happened.
* <p/>
* You can use this to prevent errors from propagating or to supply fallback data should errors be
* encountered.
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code onErrorResumeNext} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
*
* @param resumeFunction
* a function that returns a Single that will take control if the source Single encounters an
* error
* @return the original Single, with appropriately modified behavior
* @see <a href="http://reactivex.io/documentation/operators/catch.html">ReactiveX operators documentation: Catch</a>
*/
public final Single<T> onErrorResumeNext(final Func1<Throwable, ? extends Single<? extends T>> resumeFunction) {
return new Single<T>(new SingleOperatorOnErrorResumeNextViaFunction<T>(this, resumeFunction));
}

/**
* Subscribes to a Single but ignore its emission or notification.
* <dl>
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
package rx.internal.operators;

import rx.Single;
import rx.SingleSubscriber;
import rx.exceptions.Exceptions;
import rx.functions.Func1;

public class SingleOperatorOnErrorResumeNextViaFunction<T> implements Single.OnSubscribe<T> {

private final Single<? extends T> originalSingle;
private final Func1<Throwable, ? extends Single<? extends T>> resumeFunction;

public SingleOperatorOnErrorResumeNextViaFunction(Single<? extends T> originalSingle, Func1<Throwable, ? extends Single<? extends T>> resumeFunction) {
if (originalSingle == null) {
throw new NullPointerException("originalSingle must not be null");
}

if (resumeFunction == null) {
throw new NullPointerException("resumeFunction must not be null");
}

this.originalSingle = originalSingle;
this.resumeFunction = resumeFunction;
}

@Override
public void call(final SingleSubscriber<? super T> child) {
final SingleSubscriber<? super T> parent = new SingleSubscriber<T>() {
@Override
public void onSuccess(T value) {
child.onSuccess(value);
}

@Override
public void onError(Throwable error) {
try {
unsubscribe();
resumeFunction.call(error).subscribe(child);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
child.onError(e);
}
}
};

child.add(parent);
originalSingle.subscribe(parent);
}
}
66 changes: 65 additions & 1 deletion src/test/java/rx/SingleTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -1220,13 +1220,77 @@ public void onErrorResumeNextViaSingleShouldPreventNullSingle() {
try {
Single
.just("value")
.onErrorResumeNext(null);
.onErrorResumeNext((Single<String>)null);
fail();
} catch (NullPointerException expected) {
assertEquals("resumeSingleInCaseOfError must not be null", expected.getMessage());
}
}

@Test
public void onErrorResumeNextViaFunctionShouldNotInterruptSuccessfulSingle() {
TestSubscriber<String> testSubscriber = new TestSubscriber<String>();

Single
.just("success")
.onErrorResumeNext(new Func1<Throwable, Single<String>>() {
@Override
public Single<String> call(Throwable t) {
return Single.just("fail");
}
})
.subscribe(testSubscriber);

testSubscriber.assertValue("success");
}

@Test
public void onErrorResumeNextViaFunctionShouldResumeWithPassedSingleInCaseOfError() {
TestSubscriber<String> testSubscriber = new TestSubscriber<String>();

Single
.<String> error(new RuntimeException("test exception"))
.onErrorResumeNext(new Func1<Throwable, Single<String>>() {
@Override
public Single<String> call(Throwable t) {
return Single.just("fallback");
}
})
.subscribe(testSubscriber);

testSubscriber.assertValue("fallback");
}

@Test
public void onErrorResumeNextViaFunctionShouldCatchExceptionThrownByFunction() {
TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
final RuntimeException expected = new RuntimeException();

Single
.<String> error(new RuntimeException("test exception"))
.onErrorResumeNext(new Func1<Throwable, Single<String>>() {
@Override
public Single<String> call(Throwable t) {
throw expected;
}
})
.subscribe(testSubscriber);

testSubscriber.assertError(expected);
}

@Test
public void onErrorResumeNextViaFunctionShouldPreventNullSingle() {
try {
Single
.just("value")
.onErrorResumeNext((Func1<Throwable, Single<String>>)null);
fail();
} catch (NullPointerException expected) {
assertEquals("resumeFunction must not be null", expected.getMessage());
}
}

@Test(expected = NullPointerException.class)
public void iterableToArrayShouldThrowNullPointerExceptionIfIterableNull() {
Single.iterableToArray(null);
Expand Down