diff --git a/src/main/java/rx/Observable.java b/src/main/java/rx/Observable.java
index f3be61283d..5b474e7791 100644
--- a/src/main/java/rx/Observable.java
+++ b/src/main/java/rx/Observable.java
@@ -2814,7 +2814,36 @@ public static <T> Observable<Boolean> sequenceEqual(Observable<? extends T> firs
      * @see <a href="http://reactivex.io/documentation/operators/switch.html">ReactiveX operators documentation: Switch</a>
      */
     public static <T> Observable<T> switchOnNext(Observable<? extends Observable<? extends T>> sequenceOfSequences) {
-        return sequenceOfSequences.lift(OperatorSwitch.<T>instance());
+        return sequenceOfSequences.lift(OperatorSwitch.<T>instance(false));
+    }
+
+    /**
+     * Converts an Observable that emits Observables into an Observable that emits the items emitted by the
+     * most recently emitted of those Observables and delays any exception until all Observables terminate.
+     * <p>
+     * <img width="640" height="370" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchDo.png" alt="">
+     * <p>
+     * {@code switchOnNext} subscribes to an Observable that emits Observables. Each time it observes one of
+     * these emitted Observables, the Observable returned by {@code switchOnNext} begins emitting the items
+     * emitted by that Observable. When a new Observable is emitted, {@code switchOnNext} stops emitting items
+     * from the earlier-emitted Observable and begins emitting items from the new one.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code switchOnNext} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param <T> the item type
+     * @param sequenceOfSequences
+     *            the source Observable that emits Observables
+     * @return an Observable that emits the items emitted by the Observable most recently emitted by the source
+     *         Observable
+     * @see <a href="http://reactivex.io/documentation/operators/switch.html">ReactiveX operators documentation: Switch</a>
+     * @Experimental The behavior of this can change at any time.
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public static <T> Observable<T> switchOnNextDelayError(Observable<? extends Observable<? extends T>> sequenceOfSequences) {
+        return sequenceOfSequences.lift(OperatorSwitch.<T>instance(true));
     }
 
     /**
@@ -8637,6 +8666,30 @@ public final <R> Observable<R> switchMap(Func1<? super T, ? extends Observable<?
         return switchOnNext(map(func));
     }
 
+    /**
+     * Returns a new Observable by applying a function that you supply to each item emitted by the source
+     * Observable that returns an Observable, and then emitting the items emitted by the most recently emitted
+     * of these Observables and delays any error until all Observables terminate.
+     * <p>
+     * <img width="640" height="350" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchMap.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code switchMap} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param func
+     *            a function that, when applied to an item emitted by the source Observable, returns an
+     *            Observable
+     * @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
+     * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     * @Experimental The behavior of this can change at any time.
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public final <R> Observable<R> switchMapDelayError(Func1<? super T, ? extends Observable<? extends R>> func) {
+        return switchOnNextDelayError(map(func));
+    }
+
     /**
      * Returns an Observable that emits only the first {@code count} items emitted by the source Observable. If the source emits fewer than 
      * {@code count} items then all of its items are emitted.
diff --git a/src/main/java/rx/internal/operators/OperatorSwitch.java b/src/main/java/rx/internal/operators/OperatorSwitch.java
index 5f95f38c3d..7d706f2a95 100644
--- a/src/main/java/rx/internal/operators/OperatorSwitch.java
+++ b/src/main/java/rx/internal/operators/OperatorSwitch.java
@@ -15,15 +15,14 @@
  */
 package rx.internal.operators;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
 
+import rx.*;
 import rx.Observable;
 import rx.Observable.Operator;
-import rx.Producer;
-import rx.Subscriber;
+import rx.exceptions.CompositeException;
 import rx.internal.producers.ProducerArbiter;
-import rx.observers.SerializedSubscriber;
+import rx.plugins.RxJavaPlugins;
 import rx.subscriptions.SerialSubscription;
 
 /**
@@ -38,49 +37,67 @@ public final class OperatorSwitch<T> implements Operator<T, Observable<? extends
     /** Lazy initialization via inner-class holder. */
     private static final class Holder {
         /** A singleton instance. */
-        static final OperatorSwitch<Object> INSTANCE = new OperatorSwitch<Object>();
+        static final OperatorSwitch<Object> INSTANCE = new OperatorSwitch<Object>(false);
+    }
+    /** Lazy initialization via inner-class holder. */
+    private static final class HolderDelayError {
+        /** A singleton instance. */
+        static final OperatorSwitch<Object> INSTANCE = new OperatorSwitch<Object>(true);
     }
     /**
      * @return a singleton instance of this stateless operator.
      */
     @SuppressWarnings({ "unchecked" })
-    public static <T> OperatorSwitch<T> instance() {
+    public static <T> OperatorSwitch<T> instance(boolean delayError) {
+        if (delayError) {
+            return (OperatorSwitch<T>)HolderDelayError.INSTANCE;
+        }
         return (OperatorSwitch<T>)Holder.INSTANCE;
     }
 
-    OperatorSwitch() { }
+    final boolean delayError;
+    
+    OperatorSwitch(boolean delayError) { 
+        this.delayError = delayError;
+    }
 
     @Override
     public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
-        SwitchSubscriber<T> sws = new SwitchSubscriber<T>(child);
+        SwitchSubscriber<T> sws = new SwitchSubscriber<T>(child, delayError);
         child.add(sws);
+        sws.init();
         return sws;
     }
 
     private static final class SwitchSubscriber<T> extends Subscriber<Observable<? extends T>> {
-        final SerializedSubscriber<T> serializedChild;
+        final Subscriber<? super T> child;
         final SerialSubscription ssub;
-        final Object guard = new Object();
-        final NotificationLite<?> nl = NotificationLite.instance();
         final ProducerArbiter arbiter;
         
-        /** Guarded by guard. */
-        int index;
-        /** Guarded by guard. */
-        boolean active;
-        /** Guarded by guard. */
+        final boolean delayError;
+        
+        long index;
+        
+        Throwable error;
+        
         boolean mainDone;
-        /** Guarded by guard. */
-        List<Object> queue;
-        /** Guarded by guard. */
+        
+        List<T> queue;
+        
+        boolean innerActive;
+        
         boolean emitting;
-        /** Guarded by guard. */
-        InnerSubscriber<T> currentSubscriber;
+        
+        boolean missed;
 
-        SwitchSubscriber(Subscriber<? super T> child) {
-            serializedChild = new SerializedSubscriber<T>(child);
-            arbiter = new ProducerArbiter();
-            ssub = new SerialSubscription();
+        SwitchSubscriber(Subscriber<? super T> child, boolean delayError) {
+            this.child = child;
+            this.arbiter = new ProducerArbiter();
+            this.ssub = new SerialSubscription();
+            this.delayError = delayError;
+        }
+        
+        void init() {
             child.add(ssub);
             child.setProducer(new Producer(){
 
@@ -95,186 +112,232 @@ public void request(long n) {
 
         @Override
         public void onNext(Observable<? extends T> t) {
-            final int id;
-            synchronized (guard) {
-                id = ++index;
-                active = true;
-                currentSubscriber = new InnerSubscriber<T>(id, arbiter, this);
+            InnerSubscriber<T> inner;
+            synchronized (this) {
+                long id = ++index;
+                inner = new InnerSubscriber<T>(id, this);
+                innerActive = true;
             }
-            ssub.set(currentSubscriber);
-            t.unsafeSubscribe(currentSubscriber);
+            ssub.set(inner);
+            
+            t.unsafeSubscribe(inner);
         }
 
         @Override
         public void onError(Throwable e) {
-            serializedChild.onError(e);
-            unsubscribe();
+            synchronized (this) {
+                e = updateError(e);
+                mainDone = true;
+                
+                if (emitting) {
+                    missed = true;
+                    return;
+                }
+                if (delayError && innerActive) {
+                    return;
+                }
+                emitting = true;
+            }
+            
+            child.onError(e);
         }
 
         @Override
         public void onCompleted() {
-            List<Object> localQueue;
-            synchronized (guard) {
+            Throwable ex;
+            synchronized (this) {
                 mainDone = true;
-                if (active) {
+                if (emitting) {
+                    missed = true;
                     return;
                 }
-                if (emitting) {
-                    if (queue == null) {
-                        queue = new ArrayList<Object>();
-                    }
-                    queue.add(nl.completed());
+                if (innerActive) {
                     return;
                 }
-                localQueue = queue;
-                queue = null;
                 emitting = true;
+                ex = error;
             }
-            drain(localQueue);
-            serializedChild.onCompleted();
-            unsubscribe();
+            if (ex == null) {
+                child.onCompleted();
+            } else {
+                child.onError(ex);
+            }
+        }
+        
+        Throwable updateError(Throwable e) {
+            Throwable ex = error;
+            if (ex == null) {
+                error = e;
+            } else
+            if (ex instanceof CompositeException) {
+                CompositeException ce = (CompositeException) ex;
+                List<Throwable> list = new ArrayList<Throwable>(ce.getExceptions());
+                list.add(e);
+                e = new CompositeException(list);
+                error = e;
+            } else {
+                e = new CompositeException(Arrays.asList(ex, e));
+                error = e;
+            }
+            return e;
         }
-        void emit(T value, int id, InnerSubscriber<T> innerSubscriber) {
-            List<Object> localQueue;
-            synchronized (guard) {
+        
+        void emit(T value, long id) {
+            synchronized (this) {
                 if (id != index) {
                     return;
                 }
+                
                 if (emitting) {
-                    if (queue == null) {
-                        queue = new ArrayList<Object>();
+                    List<T> q = queue;
+                    if (q == null) {
+                        q = new ArrayList<T>(4);
+                        queue = q;
                     }
-                    queue.add(value);
+                    q.add(value);
+                    missed = true;
                     return;
                 }
-                localQueue = queue;
-                queue = null;
+                
                 emitting = true;
             }
-            boolean once = true;
-            boolean skipFinal = false;
-            try {
-                do {
-                    drain(localQueue);
-                    if (once) {
-                        once = false;
-                        serializedChild.onNext(value);
-                        arbiter.produced(1);                        
+            
+            child.onNext(value);
+            
+            arbiter.produced(1);
+            
+            for (;;) {
+                if (child.isUnsubscribed()) {
+                    return;
+                }
+                
+                Throwable localError;
+                boolean localMainDone;
+                boolean localActive;
+                List<T> localQueue;
+                synchronized (this) {
+                    if (!missed) {
+                        emitting = false;
+                        return;
                     }
-                    synchronized (guard) {
-                        localQueue = queue;
-                        queue = null;
-                        if (localQueue == null) {
-                            emitting = false;
-                            skipFinal = true;
-                            break;
-                        }
+                    
+                    localError = error;
+                    localMainDone = mainDone;
+                    localQueue = queue;
+                    localActive = innerActive;
+                }
+                
+                if (!delayError && localError != null) {
+                    child.onError(localError);
+                    return;
+                }
+                
+                if (localQueue == null && !localActive && localMainDone) {
+                    if (localError != null) {
+                        child.onError(localError);
+                    } else {
+                        child.onCompleted();
                     }
-                } while (!serializedChild.isUnsubscribed());
-            } finally {
-                if (!skipFinal) {
-                    synchronized (guard) {
-                        emitting = false;
+                    return;
+                }
+                
+                if (localQueue != null) {
+                    int n = 0;
+                    for (T v : localQueue) {
+                        if (child.isUnsubscribed()) {
+                            return;
+                        }
+
+                        child.onNext(v);
+                        n++;
                     }
+                    
+                    arbiter.produced(n);
                 }
             }
         }
-        void drain(List<Object> localQueue) {
-            if (localQueue == null) {
-                return;
-            }
-            for (Object o : localQueue) {
-                if (nl.isCompleted(o)) {
-                    serializedChild.onCompleted();
-                    break;
-                } else
-                if (nl.isError(o)) {
-                    serializedChild.onError(nl.getError(o));
-                    break;
+
+        void error(Throwable e, long id) {
+            boolean drop;
+            synchronized (this) {
+                if (id == index) {
+                    innerActive = false;
+                    
+                    e = updateError(e);
+                    
+                    if (emitting) {
+                        missed = true;
+                        return;
+                    }
+                    if (delayError && !mainDone) {
+                        return;
+                    }
+                    emitting = true;
+                    
+                    drop = false;
                 } else {
-                    @SuppressWarnings("unchecked")
-                    T t = (T)o;
-                    serializedChild.onNext(t);
-                    arbiter.produced(1);
+                    drop = true;
                 }
             }
+            
+            if (drop) {
+                pluginError(e);
+            } else {
+                child.onError(e);
+            }
         }
-
-        void error(Throwable e, int id) {
-            List<Object> localQueue;
-            synchronized (guard) {
+        
+        void complete(long id) {
+            Throwable ex;
+            synchronized (this) {
                 if (id != index) {
                     return;
                 }
+                innerActive = false;
+                
                 if (emitting) {
-                    if (queue == null) {
-                        queue = new ArrayList<Object>();
-                    }
-                    queue.add(nl.error(e));
+                    missed = true;
                     return;
                 }
+                
+                ex = error;
 
-                localQueue = queue;
-                queue = null;
-                emitting = true;
-            }
-
-            drain(localQueue);
-            serializedChild.onError(e);
-            unsubscribe();
-        }
-        void complete(int id) {
-            List<Object> localQueue;
-            synchronized (guard) {
-                if (id != index) {
-                    return;
-                }
-                active = false;
                 if (!mainDone) {
                     return;
                 }
-                if (emitting) {
-                    if (queue == null) {
-                        queue = new ArrayList<Object>();
-                    }
-                    queue.add(nl.completed());
-                    return;
-                }
-
-                localQueue = queue;
-                queue = null;
-                emitting = true;
             }
-
-            drain(localQueue);
-            serializedChild.onCompleted();
-            unsubscribe();
+            
+            if (ex != null) {
+                child.onError(ex);
+            } else {
+                child.onCompleted();
+            }
         }
 
+        void pluginError(Throwable e) {
+            RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
+        }
     }
     
     private static final class InnerSubscriber<T> extends Subscriber<T> {
 
-        private final int id;
-
-        private final ProducerArbiter arbiter;
+        private final long id;
 
         private final SwitchSubscriber<T> parent;
 
-        InnerSubscriber(int id, ProducerArbiter arbiter, SwitchSubscriber<T> parent) {
+        InnerSubscriber(long id, SwitchSubscriber<T> parent) {
             this.id = id;
-            this.arbiter = arbiter;
             this.parent = parent;
         }
         
         @Override
         public void setProducer(Producer p) {
-            arbiter.setProducer(p);
+            parent.arbiter.setProducer(p);
         }
 
         @Override
         public void onNext(T t) {
-            parent.emit(t, id, this);
+            parent.emit(t, id);
         }
 
         @Override
diff --git a/src/test/java/rx/internal/operators/OperatorSwitchTest.java b/src/test/java/rx/internal/operators/OperatorSwitchTest.java
index 63de5d0d81..55170ab9ff 100644
--- a/src/test/java/rx/internal/operators/OperatorSwitchTest.java
+++ b/src/test/java/rx/internal/operators/OperatorSwitchTest.java
@@ -15,38 +15,25 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
-import static org.mockito.Mockito.inOrder;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.TimeUnit;
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+
+import java.util.*;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.*;
 import org.mockito.InOrder;
 
+import rx.*;
 import rx.Observable;
 import rx.Observer;
-import rx.Producer;
-import rx.Scheduler;
-import rx.Subscriber;
-import rx.exceptions.TestException;
-import rx.functions.Action0;
-import rx.functions.Action1;
-import rx.functions.Func1;
+import rx.exceptions.*;
+import rx.functions.*;
 import rx.observers.TestSubscriber;
 import rx.schedulers.TestScheduler;
+import rx.subjects.PublishSubject;
 
 public class OperatorSwitchTest {
 
@@ -667,8 +654,140 @@ public Observable<Long> call(Long t) {
         ts.requestMore(Long.MAX_VALUE - 1);
         ts.awaitTerminalEvent();
         assertTrue(ts.getOnNextEvents().size() > 0);
-        assertEquals(5, (int) requests.size());
+        assertEquals(5, requests.size());
         assertEquals(Long.MAX_VALUE, (long) requests.get(requests.size()-1));
     }
 
+    @Test
+    public void mainError() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        PublishSubject<Integer> source = PublishSubject.create();
+        
+        source.switchMapDelayError(new Func1<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(Integer v) {
+                return Observable.range(v, 2);
+            }
+        }).subscribe(ts);
+        
+        source.onNext(1);
+        source.onNext(2);
+        source.onError(new TestException());
+        
+        ts.assertValues(1, 2, 2, 3);
+        ts.assertError(TestException.class);
+        ts.assertNotCompleted();
+    }
+
+    @Test
+    public void innerError() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        Observable.range(0, 3).switchMapDelayError(new Func1<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(Integer v) {
+                return v == 1 ? Observable.<Integer>error(new TestException()) : Observable.range(v, 2);
+            }
+        }).subscribe(ts);
+        
+        ts.assertValues(0, 1, 2, 3);
+        ts.assertError(TestException.class);
+        ts.assertNotCompleted();
+    }
+
+    @Test
+    public void innerAllError() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        Observable.range(0, 3).switchMapDelayError(new Func1<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(Integer v) {
+                return Observable.range(v, 2).concatWith(Observable.<Integer>error(new TestException()));
+            }
+        }).subscribe(ts);
+        
+        ts.assertValues(0, 1, 1, 2, 2, 3);
+        ts.assertError(CompositeException.class);
+        ts.assertNotCompleted();
+        
+        List<Throwable> exceptions = ((CompositeException)ts.getOnErrorEvents().get(0)).getExceptions();
+        
+        assertEquals(3, exceptions.size());
+        
+        for (Throwable ex : exceptions) {
+            assertTrue(ex.toString(), ex instanceof TestException);
+        }
+    }
+
+    @Test
+    public void backpressure() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        
+        Observable.range(0, 3).switchMapDelayError(new Func1<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(Integer v) {
+                return Observable.range(v, 2);
+            }
+        }).subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+        
+        ts.requestMore(2);
+        
+        ts.assertValues(2, 3);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
+
+    @Test
+    public void backpressureWithSwitch() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+
+        PublishSubject<Integer> source = PublishSubject.create();
+
+        source.switchMapDelayError(new Func1<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(Integer v) {
+                return Observable.range(v, 2);
+            }
+        }).subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+        
+        ts.requestMore(1);
+        
+        source.onNext(0);
+        
+        ts.assertValues(0);
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+        
+        source.onNext(1);
+
+        ts.assertValues(0);
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+        
+        ts.requestMore(1);
+        
+        ts.assertValues(0, 1);
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+
+        source.onNext(2);
+
+        ts.requestMore(2);
+
+        source.onCompleted();
+        
+        ts.assertValues(0, 1, 2, 3);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
+
 }