diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index e332098ba2..95e3830979 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -27,6 +27,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Before;
@@ -41,9 +42,6 @@
 import rx.operators.OperationDefer;
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationFilter;
-import rx.operators.OperationTake;
-import rx.operators.OperationTakeWhile;
-import rx.operators.OperationWhere;
 import rx.operators.OperationMap;
 import rx.operators.OperationMaterialize;
 import rx.operators.OperationMerge;
@@ -56,17 +54,21 @@
 import rx.operators.OperationScan;
 import rx.operators.OperationSkip;
 import rx.operators.OperationSynchronize;
+import rx.operators.OperationTake;
 import rx.operators.OperationTakeLast;
+import rx.operators.OperationTakeWhile;
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationToObservableIterable;
 import rx.operators.OperationToObservableList;
 import rx.operators.OperationToObservableSortedList;
+import rx.operators.OperationWhere;
 import rx.operators.OperationZip;
 import rx.operators.OperatorGroupBy;
 import rx.operators.OperatorTakeUntil;
 import rx.operators.OperatorToIterator;
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaPlugins;
+import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.AtomicObservableSubscription;
 import rx.util.AtomicObserver;
@@ -79,6 +81,7 @@
 import rx.util.functions.Func3;
 import rx.util.functions.Func4;
 import rx.util.functions.FuncN;
+import rx.util.functions.Function;
 import rx.util.functions.FunctionLanguageAdaptor;
 import rx.util.functions.Functions;
 
@@ -99,10 +102,9 @@
 public class Observable<T> {
 
     private final Func1<Observer<T>, Subscription> onSubscribe;
-    private final boolean isTrusted;
 
     protected Observable() {
-        this(null, false);
+        this(null);
     }
 
     /**
@@ -114,18 +116,7 @@ protected Observable() {
      *            {@link Func1} to be executed when {@link #subscribe(Observer)} is called.
      */
     protected Observable(Func1<Observer<T>, Subscription> onSubscribe) {
-        this(onSubscribe, false);
-    }
-
-    /**
-     * @param onSubscribe
-     *            {@link Func1} to be executed when {@link #subscribe(Observer)} is called.
-     * @param isTrusted
-     *            boolean true if the <code>onSubscribe</code> function is guaranteed to conform to the correct contract and thus shortcuts can be taken.
-     */
-    private Observable(Func1<Observer<T>, Subscription> onSubscribe, boolean isTrusted) {
         this.onSubscribe = onSubscribe;
-        this.isTrusted = isTrusted;
     }
 
     /**
@@ -159,7 +150,10 @@ public Subscription subscribe(Observer<T> observer) {
             // the subscribe function can also be overridden but generally that's not the appropriate approach so I won't mention that in the exception
         }
         try {
-            if (isTrusted) {
+            /**
+             * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+             */
+            if (isInternalImplementation(observer)) {
                 Subscription s = onSubscribe.call(observer);
                 if (s == null) {
                     // this generally shouldn't be the case on a 'trusted' onSubscribe but in case it happens
@@ -185,6 +179,16 @@ public Subscription subscribe(Observer<T> observer) {
         }
     }
 
+    /**
+     * Used for protecting against errors being thrown from Observer implementations and ensuring onNext/onError/onCompleted contract compliance.
+     * <p>
+     * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+     */
+    private Subscription protectivelyWrapAndSubscribe(Observer<T> o) {
+        AtomicObservableSubscription subscription = new AtomicObservableSubscription();
+        return subscription.wrap(subscribe(new AtomicObserver<T>(subscription, o)));
+    }
+
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public Subscription subscribe(final Map<String, Object> callbacks) {
         // lookup and memoize onNext
@@ -194,7 +198,12 @@ public Subscription subscribe(final Map<String, Object> callbacks) {
         }
         final FuncN onNext = Functions.from(_onNext);
 
-        return subscribe(new Observer() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        return protectivelyWrapAndSubscribe(new Observer() {
 
             public void onCompleted() {
                 Object onComplete = callbacks.get("onCompleted");
@@ -231,7 +240,12 @@ public Subscription subscribe(final Object o) {
         }
         final FuncN onNext = Functions.from(o);
 
-        return subscribe(new Observer() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        return protectivelyWrapAndSubscribe(new Observer() {
 
             public void onCompleted() {
                 // do nothing
@@ -251,7 +265,12 @@ public void onNext(Object args) {
 
     public Subscription subscribe(final Action1<T> onNext) {
 
-        return subscribe(new Observer<T>() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        return protectivelyWrapAndSubscribe(new Observer<T>() {
 
             public void onCompleted() {
                 // do nothing
@@ -280,7 +299,12 @@ public Subscription subscribe(final Object onNext, final Object onError) {
         }
         final FuncN onNextFunction = Functions.from(onNext);
 
-        return subscribe(new Observer() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        return protectivelyWrapAndSubscribe(new Observer() {
 
             public void onCompleted() {
                 // do nothing
@@ -302,7 +326,12 @@ public void onNext(Object args) {
 
     public Subscription subscribe(final Action1<T> onNext, final Action1<Exception> onError) {
 
-        return subscribe(new Observer<T>() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        return protectivelyWrapAndSubscribe(new Observer<T>() {
 
             public void onCompleted() {
                 // do nothing
@@ -333,7 +362,12 @@ public Subscription subscribe(final Object onNext, final Object onError, final O
         }
         final FuncN onNextFunction = Functions.from(onNext);
 
-        return subscribe(new Observer() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        return protectivelyWrapAndSubscribe(new Observer() {
 
             public void onCompleted() {
                 if (onComplete != null) {
@@ -357,7 +391,12 @@ public void onNext(Object args) {
 
     public Subscription subscribe(final Action1<T> onNext, final Action1<Exception> onError, final Action0 onComplete) {
 
-        return subscribe(new Observer<T>() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        return protectivelyWrapAndSubscribe(new Observer<T>() {
 
             public void onCompleted() {
                 onComplete.call();
@@ -396,7 +435,12 @@ public void forEach(final Action1<T> onNext) {
         final CountDownLatch latch = new CountDownLatch(1);
         final AtomicReference<Exception> exceptionFromOnError = new AtomicReference<Exception>();
 
-        subscribe(new Observer<T>() {
+        /**
+         * Wrapping since raw functions provided by the user are being invoked.
+         * 
+         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         */
+        protectivelyWrapAndSubscribe(new Observer<T>() {
             public void onCompleted() {
                 latch.countDown();
             }
@@ -565,7 +609,7 @@ public Subscription call(Observer<T> t1) {
                     return Subscriptions.empty();
                 }
 
-            }, true);
+            });
         }
     }
 
@@ -593,7 +637,7 @@ public Subscription call(Observer<T> observer) {
                     return Subscriptions.empty();
                 }
 
-            }, true);
+            });
         }
 
     }
@@ -619,13 +663,6 @@ public static <T> Observable<T> create(Func1<Observer<T>, Subscription> func) {
         return new Observable<T>(func);
     }
 
-    /*
-     * Private version that creates a 'trusted' Observable to allow performance optimizations.
-     */
-    private static <T> Observable<T> _create(Func1<Observer<T>, Subscription> func) {
-        return new Observable<T>(func, true);
-    }
-
     /**
      * Creates an Observable that will execute the given function when a {@link Observer} subscribes to it.
      * <p>
@@ -697,7 +734,7 @@ public static <T> Observable<T> error(Exception exception) {
      * @return an Observable that emits only those items in the original Observable that the filter evaluates as true
      */
     public static <T> Observable<T> filter(Observable<T> that, Func1<T, Boolean> predicate) {
-        return _create(OperationFilter.filter(that, predicate));
+        return create(OperationFilter.filter(that, predicate));
     }
 
     /**
@@ -729,7 +766,7 @@ public Boolean call(T t1) {
      * Filters an Observable by discarding any of its emissions that do not meet some test.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/filter.png">
-     *
+     * 
      * @param that
      *            the Observable to filter
      * @param predicate
@@ -737,7 +774,7 @@ public Boolean call(T t1) {
      * @return an Observable that emits only those items in the original Observable that the filter evaluates as true
      */
     public static <T> Observable<T> where(Observable<T> that, Func1<T, Boolean> predicate) {
-        return _create(OperationWhere.where(that, predicate));
+        return create(OperationWhere.where(that, predicate));
     }
 
     /**
@@ -797,7 +834,7 @@ public static Observable<Integer> range(int start, int count) {
      * @return the observable sequence whose observers trigger an invocation of the given observable factory function.
      */
     public static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {
-        return _create(OperationDefer.defer(observableFactory));
+        return create(OperationDefer.defer(observableFactory));
     }
 
     /**
@@ -816,7 +853,7 @@ public static <T> Observable<T> defer(Object observableFactory) {
         @SuppressWarnings("rawtypes")
         final FuncN _f = Functions.from(observableFactory);
 
-        return _create(OperationDefer.defer(new Func0<Observable<T>>() {
+        return create(OperationDefer.defer(new Func0<Observable<T>>() {
 
             @Override
             @SuppressWarnings("unchecked")
@@ -980,7 +1017,7 @@ public Boolean call(T args) {
      *         in the sequence emitted by the source Observable
      */
     public static <T, R> Observable<R> map(Observable<T> sequence, Func1<T, R> func) {
-        return _create(OperationMap.map(sequence, func));
+        return create(OperationMap.map(sequence, func));
     }
 
     /**
@@ -1036,7 +1073,7 @@ public R call(T t1) {
      *         the Observables obtained from this transformation
      */
     public static <T, R> Observable<R> mapMany(Observable<T> sequence, Func1<T, Observable<R>> func) {
-        return _create(OperationMap.mapMany(sequence, func));
+        return create(OperationMap.mapMany(sequence, func));
     }
 
     /**
@@ -1085,7 +1122,7 @@ public R call(T t1) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229453(v=VS.103).aspx">MSDN: Observable.Materialize</a>
      */
     public static <T> Observable<Notification<T>> materialize(final Observable<T> sequence) {
-        return _create(OperationMaterialize.materialize(sequence));
+        return create(OperationMaterialize.materialize(sequence));
     }
 
     /**
@@ -1097,7 +1134,7 @@ public static <T> Observable<Notification<T>> materialize(final Observable<T> se
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229047(v=vs.103).aspx">MSDN: Observable.Dematerialize</a>
      */
     public static <T> Observable<T> dematerialize(final Observable<Notification<T>> sequence) {
-        return _create(OperationDematerialize.dematerialize(sequence));
+        return create(OperationDematerialize.dematerialize(sequence));
     }
 
     /**
@@ -1114,7 +1151,7 @@ public static <T> Observable<T> dematerialize(final Observable<Notification<T>>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge</a>
      */
     public static <T> Observable<T> merge(List<Observable<T>> source) {
-        return _create(OperationMerge.merge(source));
+        return create(OperationMerge.merge(source));
     }
 
     /**
@@ -1131,7 +1168,7 @@ public static <T> Observable<T> merge(List<Observable<T>> source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> merge(Observable<Observable<T>> source) {
-        return _create(OperationMerge.merge(source));
+        return create(OperationMerge.merge(source));
     }
 
     /**
@@ -1148,7 +1185,7 @@ public static <T> Observable<T> merge(Observable<Observable<T>> source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> merge(Observable<T>... source) {
-        return _create(OperationMerge.merge(source));
+        return create(OperationMerge.merge(source));
     }
 
     /**
@@ -1181,7 +1218,7 @@ public static <T, E> Observable<T> takeUntil(final Observable<T> source, final O
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat(v=vs.103).aspx">MSDN: Observable.Concat Method</a>
      */
     public static <T> Observable<T> concat(Observable<T>... source) {
-        return _create(OperationConcat.concat(source));
+        return create(OperationConcat.concat(source));
     }
 
     /**
@@ -1202,7 +1239,7 @@ public static <T> Observable<T> concat(Observable<T>... source) {
      * @return an observable of observable groups, each of which corresponds to a unique key value, containing all elements that share that same key value.
      */
     public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T> source, final Func1<T, K> keySelector, final Func1<T, R> elementSelector) {
-        return _create(OperatorGroupBy.groupBy(source, keySelector, elementSelector));
+        return create(OperatorGroupBy.groupBy(source, keySelector, elementSelector));
     }
 
     /**
@@ -1219,7 +1256,7 @@ public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T
      * @return an observable of observable groups, each of which corresponds to a unique key value, containing all elements that share that same key value.
      */
     public static <K, T> Observable<GroupedObservable<K, T>> groupBy(Observable<T> source, final Func1<T, K> keySelector) {
-        return _create(OperatorGroupBy.groupBy(source, keySelector));
+        return create(OperatorGroupBy.groupBy(source, keySelector));
     }
 
     /**
@@ -1238,7 +1275,7 @@ public static <K, T> Observable<GroupedObservable<K, T>> groupBy(Observable<T> s
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> mergeDelayError(List<Observable<T>> source) {
-        return _create(OperationMergeDelayError.mergeDelayError(source));
+        return create(OperationMergeDelayError.mergeDelayError(source));
     }
 
     /**
@@ -1257,7 +1294,7 @@ public static <T> Observable<T> mergeDelayError(List<Observable<T>> source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> mergeDelayError(Observable<Observable<T>> source) {
-        return _create(OperationMergeDelayError.mergeDelayError(source));
+        return create(OperationMergeDelayError.mergeDelayError(source));
     }
 
     /**
@@ -1276,7 +1313,7 @@ public static <T> Observable<T> mergeDelayError(Observable<Observable<T>> source
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> mergeDelayError(Observable<T>... source) {
-        return _create(OperationMergeDelayError.mergeDelayError(source));
+        return create(OperationMergeDelayError.mergeDelayError(source));
     }
 
     /**
@@ -1317,7 +1354,7 @@ public static <T> Observable<T> never() {
      * @return the source Observable, with its behavior modified as described
      */
     public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, final Func1<Exception, Observable<T>> resumeFunction) {
-        return _create(OperationOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(that, resumeFunction));
+        return create(OperationOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(that, resumeFunction));
     }
 
     /**
@@ -1381,7 +1418,7 @@ public Observable<T> call(Exception e) {
      * @return the source Observable, with its behavior modified as described
      */
     public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, final Observable<T> resumeSequence) {
-        return _create(OperationOnErrorResumeNextViaObservable.onErrorResumeNextViaObservable(that, resumeSequence));
+        return create(OperationOnErrorResumeNextViaObservable.onErrorResumeNextViaObservable(that, resumeSequence));
     }
 
     /**
@@ -1405,7 +1442,7 @@ public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, fina
      * @return the source Observable, with its behavior modified as described
      */
     public static <T> Observable<T> onErrorReturn(final Observable<T> that, Func1<Exception, T> resumeFunction) {
-        return _create(OperationOnErrorReturn.onErrorReturn(that, resumeFunction));
+        return create(OperationOnErrorReturn.onErrorReturn(that, resumeFunction));
     }
 
     /**
@@ -1435,7 +1472,7 @@ public static <T> Observable<T> onErrorReturn(final Observable<T> that, Func1<Ex
      * @see <a href="http://en.wikipedia.org/wiki/Fold_(higher-order_function)">Wikipedia: Fold (higher-order function)</a>
      */
     public static <T> Observable<T> reduce(Observable<T> sequence, Func2<T, T, T> accumulator) {
-        return takeLast(_create(OperationScan.scan(sequence, accumulator)), 1);
+        return takeLast(create(OperationScan.scan(sequence, accumulator)), 1);
     }
 
     /**
@@ -1507,7 +1544,7 @@ public T call(T t1, T t2) {
      * @see <a href="http://en.wikipedia.org/wiki/Fold_(higher-order_function)">Wikipedia: Fold (higher-order function)</a>
      */
     public static <T> Observable<T> reduce(Observable<T> sequence, T initialValue, Func2<T, T, T> accumulator) {
-        return takeLast(_create(OperationScan.scan(sequence, initialValue, accumulator)), 1);
+        return takeLast(create(OperationScan.scan(sequence, initialValue, accumulator)), 1);
     }
 
     /**
@@ -1571,7 +1608,7 @@ public T call(T t1, T t2) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v%3Dvs.103).aspx">MSDN: Observable.Scan</a>
      */
     public static <T> Observable<T> scan(Observable<T> sequence, Func2<T, T, T> accumulator) {
-        return _create(OperationScan.scan(sequence, accumulator));
+        return create(OperationScan.scan(sequence, accumulator));
     }
 
     /**
@@ -1629,7 +1666,7 @@ public T call(T t1, T t2) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v%3Dvs.103).aspx">MSDN: Observable.Scan</a>
      */
     public static <T> Observable<T> scan(Observable<T> sequence, T initialValue, Func2<T, T, T> accumulator) {
-        return _create(OperationScan.scan(sequence, initialValue, accumulator));
+        return create(OperationScan.scan(sequence, initialValue, accumulator));
     }
 
     /**
@@ -1669,20 +1706,28 @@ public T call(T t1, T t2) {
 
     /**
      * Determines whether all elements of an observable sequence satisfies a condition.
-     * @param sequence an observable sequence whose elements to apply the predicate to.
-     * @param predicate a function to test each element for a condition.
-     * @param <T> the type of observable.
+     * 
+     * @param sequence
+     *            an observable sequence whose elements to apply the predicate to.
+     * @param predicate
+     *            a function to test each element for a condition.
+     * @param <T>
+     *            the type of observable.
      * @return true if all elements of an observable sequence satisfies a condition; otherwise, false.
      */
     public static <T> Observable<Boolean> all(final Observable<T> sequence, final Func1<T, Boolean> predicate) {
-        return _create(OperationAll.all(sequence, predicate));
+        return create(OperationAll.all(sequence, predicate));
     }
 
     /**
      * Determines whether all elements of an observable sequence satisfies a condition.
-     * @param sequence an observable sequence whose elements to apply the predicate to.
-     * @param predicate a function to test each element for a condition.
-     * @param <T> the type of observable.
+     * 
+     * @param sequence
+     *            an observable sequence whose elements to apply the predicate to.
+     * @param predicate
+     *            a function to test each element for a condition.
+     * @param <T>
+     *            the type of observable.
      * @return true if all elements of an observable sequence satisfies a condition; otherwise, false.
      */
     public static <T> Observable<Boolean> all(final Observable<T> sequence, Object predicate) {
@@ -1712,7 +1757,7 @@ public Boolean call(T t) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229847(v=vs.103).aspx">MSDN: Observable.Skip Method</a>
      */
     public static <T> Observable<T> skip(final Observable<T> items, int num) {
-        return _create(OperationSkip.skip(items, num));
+        return create(OperationSkip.skip(items, num));
     }
 
     /**
@@ -1730,7 +1775,7 @@ public static <T> Observable<T> skip(final Observable<T> items, int num) {
      * @return an Observable that is a chronologically well-behaved version of the source Observable
      */
     public static <T> Observable<T> synchronize(Observable<T> observable) {
-        return _create(OperationSynchronize.synchronize(observable));
+        return create(OperationSynchronize.synchronize(observable));
     }
 
     /**
@@ -1752,7 +1797,7 @@ public static <T> Observable<T> synchronize(Observable<T> observable) {
      *         Observable
      */
     public static <T> Observable<T> take(final Observable<T> items, final int num) {
-        return _create(OperationTake.take(items, num));
+        return create(OperationTake.take(items, num));
     }
 
     /**
@@ -1768,7 +1813,7 @@ public static <T> Observable<T> take(final Observable<T> items, final int num) {
      *         Observable
      */
     public static <T> Observable<T> takeLast(final Observable<T> items, final int count) {
-        return _create(OperationTakeLast.takeLast(items, count));
+        return create(OperationTakeLast.takeLast(items, count));
     }
 
     /**
@@ -1849,7 +1894,7 @@ public Boolean call(T t, Integer integer)
      *         items emitted by the source Observable
      */
     public static <T> Observable<List<T>> toList(final Observable<T> that) {
-        return _create(OperationToObservableList.toObservableList(that));
+        return create(OperationToObservableList.toObservableList(that));
     }
 
     /**
@@ -2047,7 +2092,7 @@ private static <T> T singleOrDefault(Observable<T> that, boolean hasDefault, T d
      * @return an Observable that emits each item in the source Iterable sequence
      */
     public static <T> Observable<T> toObservable(Iterable<T> iterable) {
-        return _create(OperationToObservableIterable.toObservableIterable(iterable));
+        return create(OperationToObservableIterable.toObservableIterable(iterable));
     }
 
     /**
@@ -2066,7 +2111,7 @@ public static <T> Observable<T> toObservable(Iterable<T> iterable) {
      * @return an Observable that emits the item from the source Future
      */
     public static <T> Observable<T> toObservable(Future<T> future) {
-        return _create(OperationToObservableFuture.toObservableFuture(future));
+        return create(OperationToObservableFuture.toObservableFuture(future));
     }
 
     /**
@@ -2090,7 +2135,7 @@ public static <T> Observable<T> toObservable(Future<T> future) {
      * @return an Observable that emits the item from the source Future
      */
     public static <T> Observable<T> toObservable(Future<T> future, long time, TimeUnit unit) {
-        return _create(OperationToObservableFuture.toObservableFuture(future, time, unit));
+        return create(OperationToObservableFuture.toObservableFuture(future, time, unit));
     }
 
     /**
@@ -2123,7 +2168,7 @@ public static <T> Observable<T> toObservable(T... items) {
      * @return
      */
     public static <T> Observable<List<T>> toSortedList(Observable<T> sequence) {
-        return _create(OperationToObservableSortedList.toSortedList(sequence));
+        return create(OperationToObservableSortedList.toSortedList(sequence));
     }
 
     /**
@@ -2136,7 +2181,7 @@ public static <T> Observable<List<T>> toSortedList(Observable<T> sequence) {
      * @return
      */
     public static <T> Observable<List<T>> toSortedList(Observable<T> sequence, Func2<T, T, Integer> sortFunction) {
-        return _create(OperationToObservableSortedList.toSortedList(sequence, sortFunction));
+        return create(OperationToObservableSortedList.toSortedList(sequence, sortFunction));
     }
 
     /**
@@ -2151,7 +2196,7 @@ public static <T> Observable<List<T>> toSortedList(Observable<T> sequence, Func2
     public static <T> Observable<List<T>> toSortedList(Observable<T> sequence, final Object sortFunction) {
         @SuppressWarnings("rawtypes")
         final FuncN _f = Functions.from(sortFunction);
-        return _create(OperationToObservableSortedList.toSortedList(sequence, new Func2<T, T, Integer>() {
+        return create(OperationToObservableSortedList.toSortedList(sequence, new Func2<T, T, Integer>() {
 
             @Override
             public Integer call(T t1, T t2) {
@@ -2186,7 +2231,7 @@ public Integer call(T t1, T t2) {
      * @return an Observable that emits the zipped results
      */
     public static <R, T0, T1> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Func2<T0, T1, R> reduceFunction) {
-        return _create(OperationZip.zip(w0, w1, reduceFunction));
+        return create(OperationZip.zip(w0, w1, reduceFunction));
     }
 
     /**
@@ -2310,7 +2355,7 @@ public R call(T0 t0, T1 t1) {
      * @return an Observable that emits the zipped results
      */
     public static <R, T0, T1, T2> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Observable<T2> w2, Func3<T0, T1, T2, R> function) {
-        return _create(OperationZip.zip(w0, w1, w2, function));
+        return create(OperationZip.zip(w0, w1, w2, function));
     }
 
     /**
@@ -2385,7 +2430,7 @@ public R call(T0 t0, T1 t1, T2 t2) {
      * @return an Observable that emits the zipped results
      */
     public static <R, T0, T1, T2, T3> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Observable<T2> w2, Observable<T3> w3, Func4<T0, T1, T2, T3, R> reduceFunction) {
-        return _create(OperationZip.zip(w0, w1, w2, w3, reduceFunction));
+        return create(OperationZip.zip(w0, w1, w2, w3, reduceFunction));
     }
 
     /**
@@ -2472,7 +2517,7 @@ public Boolean call(T t1) {
      * Filters an Observable by discarding any of its emissions that do not meet some test.
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/filter.png">
-     *
+     * 
      * @param predicate
      *            a function that evaluates the items emitted by the source Observable, returning
      *            <code>true</code> if they pass the filter
@@ -2663,7 +2708,7 @@ public Observable<Notification<T>> materialize() {
      */
     @SuppressWarnings("unchecked")
     public <T2> Observable<T2> dematerialize() {
-        return dematerialize((Observable<Notification<T2>>)this);
+        return dematerialize((Observable<Notification<T2>>) this);
     }
 
     /**
@@ -3008,7 +3053,9 @@ public Observable<T> scan(final T initialValue, final Object accumulator) {
 
     /**
      * Determines whether all elements of an observable sequence satisfies a condition.
-     * @param predicate a function to test each element for a condition.
+     * 
+     * @param predicate
+     *            a function to test each element for a condition.
      * @return true if all elements of an observable sequence satisfies a condition; otherwise, false.
      */
     public Observable<Boolean> all(Func1<T, Boolean> predicate) {
@@ -3017,7 +3064,9 @@ public Observable<Boolean> all(Func1<T, Boolean> predicate) {
 
     /**
      * Determines whether all elements of an observable sequence satisfies a condition.
-     * @param predicate a function to test each element for a condition.
+     * 
+     * @param predicate
+     *            a function to test each element for a condition.
      * @return true if all elements of an observable sequence satisfies a condition; otherwise, false.
      */
     public Observable<Boolean> all(Object predicate) {
@@ -3262,6 +3311,27 @@ public Iterable<T> mostRecent(T initialValue) {
         return mostRecent(this, initialValue);
     }
 
+    /**
+     * Whether a given {@link Function} is an internal implementation inside rx.* packages or not.
+     * <p>
+     * For why this is being used see https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+     * 
+     * NOTE: If strong reasons for not depending on package names comes up then the implementation of this method can change to looking for a marker interface.
+     * 
+     * @param f
+     * @return
+     */
+    private boolean isInternalImplementation(Object o) {
+        if (o == null) {
+            return true;
+        }
+        // prevent double-wrapping (yeah it happens)
+        if (o instanceof AtomicObserver)
+            return true;
+        // we treat the following package as "internal" and don't wrap it
+        return o.getClass().getPackage().getName().startsWith("rx.operators");
+    }
+
     public static class UnitTest {
 
         @Mock
@@ -3556,6 +3626,204 @@ public void testMaterializeDematerializeChaining() {
             verify(observer, times(0)).onError(any(Exception.class));
         }
 
+        /**
+         * The error from the user provided Observer is not handled by the subscribe method try/catch.
+         * 
+         * It is handled by the AtomicObserver that wraps the provided Observer.
+         * 
+         * Result: Passes (if AtomicObserver functionality exists)
+         */
+        @Test
+        public void testCustomObservableWithErrorInObserverAsynchronous() throws InterruptedException {
+            final CountDownLatch latch = new CountDownLatch(1);
+            final AtomicInteger count = new AtomicInteger();
+            final AtomicReference<Exception> error = new AtomicReference<Exception>();
+            Observable.create(new Func1<Observer<String>, Subscription>() {
+
+                @Override
+                public Subscription call(final Observer<String> observer) {
+                    final BooleanSubscription s = new BooleanSubscription();
+                    new Thread(new Runnable() {
+
+                        @Override
+                        public void run() {
+                            try {
+                                if (!s.isUnsubscribed()) {
+                                    observer.onNext("1");
+                                    observer.onNext("2");
+                                    observer.onNext("three");
+                                    observer.onNext("4");
+                                    observer.onCompleted();
+                                }
+                            } finally {
+                                latch.countDown();
+                            }
+                        }
+                    }).start();
+                    return s;
+                }
+            }).subscribe(new Observer<String>() {
+                @Override
+                public void onCompleted() {
+                    System.out.println("completed");
+                }
+
+                @Override
+                public void onError(Exception e) {
+                    error.set(e);
+                    System.out.println("error");
+                    e.printStackTrace();
+                }
+
+                @Override
+                public void onNext(String v) {
+                    int num = Integer.parseInt(v);
+                    System.out.println(num);
+                    // doSomething(num);
+                    count.incrementAndGet();
+                }
+
+            });
+
+            // wait for async sequence to complete
+            latch.await();
+
+            assertEquals(2, count.get());
+            assertNotNull(error.get());
+            if (!(error.get() instanceof NumberFormatException)) {
+                fail("It should be a NumberFormatException");
+            }
+        }
+
+        /**
+         * The error from the user provided Observer is handled by the subscribe try/catch because this is synchronous
+         * 
+         * Result: Passes
+         */
+        @Test
+        public void testCustomObservableWithErrorInObserverSynchronous() {
+            final AtomicInteger count = new AtomicInteger();
+            final AtomicReference<Exception> error = new AtomicReference<Exception>();
+            Observable.create(new Func1<Observer<String>, Subscription>() {
+
+                @Override
+                public Subscription call(Observer<String> observer) {
+                    observer.onNext("1");
+                    observer.onNext("2");
+                    observer.onNext("three");
+                    observer.onNext("4");
+                    observer.onCompleted();
+                    return Subscriptions.empty();
+                }
+            }).subscribe(new Observer<String>() {
+
+                @Override
+                public void onCompleted() {
+                    System.out.println("completed");
+                }
+
+                @Override
+                public void onError(Exception e) {
+                    error.set(e);
+                    System.out.println("error");
+                    e.printStackTrace();
+                }
+
+                @Override
+                public void onNext(String v) {
+                    int num = Integer.parseInt(v);
+                    System.out.println(num);
+                    // doSomething(num);
+                    count.incrementAndGet();
+                }
+
+            });
+            assertEquals(2, count.get());
+            assertNotNull(error.get());
+            if (!(error.get() instanceof NumberFormatException)) {
+                fail("It should be a NumberFormatException");
+            }
+        }
+
+        /**
+         * The error from the user provided Observable is handled by the subscribe try/catch because this is synchronous
+         * 
+         * 
+         * Result: Passes
+         */
+        @Test
+        public void testCustomObservableWithErrorInObservableSynchronous() {
+            final AtomicInteger count = new AtomicInteger();
+            final AtomicReference<Exception> error = new AtomicReference<Exception>();
+            Observable.create(new Func1<Observer<String>, Subscription>() {
+
+                @Override
+                public Subscription call(Observer<String> observer) {
+                    observer.onNext("1");
+                    observer.onNext("2");
+                    throw new NumberFormatException();
+                }
+            }).subscribe(new Observer<String>() {
+
+                @Override
+                public void onCompleted() {
+                    System.out.println("completed");
+                }
+
+                @Override
+                public void onError(Exception e) {
+                    error.set(e);
+                    System.out.println("error");
+                    e.printStackTrace();
+                }
+
+                @Override
+                public void onNext(String v) {
+                    System.out.println(v);
+                    count.incrementAndGet();
+                }
+
+            });
+            assertEquals(2, count.get());
+            assertNotNull(error.get());
+            if (!(error.get() instanceof NumberFormatException)) {
+                fail("It should be a NumberFormatException");
+            }
+        }
+
+        @Test
+        public void testForEachWithError() {
+            try {
+                Observable.create(new Func1<Observer<String>, Subscription>() {
+
+                    @Override
+                    public Subscription call(final Observer<String> observer) {
+                        final BooleanSubscription subscription = new BooleanSubscription();
+                        new Thread(new Runnable() {
+
+                            @Override
+                            public void run() {
+                                observer.onNext("one");
+                                observer.onNext("two");
+                                observer.onNext("three");
+                                observer.onCompleted();
+                            }
+                        }).start();
+                        return subscription;
+                    }
+                }).forEach(new Action1<String>() {
+
+                    @Override
+                    public void call(String t1) {
+                        throw new RuntimeException("fail");
+                    }
+                });
+                fail("we expect an exception to be thrown");
+            } catch (Exception e) {
+                // do nothing as we expect this
+            }
+        }
+
         private static class TestException extends RuntimeException {
             private static final long serialVersionUID = 1L;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTake.java b/rxjava-core/src/main/java/rx/operators/OperationTake.java
index 5ea6b627e4..a4b1922e22 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationTake.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationTake.java
@@ -18,7 +18,7 @@
 import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
-import static rx.operators.AbstractOperation.UnitTest.*;
+import static rx.operators.Tester.UnitTest.*;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -281,6 +281,7 @@ public void run() {
             }
 
         }
+
     }
 
 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTakeWhile.java b/rxjava-core/src/main/java/rx/operators/OperationTakeWhile.java
index 1bad2d36e5..d00d66c3aa 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationTakeWhile.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationTakeWhile.java
@@ -26,11 +26,13 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.subjects.Subject;
 import rx.subscriptions.Subscriptions;
 import rx.util.AtomicObservableSubscription;
+import rx.util.AtomicObserver;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
-import rx.subjects.Subject;
+
 /**
  * Returns values from an observable sequence as long as a specified condition is true, and then skips the remaining values.
  */
@@ -45,7 +47,7 @@ public final class OperationTakeWhile {
      * @return
      */
     public static <T> Func1<Observer<T>, Subscription> takeWhile(final Observable<T> items, final Func1<T, Boolean> predicate) {
-        return takeWhileWithIndex(items, OperationTakeWhile.<T>skipIndex(predicate));
+        return takeWhileWithIndex(items, OperationTakeWhile.<T> skipIndex(predicate));
     }
 
     /**
@@ -108,7 +110,10 @@ private class ItemObserver implements Observer<T> {
             private final Observer<T> observer;
 
             public ItemObserver(Observer<T> observer) {
-                this.observer = observer;
+                // Using AtomicObserver because the unsubscribe, onCompleted, onError and error handling behavior
+                // needs "isFinished" logic to not send duplicated events
+                // The 'testTakeWhile1' and 'testTakeWhile2' tests fail without this.
+                this.observer = new AtomicObserver<T>(subscription, observer);
             }
 
             @Override
@@ -126,8 +131,7 @@ public void onNext(T args) {
                 Boolean isSelected;
                 try {
                     isSelected = predicate.call(args, counter.getAndIncrement());
-                }
-                catch (Exception e) {
+                } catch (Exception e) {
                     observer.onError(e);
                     return;
                 }
@@ -171,7 +175,7 @@ public Boolean call(Integer input)
         @Test
         public void testTakeWhileOnSubject1() {
             Subject<Integer> s = Subject.create();
-            Observable<Integer> w = (Observable<Integer>)s;
+            Observable<Integer> w = (Observable<Integer>) s;
             Observable<Integer> take = Observable.create(takeWhile(w, new Func1<Integer, Boolean>()
             {
                 @Override
diff --git a/rxjava-core/src/main/java/rx/operators/AbstractOperation.java b/rxjava-core/src/main/java/rx/operators/Tester.java
similarity index 95%
rename from rxjava-core/src/main/java/rx/operators/AbstractOperation.java
rename to rxjava-core/src/main/java/rx/operators/Tester.java
index dd71bd837a..9692323015 100644
--- a/rxjava-core/src/main/java/rx/operators/AbstractOperation.java
+++ b/rxjava-core/src/main/java/rx/operators/Tester.java
@@ -17,11 +17,16 @@
 import rx.util.functions.Func1;
 
 /**
- * Common utility functions for operator implementations and tests.
+ * Common utility functions for testing operator implementations.
  */
-/* package */class AbstractOperation
-{
-    private AbstractOperation() {
+/* package */class Tester {
+    /*
+     * This is purposefully package-only so it does not leak into the public API outside of this package.
+     * 
+     * This package is implementation details and not part of the Javadocs and thus can change without breaking backwards compatibility.
+     */
+
+    private Tester() {
     }
 
     public static class UnitTest {
diff --git a/rxjava-core/src/main/java/rx/util/AtomicObserver.java b/rxjava-core/src/main/java/rx/util/AtomicObserver.java
index b3e18f5b3e..24519dc276 100644
--- a/rxjava-core/src/main/java/rx/util/AtomicObserver.java
+++ b/rxjava-core/src/main/java/rx/util/AtomicObserver.java
@@ -1,8 +1,10 @@
 package rx.util;
 
+import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Observer;
+import rx.plugins.RxJavaPlugins;
 
 /**
  * Wrapper around Observer to ensure compliance with Rx contract.
@@ -32,7 +34,7 @@
  * <li>When onError or onComplete occur it will unsubscribe from the Observable (if executing asynchronously).</li>
  * </ul>
  * <p>
- * It will not synchronized onNext execution. Use the {@link SynchronizedObserver} to do that.
+ * It will not synchronize onNext execution. Use the {@link SynchronizedObserver} to do that.
  * 
  * @param <T>
  */
@@ -50,7 +52,12 @@ public AtomicObserver(AtomicObservableSubscription subscription, Observer<T> act
     @Override
     public void onCompleted() {
         if (isFinished.compareAndSet(false, true)) {
-            actual.onCompleted();
+            try {
+                actual.onCompleted();
+            } catch (Exception e) {
+                // handle errors if the onCompleted implementation fails, not just if the Observable fails
+                onError(e);
+            }
             // auto-unsubscribe
             subscription.unsubscribe();
         }
@@ -59,7 +66,17 @@ public void onCompleted() {
     @Override
     public void onError(Exception e) {
         if (isFinished.compareAndSet(false, true)) {
-            actual.onError(e);
+            try {
+                actual.onError(e);
+            } catch (Exception e2) {
+                // if the onError itself fails then pass to the plugin
+                // see https://github.com/Netflix/RxJava/issues/216 for further discussion
+                RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
+                RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);
+                // and throw exception despite that not being proper for Rx
+                // https://github.com/Netflix/RxJava/issues/198
+                throw new RuntimeException("Error occurred when trying to propagate error to Observer.onError", new CompositeException(Arrays.asList(e, e2)));
+            }
             // auto-unsubscribe
             subscription.unsubscribe();
         }