diff --git a/rxjava-core/src/main/java/rx/operators/OperationDoOnEach.java b/rxjava-core/src/main/java/rx/operators/OperationDoOnEach.java
index acd841fc2f..1b0aafb578 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationDoOnEach.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationDoOnEach.java
@@ -24,23 +24,24 @@
  * Converts the elements of an observable sequence to the specified type.
  */
 public class OperationDoOnEach {
-    public static <T> OnSubscribeFunc<T> doOnEach(Observable<? extends T> source, Observer<? super T> observer) {
-        return new DoOnEachObservable<T>(source, observer);
+    public static <T> OnSubscribeFunc<T> doOnEach(Observable<? extends T> sequence, Observer<? super T> observer) {
+        return new DoOnEachObservable<T>(sequence, observer);
     }
 
     private static class DoOnEachObservable<T> implements OnSubscribeFunc<T> {
 
-        private final Observable<? extends T> source;
+        private final Observable<? extends T> sequence;
         private final Observer<? super T> doOnEachObserver;
 
-        public DoOnEachObservable(Observable<? extends T> source, Observer<? super T> doOnEachObserver) {
-            this.source = source;
+        public DoOnEachObservable(Observable<? extends T> sequence, Observer<? super T> doOnEachObserver) {
+            this.sequence = sequence;
             this.doOnEachObserver = doOnEachObserver;
         }
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            return source.subscribe(new Observer<T>() {
+            final SafeObservableSubscription subscription = new SafeObservableSubscription();
+            return subscription.wrap(sequence.subscribe(new SafeObserver<T>(subscription, new Observer<T>() {
                 @Override
                 public void onCompleted() {
                     doOnEachObserver.onCompleted();
@@ -58,8 +59,7 @@ public void onNext(T value) {
                     doOnEachObserver.onNext(value);
                     observer.onNext(value);
                 }
-
-            });
+            })));
         }
 
     }
diff --git a/rxjava-core/src/test/java/rx/operators/OperationDoOnEachTest.java b/rxjava-core/src/test/java/rx/operators/OperationDoOnEachTest.java
index 4bf017761c..6c1407ebea 100644
--- a/rxjava-core/src/test/java/rx/operators/OperationDoOnEachTest.java
+++ b/rxjava-core/src/test/java/rx/operators/OperationDoOnEachTest.java
@@ -36,6 +36,7 @@
 import rx.concurrency.Schedulers;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
+import rx.util.functions.Action1;
 
 public class OperationDoOnEachTest {
 
@@ -104,5 +105,25 @@ public String call(String s) {
         verify(sideEffectObserver, times(1)).onError(any(Throwable.class));
     }
 
+    @Test
+    public void testDoOnEachWithErrorInCallback() {
+        Observable<String> base     = Observable.from("one", "two", "fail", "three");
+        Observable<String> doOnEach = base.doOnEach(new Action1<String>() {
+            @Override
+            public void call(String s) {
+                if ("fail".equals(s)) {
+                    throw new RuntimeException("Forced Failure");
+                }
+            }
+        });
+
+        doOnEach.subscribe(subscribedObserver);
+        verify(subscribedObserver, times(1)).onNext("one");
+        verify(subscribedObserver, times(1)).onNext("two");
+        verify(subscribedObserver, never()).onNext("three");
+        verify(subscribedObserver, never()).onCompleted();
+        verify(subscribedObserver, times(1)).onError(any(Throwable.class));
+
+    }
   
 }