Skip to content
Merged
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
7 changes: 7 additions & 0 deletions src/main/java/rx/Observable.java
Original file line number Diff line number Diff line change
Expand Up @@ -4362,6 +4362,10 @@ public final void onNext(T v) {
/**
* Modifies the source Observable so that it notifies an Observer for each item it emits.
* <p>
* In case the onError of the supplied observer throws, the downstream will receive a composite exception containing
* the original exception and the exception thrown by onError. If the onNext or the onCompleted methods
* of the supplied observer throws, the downstream will be terminated and wil receive this thrown exception.
* <p>
* <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/doOnEach.png" alt="">
* <dl>
* <dt><b>Scheduler:</b></dt>
Expand All @@ -4380,6 +4384,9 @@ public final Observable<T> doOnEach(Observer<? super T> observer) {
/**
* Modifies the source Observable so that it invokes an action if it calls {@code onError}.
* <p>
* In case the onError action throws, the downstream will receive a composite exception containing
* the original exception and the exception thrown by onError.
* <p>
* <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/doOnError.png" alt="">
* <dl>
* <dt><b>Scheduler:</b></dt>
Expand Down
7 changes: 5 additions & 2 deletions src/main/java/rx/internal/operators/OperatorDoOnEach.java
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,11 @@
*/
package rx.internal.operators;

import java.util.Arrays;

import rx.*;
import rx.Observable.Operator;
import rx.exceptions.Exceptions;
import rx.exceptions.*;

/**
* Converts the elements of an observable sequence to the specified type.
Expand Down Expand Up @@ -62,7 +64,8 @@ public void onError(Throwable e) {
try {
doOnEachObserver.onError(e);
} catch (Throwable e2) {
Exceptions.throwOrReport(e2, observer);
Exceptions.throwIfFatal(e2);
observer.onError(new CompositeException(Arrays.asList(e, e2)));
return;
}
observer.onError(e);
Expand Down
48 changes: 33 additions & 15 deletions src/test/java/rx/internal/operators/OperatorDoOnEachTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -17,25 +17,19 @@

import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.exceptions.OnErrorNotImplementedException;
import rx.functions.Action1;
import rx.functions.Func1;
import static org.mockito.Mockito.*;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.*;
import org.mockito.*;

import rx.*;
import rx.exceptions.*;
import rx.functions.*;
import rx.observers.TestSubscriber;

public class OperatorDoOnEachTest {

@Mock
Expand Down Expand Up @@ -201,4 +195,28 @@ public void call(Object o) {
System.out.println("Received exception: " + e);
}
}

@Test
public void testOnErrorThrows() {
TestSubscriber<Object> ts = TestSubscriber.create();

Observable.error(new TestException())
.doOnError(new Action1<Throwable>() {
@Override
public void call(Throwable e) {
throw new TestException();
}
}).subscribe(ts);

ts.assertNoValues();
ts.assertNotCompleted();
ts.assertError(CompositeException.class);

CompositeException ex = (CompositeException)ts.getOnErrorEvents().get(0);

List<Throwable> exceptions = ex.getExceptions();
assertEquals(2, exceptions.size());
assertTrue(exceptions.get(0) instanceof TestException);
assertTrue(exceptions.get(1) instanceof TestException);
}
}