diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundary.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundary.java
index d78570ee22..a0fb1f9f10 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundary.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundary.java
@@ -18,167 +18,199 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
-import io.reactivex.internal.observers.QueueDrainObserver;
 import io.reactivex.internal.queue.MpscLinkedQueue;
-import io.reactivex.internal.util.NotificationLite;
-import io.reactivex.observers.*;
+import io.reactivex.internal.util.AtomicThrowable;
+import io.reactivex.observers.DisposableObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.UnicastSubject;
 
 public final class ObservableWindowBoundary<T, B> extends AbstractObservableWithUpstream<T, Observable<T>> {
     final ObservableSource<B> other;
-    final int bufferSize;
+    final int capacityHint;
 
-    public ObservableWindowBoundary(ObservableSource<T> source, ObservableSource<B> other, int bufferSize) {
+    public ObservableWindowBoundary(ObservableSource<T> source, ObservableSource<B> other, int capacityHint) {
         super(source);
         this.other = other;
-        this.bufferSize = bufferSize;
+        this.capacityHint = capacityHint;
     }
 
     @Override
-    public void subscribeActual(Observer<? super Observable<T>> t) {
-        source.subscribe(new WindowBoundaryMainObserver<T, B>(new SerializedObserver<Observable<T>>(t), other, bufferSize));
+    public void subscribeActual(Observer<? super Observable<T>> observer) {
+        WindowBoundaryMainObserver<T, B> parent = new WindowBoundaryMainObserver<T, B>(observer, capacityHint);
+
+        observer.onSubscribe(parent);
+        other.subscribe(parent.boundaryObserver);
+
+        source.subscribe(parent);
     }
 
     static final class WindowBoundaryMainObserver<T, B>
-    extends QueueDrainObserver<T, Object, Observable<T>>
-    implements Disposable {
+    extends AtomicInteger
+    implements Observer<T>, Disposable, Runnable {
 
-        final ObservableSource<B> other;
-        final int bufferSize;
+        private static final long serialVersionUID = 2233020065421370272L;
 
-        Disposable s;
+        final Observer<? super Observable<T>> downstream;
 
-        final AtomicReference<Disposable> boundary = new AtomicReference<Disposable>();
+        final int capacityHint;
 
-        UnicastSubject<T> window;
+        final WindowBoundaryInnerObserver<T, B> boundaryObserver;
 
-        static final Object NEXT = new Object();
+        final AtomicReference<Disposable> upstream;
 
-        final AtomicLong windows = new AtomicLong();
+        final AtomicInteger windows;
 
-        WindowBoundaryMainObserver(Observer<? super Observable<T>> actual, ObservableSource<B> other,
-                int bufferSize) {
-            super(actual, new MpscLinkedQueue<Object>());
-            this.other = other;
-            this.bufferSize = bufferSize;
-            windows.lazySet(1);
-        }
+        final MpscLinkedQueue<Object> queue;
 
-        @Override
-        public void onSubscribe(Disposable s) {
-            if (DisposableHelper.validate(this.s, s)) {
-                this.s = s;
+        final AtomicThrowable errors;
 
-                Observer<? super Observable<T>> a = actual;
-                a.onSubscribe(this);
+        final AtomicBoolean stopWindows;
 
-                if (cancelled) {
-                    return;
-                }
+        static final Object NEXT_WINDOW = new Object();
 
-                UnicastSubject<T> w = UnicastSubject.create(bufferSize);
+        volatile boolean done;
 
-                window = w;
+        UnicastSubject<T> window;
 
-                a.onNext(w);
+        WindowBoundaryMainObserver(Observer<? super Observable<T>> downstream, int capacityHint) {
+            this.downstream = downstream;
+            this.capacityHint = capacityHint;
+            this.boundaryObserver = new WindowBoundaryInnerObserver<T, B>(this);
+            this.upstream = new AtomicReference<Disposable>();
+            this.windows = new AtomicInteger(1);
+            this.queue = new MpscLinkedQueue<Object>();
+            this.errors = new AtomicThrowable();
+            this.stopWindows = new AtomicBoolean();
+        }
 
-                WindowBoundaryInnerObserver<T, B> inner = new WindowBoundaryInnerObserver<T, B>(this);
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.setOnce(upstream, d)) {
 
-                if (boundary.compareAndSet(null, inner)) {
-                    windows.getAndIncrement();
-                    other.subscribe(inner);
-                }
+                innerNext();
             }
         }
 
         @Override
         public void onNext(T t) {
-            if (fastEnter()) {
-                UnicastSubject<T> w = window;
-
-                w.onNext(t);
+            queue.offer(t);
+            drain();
+        }
 
-                if (leave(-1) == 0) {
-                    return;
-                }
+        @Override
+        public void onError(Throwable e) {
+            boundaryObserver.dispose();
+            if (errors.addThrowable(e)) {
+                done = true;
+                drain();
             } else {
-                queue.offer(NotificationLite.next(t));
-                if (!enter()) {
-                    return;
-                }
+                RxJavaPlugins.onError(e);
             }
-            drainLoop();
         }
 
         @Override
-        public void onError(Throwable t) {
-            if (done) {
-                RxJavaPlugins.onError(t);
-                return;
-            }
-            error = t;
+        public void onComplete() {
+            boundaryObserver.dispose();
             done = true;
-            if (enter()) {
-                drainLoop();
-            }
+            drain();
+        }
 
-            if (windows.decrementAndGet() == 0) {
-                DisposableHelper.dispose(boundary);
+        @Override
+        public void dispose() {
+            if (stopWindows.compareAndSet(false, true)) {
+                boundaryObserver.dispose();
+                if (windows.decrementAndGet() == 0) {
+                    DisposableHelper.dispose(upstream);
+                }
             }
-
-            actual.onError(t);
         }
 
         @Override
-        public void onComplete() {
-            if (done) {
-                return;
-            }
-            done = true;
-            if (enter()) {
-                drainLoop();
-            }
+        public boolean isDisposed() {
+            return stopWindows.get();
+        }
 
+        @Override
+        public void run() {
             if (windows.decrementAndGet() == 0) {
-                DisposableHelper.dispose(boundary);
+                DisposableHelper.dispose(upstream);
             }
+        }
 
-            actual.onComplete();
-
+        void innerNext() {
+            queue.offer(NEXT_WINDOW);
+            drain();
         }
 
-        @Override
-        public void dispose() {
-            cancelled = true;
+        void innerError(Throwable e) {
+            DisposableHelper.dispose(upstream);
+            if (errors.addThrowable(e)) {
+                done = true;
+                drain();
+            } else {
+                RxJavaPlugins.onError(e);
+            }
         }
 
-        @Override
-        public boolean isDisposed() {
-            return cancelled;
+        void innerComplete() {
+            DisposableHelper.dispose(upstream);
+            done = true;
+            drain();
         }
 
-        void drainLoop() {
-            final MpscLinkedQueue<Object> q = (MpscLinkedQueue<Object>)queue;
-            final Observer<? super Observable<T>> a = actual;
+        @SuppressWarnings("unchecked")
+        void drain() {
+            if (getAndIncrement() != 0) {
+                return;
+            }
+
             int missed = 1;
-            UnicastSubject<T> w = window;
+            Observer<? super Observable<T>> downstream = this.downstream;
+            MpscLinkedQueue<Object> queue = this.queue;
+            AtomicThrowable errors = this.errors;
+
             for (;;) {
 
                 for (;;) {
+                    if (windows.get() == 0) {
+                        queue.clear();
+                        window = null;
+                        return;
+                    }
+
+                    UnicastSubject<T> w = window;
+
                     boolean d = done;
 
-                    Object o = q.poll();
+                    if (d && errors.get() != null) {
+                        queue.clear();
+                        Throwable ex = errors.terminate();
+                        if (w != null) {
+                            window = null;
+                            w.onError(ex);
+                        }
+                        downstream.onError(ex);
+                        return;
+                    }
+
+                    Object v = queue.poll();
 
-                    boolean empty = o == null;
+                    boolean empty = v == null;
 
                     if (d && empty) {
-                        DisposableHelper.dispose(boundary);
-                        Throwable e = error;
-                        if (e != null) {
-                            w.onError(e);
+                        Throwable ex = errors.terminate();
+                        if (ex == null) {
+                            if (w != null) {
+                                window = null;
+                                w.onComplete();
+                            }
+                            downstream.onComplete();
                         } else {
-                            w.onComplete();
+                            if (w != null) {
+                                window = null;
+                                w.onError(ex);
+                            }
+                            downstream.onError(ex);
                         }
                         return;
                     }
@@ -187,48 +219,35 @@ void drainLoop() {
                         break;
                     }
 
-                    if (o == NEXT) {
-                        w.onComplete();
-
-                        if (windows.decrementAndGet() == 0) {
-                            DisposableHelper.dispose(boundary);
-                            return;
-                        }
-
-                        if (cancelled) {
-                            continue;
-                        }
-
-                        w = UnicastSubject.create(bufferSize);
+                    if (v != NEXT_WINDOW) {
+                        w.onNext((T)v);
+                        continue;
+                    }
 
-                        windows.getAndIncrement();
+                    if (w != null) {
+                        window = null;
+                        w.onComplete();
+                    }
 
+                    if (!stopWindows.get()) {
+                        w = UnicastSubject.create(capacityHint, this);
                         window = w;
+                        windows.getAndIncrement();
 
-                        a.onNext(w);
-
-                        continue;
+                        downstream.onNext(w);
                     }
-
-                    w.onNext(NotificationLite.<T>getValue(o));
                 }
 
-                missed = leave(-missed);
+                missed = addAndGet(-missed);
                 if (missed == 0) {
-                    return;
+                    break;
                 }
             }
         }
-
-        void next() {
-            queue.offer(NEXT);
-            if (enter()) {
-                drainLoop();
-            }
-        }
     }
 
     static final class WindowBoundaryInnerObserver<T, B> extends DisposableObserver<B> {
+
         final WindowBoundaryMainObserver<T, B> parent;
 
         boolean done;
@@ -242,7 +261,7 @@ public void onNext(B t) {
             if (done) {
                 return;
             }
-            parent.next();
+            parent.innerNext();
         }
 
         @Override
@@ -252,7 +271,7 @@ public void onError(Throwable t) {
                 return;
             }
             done = true;
-            parent.onError(t);
+            parent.innerError(t);
         }
 
         @Override
@@ -261,7 +280,7 @@ public void onComplete() {
                 return;
             }
             done = true;
-            parent.onComplete();
+            parent.innerComplete();
         }
     }
 }
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySupplier.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySupplier.java
index 47b0841262..c2d319032b 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySupplier.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableWindowBoundarySupplier.java
@@ -21,179 +21,213 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.functions.ObjectHelper;
-import io.reactivex.internal.observers.QueueDrainObserver;
 import io.reactivex.internal.queue.MpscLinkedQueue;
-import io.reactivex.internal.util.NotificationLite;
-import io.reactivex.observers.*;
+import io.reactivex.internal.util.AtomicThrowable;
+import io.reactivex.observers.DisposableObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.UnicastSubject;
 
 public final class ObservableWindowBoundarySupplier<T, B> extends AbstractObservableWithUpstream<T, Observable<T>> {
     final Callable<? extends ObservableSource<B>> other;
-    final int bufferSize;
+    final int capacityHint;
 
     public ObservableWindowBoundarySupplier(
             ObservableSource<T> source,
-            Callable<? extends ObservableSource<B>> other, int bufferSize) {
+            Callable<? extends ObservableSource<B>> other, int capacityHint) {
         super(source);
         this.other = other;
-        this.bufferSize = bufferSize;
+        this.capacityHint = capacityHint;
     }
 
     @Override
-    public void subscribeActual(Observer<? super Observable<T>> t) {
-        source.subscribe(new WindowBoundaryMainObserver<T, B>(new SerializedObserver<Observable<T>>(t), other, bufferSize));
+    public void subscribeActual(Observer<? super Observable<T>> observer) {
+        WindowBoundaryMainObserver<T, B> parent = new WindowBoundaryMainObserver<T, B>(observer, capacityHint, other);
+
+        source.subscribe(parent);
     }
 
     static final class WindowBoundaryMainObserver<T, B>
-    extends QueueDrainObserver<T, Object, Observable<T>>
-    implements Disposable {
+    extends AtomicInteger
+    implements Observer<T>, Disposable, Runnable {
 
-        final Callable<? extends ObservableSource<B>> other;
-        final int bufferSize;
+        private static final long serialVersionUID = 2233020065421370272L;
 
-        Disposable s;
+        final Observer<? super Observable<T>> downstream;
 
-        final AtomicReference<Disposable> boundary = new AtomicReference<Disposable>();
+        final int capacityHint;
 
-        UnicastSubject<T> window;
+        final AtomicReference<WindowBoundaryInnerObserver<T, B>> boundaryObserver;
 
-        static final Object NEXT = new Object();
+        static final WindowBoundaryInnerObserver<Object, Object> BOUNDARY_DISPOSED = new WindowBoundaryInnerObserver<Object, Object>(null);
 
-        final AtomicLong windows = new AtomicLong();
+        final AtomicInteger windows;
 
-        WindowBoundaryMainObserver(Observer<? super Observable<T>> actual, Callable<? extends ObservableSource<B>> other,
-                int bufferSize) {
-            super(actual, new MpscLinkedQueue<Object>());
-            this.other = other;
-            this.bufferSize = bufferSize;
-            windows.lazySet(1);
-        }
+        final MpscLinkedQueue<Object> queue;
 
-        @Override
-        public void onSubscribe(Disposable s) {
-            if (DisposableHelper.validate(this.s, s)) {
-                this.s = s;
+        final AtomicThrowable errors;
 
-                Observer<? super Observable<T>> a = actual;
-                a.onSubscribe(this);
+        final AtomicBoolean stopWindows;
 
-                if (cancelled) {
-                    return;
-                }
+        final Callable<? extends ObservableSource<B>> other;
 
-                ObservableSource<B> p;
+        static final Object NEXT_WINDOW = new Object();
 
-                try {
-                    p = ObjectHelper.requireNonNull(other.call(), "The first window ObservableSource supplied is null");
-                } catch (Throwable e) {
-                    Exceptions.throwIfFatal(e);
-                    s.dispose();
-                    a.onError(e);
-                    return;
-                }
+        Disposable upstream;
 
-                UnicastSubject<T> w = UnicastSubject.create(bufferSize);
+        volatile boolean done;
 
-                window = w;
-
-                a.onNext(w);
+        UnicastSubject<T> window;
 
-                WindowBoundaryInnerObserver<T, B> inner = new WindowBoundaryInnerObserver<T, B>(this);
+        WindowBoundaryMainObserver(Observer<? super Observable<T>> downstream, int capacityHint, Callable<? extends ObservableSource<B>> other) {
+            this.downstream = downstream;
+            this.capacityHint = capacityHint;
+            this.boundaryObserver = new AtomicReference<WindowBoundaryInnerObserver<T, B>>();
+            this.windows = new AtomicInteger(1);
+            this.queue = new MpscLinkedQueue<Object>();
+            this.errors = new AtomicThrowable();
+            this.stopWindows = new AtomicBoolean();
+            this.other = other;
+        }
 
-                if (boundary.compareAndSet(null, inner)) {
-                    windows.getAndIncrement();
-                    p.subscribe(inner);
-                }
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(upstream, d)) {
+                upstream = d;
+                downstream.onSubscribe(this);
+                queue.offer(NEXT_WINDOW);
+                drain();
             }
         }
 
         @Override
         public void onNext(T t) {
-            if (fastEnter()) {
-                UnicastSubject<T> w = window;
-
-                w.onNext(t);
+            queue.offer(t);
+            drain();
+        }
 
-                if (leave(-1) == 0) {
-                    return;
-                }
+        @Override
+        public void onError(Throwable e) {
+            disposeBoundary();
+            if (errors.addThrowable(e)) {
+                done = true;
+                drain();
             } else {
-                queue.offer(NotificationLite.next(t));
-                if (!enter()) {
-                    return;
-                }
+                RxJavaPlugins.onError(e);
             }
-            drainLoop();
         }
 
         @Override
-        public void onError(Throwable t) {
-            if (done) {
-                RxJavaPlugins.onError(t);
-                return;
-            }
-            error = t;
+        public void onComplete() {
+            disposeBoundary();
             done = true;
-            if (enter()) {
-                drainLoop();
-            }
+            drain();
+        }
 
-            if (windows.decrementAndGet() == 0) {
-                DisposableHelper.dispose(boundary);
+        @Override
+        public void dispose() {
+            if (stopWindows.compareAndSet(false, true)) {
+                disposeBoundary();
+                if (windows.decrementAndGet() == 0) {
+                    upstream.dispose();
+                }
             }
+        }
 
-            actual.onError(t);
+        @SuppressWarnings({ "rawtypes", "unchecked" })
+        void disposeBoundary() {
+            Disposable d = boundaryObserver.getAndSet((WindowBoundaryInnerObserver)BOUNDARY_DISPOSED);
+            if (d != null && d != BOUNDARY_DISPOSED) {
+                d.dispose();
+            }
         }
 
         @Override
-        public void onComplete() {
-            if (done) {
-                return;
-            }
-            done = true;
-            if (enter()) {
-                drainLoop();
-            }
+        public boolean isDisposed() {
+            return stopWindows.get();
+        }
 
+        @Override
+        public void run() {
             if (windows.decrementAndGet() == 0) {
-                DisposableHelper.dispose(boundary);
+                upstream.dispose();
             }
+        }
 
-            actual.onComplete();
-
+        void innerNext(WindowBoundaryInnerObserver<T, B> sender) {
+            boundaryObserver.compareAndSet(sender, null);
+            queue.offer(NEXT_WINDOW);
+            drain();
         }
 
-        @Override
-        public void dispose() {
-            cancelled = true;
+        void innerError(Throwable e) {
+            upstream.dispose();
+            if (errors.addThrowable(e)) {
+                done = true;
+                drain();
+            } else {
+                RxJavaPlugins.onError(e);
+            }
         }
 
-        @Override
-        public boolean isDisposed() {
-            return cancelled;
+        void innerComplete() {
+            upstream.dispose();
+            done = true;
+            drain();
         }
 
-        void drainLoop() {
-            final MpscLinkedQueue<Object> q = (MpscLinkedQueue<Object>)queue;
-            final Observer<? super Observable<T>> a = actual;
+        @SuppressWarnings("unchecked")
+        void drain() {
+            if (getAndIncrement() != 0) {
+                return;
+            }
+
             int missed = 1;
-            UnicastSubject<T> w = window;
+            Observer<? super Observable<T>> downstream = this.downstream;
+            MpscLinkedQueue<Object> queue = this.queue;
+            AtomicThrowable errors = this.errors;
+
             for (;;) {
 
                 for (;;) {
+                    if (windows.get() == 0) {
+                        queue.clear();
+                        window = null;
+                        return;
+                    }
+
+                    UnicastSubject<T> w = window;
+
                     boolean d = done;
 
-                    Object o = q.poll();
-                    boolean empty = o == null;
+                    if (d && errors.get() != null) {
+                        queue.clear();
+                        Throwable ex = errors.terminate();
+                        if (w != null) {
+                            window = null;
+                            w.onError(ex);
+                        }
+                        downstream.onError(ex);
+                        return;
+                    }
+
+                    Object v = queue.poll();
+
+                    boolean empty = v == null;
 
                     if (d && empty) {
-                        DisposableHelper.dispose(boundary);
-                        Throwable e = error;
-                        if (e != null) {
-                            w.onError(e);
+                        Throwable ex = errors.terminate();
+                        if (ex == null) {
+                            if (w != null) {
+                                window = null;
+                                w.onComplete();
+                            }
+                            downstream.onComplete();
                         } else {
-                            w.onComplete();
+                            if (w != null) {
+                                window = null;
+                                w.onError(ex);
+                            }
+                            downstream.onError(ex);
                         }
                         return;
                     }
@@ -202,62 +236,48 @@ void drainLoop() {
                         break;
                     }
 
-                    if (o == NEXT) {
-                        w.onComplete();
+                    if (v != NEXT_WINDOW) {
+                        w.onNext((T)v);
+                        continue;
+                    }
 
-                        if (windows.decrementAndGet() == 0) {
-                            DisposableHelper.dispose(boundary);
-                            return;
-                        }
+                    if (w != null) {
+                        window = null;
+                        w.onComplete();
+                    }
 
-                        if (cancelled) {
-                            continue;
-                        }
+                    if (!stopWindows.get()) {
+                        w = UnicastSubject.create(capacityHint, this);
+                        window = w;
+                        windows.getAndIncrement();
 
-                        ObservableSource<B> p;
+                        ObservableSource<B> otherSource;
 
                         try {
-                            p = ObjectHelper.requireNonNull(other.call(), "The ObservableSource supplied is null");
-                        } catch (Throwable e) {
-                            Exceptions.throwIfFatal(e);
-                            DisposableHelper.dispose(boundary);
-                            a.onError(e);
-                            return;
+                            otherSource = ObjectHelper.requireNonNull(other.call(), "The other Callable returned a null ObservableSource");
+                        } catch (Throwable ex) {
+                            Exceptions.throwIfFatal(ex);
+                            errors.addThrowable(ex);
+                            done = true;
+                            continue;
                         }
 
-                        w = UnicastSubject.create(bufferSize);
+                        WindowBoundaryInnerObserver<T, B> bo = new WindowBoundaryInnerObserver<T, B>(this);
 
-                        windows.getAndIncrement();
+                        if (boundaryObserver.compareAndSet(null, bo)) {
+                            otherSource.subscribe(bo);
 
-                        window = w;
-
-                        a.onNext(w);
-
-                        WindowBoundaryInnerObserver<T, B> b = new WindowBoundaryInnerObserver<T, B>(this);
-
-                        if (boundary.compareAndSet(boundary.get(), b)) {
-                            p.subscribe(b);
+                            downstream.onNext(w);
                         }
-
-                        continue;
                     }
-
-                    w.onNext(NotificationLite.<T>getValue(o));
                 }
 
-                missed = leave(-missed);
+                missed = addAndGet(-missed);
                 if (missed == 0) {
-                    return;
+                    break;
                 }
             }
         }
-
-        void next() {
-            queue.offer(NEXT);
-            if (enter()) {
-                drainLoop();
-            }
-        }
     }
 
     static final class WindowBoundaryInnerObserver<T, B> extends DisposableObserver<B> {
@@ -276,7 +296,7 @@ public void onNext(B t) {
             }
             done = true;
             dispose();
-            parent.next();
+            parent.innerNext(this);
         }
 
         @Override
@@ -286,7 +306,7 @@ public void onError(Throwable t) {
                 return;
             }
             done = true;
-            parent.onError(t);
+            parent.innerError(t);
         }
 
         @Override
@@ -295,7 +315,7 @@ public void onComplete() {
                 return;
             }
             done = true;
-            parent.onComplete();
+            parent.innerComplete();
         }
     }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableWindowWithObservableTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableWindowWithObservableTest.java
index 8319a74d97..6c084f01d6 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableWindowWithObservableTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableWindowWithObservableTest.java
@@ -19,17 +19,19 @@
 
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.*;
 
 import org.junit.Test;
 
 import io.reactivex.*;
 import io.reactivex.Observable;
 import io.reactivex.Observer;
+import io.reactivex.disposables.*;
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.functions.Functions;
 import io.reactivex.observers.*;
+import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.*;
 
 public class ObservableWindowWithObservableTest {
@@ -346,8 +348,8 @@ public Observable<Integer> call() {
 
         boundary.onComplete();
 
-        // FIXME source still active because the open window
-        assertTrue(source.hasObservers());
+
+        assertFalse(source.hasObservers());
         assertFalse(boundary.hasObservers());
 
         ts.assertComplete();
@@ -374,10 +376,13 @@ public Observable<Integer> call() {
 
         ts.dispose();
 
-        // FIXME source has subscribers because the open window
         assertTrue(source.hasObservers());
-        // FIXME boundary has subscribers because the open window
-        assertTrue(boundary.hasObservers());
+
+        assertFalse(boundary.hasObservers());
+
+        ts.values().get(0).test(true);
+
+        assertFalse(source.hasObservers());
 
         ts.assertNotComplete();
         ts.assertNoErrors();
@@ -638,4 +643,627 @@ public Observable<Observable<Object>> apply(Observable<Object> f)
             }
         });
     }
+
+    @Test
+    public void upstreamDisposedWhenOutputsDisposed() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        PublishSubject<Integer> boundary = PublishSubject.create();
+
+        TestObserver<Integer> to = source.window(boundary)
+        .take(1)
+        .flatMap(new Function<Observable<Integer>, ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> apply(
+                    Observable<Integer> w) throws Exception {
+                return w.take(1);
+            }
+        })
+        .test();
+
+        source.onNext(1);
+
+        assertFalse("source not disposed", source.hasObservers());
+        assertFalse("boundary not disposed", boundary.hasObservers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void mainAndBoundaryBothError() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+            TestObserver<Observable<Object>> to = Observable.error(new TestException("main"))
+            .window(new Observable<Object>() {
+                @Override
+                protected void subscribeActual(Observer<? super Object> observer) {
+                    observer.onSubscribe(Disposables.empty());
+                    ref.set(observer);
+                }
+            })
+            .test();
+
+            to
+            .assertValueCount(1)
+            .assertError(TestException.class)
+            .assertErrorMessage("main")
+            .assertNotComplete();
+
+            ref.get().onError(new TestException("inner"));
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class, "inner");
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void mainCompleteBoundaryErrorRace() {
+        final TestException ex = new TestException();
+
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+            List<Throwable> errors = TestHelper.trackPluginErrors();
+            try {
+                final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+                final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+                TestObserver<Observable<Object>> to = new Observable<Object>() {
+                    @Override
+                    protected void subscribeActual(Observer<? super Object> observer) {
+                        observer.onSubscribe(Disposables.empty());
+                        refMain.set(observer);
+                    }
+                }
+                .window(new Observable<Object>() {
+                    @Override
+                    protected void subscribeActual(Observer<? super Object> observer) {
+                        observer.onSubscribe(Disposables.empty());
+                        ref.set(observer);
+                    }
+                })
+                .test();
+
+                Runnable r1 = new Runnable() {
+                    @Override
+                    public void run() {
+                        refMain.get().onComplete();
+                    }
+                };
+                Runnable r2 = new Runnable() {
+                    @Override
+                    public void run() {
+                        ref.get().onError(ex);
+                    }
+                };
+
+                TestHelper.race(r1, r2);
+
+                to
+                .assertValueCount(1)
+                .assertTerminated();
+
+                if (!errors.isEmpty()) {
+                    TestHelper.assertUndeliverable(errors, 0, TestException.class);
+                }
+            } finally {
+                RxJavaPlugins.reset();
+            }
+        }
+    }
+
+    @Test
+    public void mainNextBoundaryNextRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+            final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+            final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+            TestObserver<Observable<Object>> to = new Observable<Object>() {
+                @Override
+                protected void subscribeActual(Observer<? super Object> observer) {
+                    observer.onSubscribe(Disposables.empty());
+                    refMain.set(observer);
+                }
+            }
+            .window(new Observable<Object>() {
+                @Override
+                protected void subscribeActual(Observer<? super Object> observer) {
+                    observer.onSubscribe(Disposables.empty());
+                    ref.set(observer);
+                }
+            })
+            .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    refMain.get().onNext(1);
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    ref.get().onNext(1);
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            to
+            .assertValueCount(2)
+            .assertNotComplete()
+            .assertNoErrors();
+        }
+    }
+
+    @Test
+    public void takeOneAnotherBoundary() {
+        final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+        final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+        TestObserver<Observable<Object>> to = new Observable<Object>() {
+            @Override
+            protected void subscribeActual(Observer<? super Object> observer) {
+                observer.onSubscribe(Disposables.empty());
+                refMain.set(observer);
+            }
+        }
+        .window(new Observable<Object>() {
+            @Override
+            protected void subscribeActual(Observer<? super Object> observer) {
+                observer.onSubscribe(Disposables.empty());
+                ref.set(observer);
+            }
+        })
+        .test();
+
+        to.assertValueCount(1)
+        .assertNotTerminated()
+        .cancel();
+
+        ref.get().onNext(1);
+
+        to.assertValueCount(1)
+        .assertNotTerminated();
+    }
+
+    @Test
+    public void disposeMainBoundaryCompleteRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+            final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+            final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+            final TestObserver<Observable<Object>> to = new Observable<Object>() {
+                 @Override
+                 protected void subscribeActual(Observer<? super Object> observer) {
+                     observer.onSubscribe(Disposables.empty());
+                     refMain.set(observer);
+                 }
+             }
+             .window(new Observable<Object>() {
+                 @Override
+                 protected void subscribeActual(Observer<? super Object> observer) {
+                     final AtomicInteger counter = new AtomicInteger();
+                     observer.onSubscribe(new Disposable() {
+
+                         @Override
+                         public void dispose() {
+                             // about a microsecond
+                             for (int i = 0; i < 100; i++) {
+                                 counter.incrementAndGet();
+                             }
+                         }
+
+                         @Override
+                         public boolean isDisposed() {
+                             return false;
+                         }
+                      });
+                     ref.set(observer);
+                 }
+             })
+             .test();
+
+             Runnable r1 = new Runnable() {
+                 @Override
+                 public void run() {
+                     to.cancel();
+                 }
+             };
+             Runnable r2 = new Runnable() {
+                 @Override
+                 public void run() {
+                     Observer<Object> o = ref.get();
+                     o.onNext(1);
+                     o.onComplete();
+                 }
+             };
+
+             TestHelper.race(r1, r2);
+        }
+    }
+
+    @Test
+    public void disposeMainBoundaryErrorRace() {
+        final TestException ex = new TestException();
+
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+           final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+           final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+           final TestObserver<Observable<Object>> to = new Observable<Object>() {
+               @Override
+               protected void subscribeActual(Observer<? super Object> observer) {
+                   observer.onSubscribe(Disposables.empty());
+                   refMain.set(observer);
+               }
+           }
+           .window(new Observable<Object>() {
+               @Override
+               protected void subscribeActual(Observer<? super Object> observer) {
+                   final AtomicInteger counter = new AtomicInteger();
+                   observer.onSubscribe(new Disposable() {
+
+                       @Override
+                       public void dispose() {
+                           // about a microsecond
+                           for (int i = 0; i < 100; i++) {
+                               counter.incrementAndGet();
+                           }
+                       }
+
+                       @Override
+                       public boolean isDisposed() {
+                           return false;
+                       }
+                    });
+                   ref.set(observer);
+               }
+           })
+           .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    to.cancel();
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    Observer<Object> o = ref.get();
+                    o.onNext(1);
+                    o.onError(ex);
+                }
+            };
+
+            TestHelper.race(r1, r2);
+        }
+    }
+
+    @Test
+    public void boundarySupplierDoubleOnSubscribe() {
+        TestHelper.checkDoubleOnSubscribeObservable(new Function<Observable<Object>, Observable<Observable<Object>>>() {
+            @Override
+            public Observable<Observable<Object>> apply(Observable<Object> f)
+                    throws Exception {
+                return f.window(Functions.justCallable(Observable.never())).takeLast(1);
+            }
+        });
+    }
+
+    @Test
+    public void selectorUpstreamDisposedWhenOutputsDisposed() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        PublishSubject<Integer> boundary = PublishSubject.create();
+
+        TestObserver<Integer> to = source.window(Functions.justCallable(boundary))
+        .take(1)
+        .flatMap(new Function<Observable<Integer>, ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> apply(
+                    Observable<Integer> w) throws Exception {
+                return w.take(1);
+            }
+        })
+        .test();
+
+        source.onNext(1);
+
+        assertFalse("source not disposed", source.hasObservers());
+        assertFalse("boundary not disposed", boundary.hasObservers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void supplierMainAndBoundaryBothError() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+            TestObserver<Observable<Object>> to = Observable.error(new TestException("main"))
+            .window(Functions.justCallable(new Observable<Object>() {
+                @Override
+                protected void subscribeActual(Observer<? super Object> observer) {
+                    observer.onSubscribe(Disposables.empty());
+                    ref.set(observer);
+                }
+            }))
+            .test();
+
+            to
+            .assertValueCount(1)
+            .assertError(TestException.class)
+            .assertErrorMessage("main")
+            .assertNotComplete();
+
+            ref.get().onError(new TestException("inner"));
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class, "inner");
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void supplierMainCompleteBoundaryErrorRace() {
+        final TestException ex = new TestException();
+
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+            List<Throwable> errors = TestHelper.trackPluginErrors();
+            try {
+                final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+                final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+                TestObserver<Observable<Object>> to = new Observable<Object>() {
+                    @Override
+                    protected void subscribeActual(Observer<? super Object> observer) {
+                        observer.onSubscribe(Disposables.empty());
+                        refMain.set(observer);
+                    }
+                }
+                .window(Functions.justCallable(new Observable<Object>() {
+                    @Override
+                    protected void subscribeActual(Observer<? super Object> observer) {
+                        observer.onSubscribe(Disposables.empty());
+                        ref.set(observer);
+                    }
+                }))
+                .test();
+
+                Runnable r1 = new Runnable() {
+                    @Override
+                    public void run() {
+                        refMain.get().onComplete();
+                    }
+                };
+                Runnable r2 = new Runnable() {
+                    @Override
+                    public void run() {
+                        ref.get().onError(ex);
+                    }
+                };
+
+                TestHelper.race(r1, r2);
+
+                to
+                .assertValueCount(1)
+                .assertTerminated();
+
+                if (!errors.isEmpty()) {
+                    TestHelper.assertUndeliverable(errors, 0, TestException.class);
+                }
+            } finally {
+                RxJavaPlugins.reset();
+            }
+        }
+    }
+
+    @Test
+    public void supplierMainNextBoundaryNextRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+            final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+            final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+            TestObserver<Observable<Object>> to = new Observable<Object>() {
+                @Override
+                protected void subscribeActual(Observer<? super Object> observer) {
+                    observer.onSubscribe(Disposables.empty());
+                    refMain.set(observer);
+                }
+            }
+            .window(Functions.justCallable(new Observable<Object>() {
+                @Override
+                protected void subscribeActual(Observer<? super Object> observer) {
+                    observer.onSubscribe(Disposables.empty());
+                    ref.set(observer);
+                }
+            }))
+            .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    refMain.get().onNext(1);
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    ref.get().onNext(1);
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            to
+            .assertValueCount(2)
+            .assertNotComplete()
+            .assertNoErrors();
+        }
+    }
+
+    @Test
+    public void supplierTakeOneAnotherBoundary() {
+        final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+        final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+        TestObserver<Observable<Object>> to = new Observable<Object>() {
+            @Override
+            protected void subscribeActual(Observer<? super Object> observer) {
+                observer.onSubscribe(Disposables.empty());
+                refMain.set(observer);
+            }
+        }
+        .window(Functions.justCallable(new Observable<Object>() {
+            @Override
+            protected void subscribeActual(Observer<? super Object> observer) {
+                observer.onSubscribe(Disposables.empty());
+                ref.set(observer);
+            }
+        }))
+        .test();
+
+        to.assertValueCount(1)
+        .assertNotTerminated()
+        .cancel();
+
+        ref.get().onNext(1);
+
+        to.assertValueCount(1)
+        .assertNotTerminated();
+    }
+
+    @Test
+    public void supplierDisposeMainBoundaryCompleteRace() {
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+            final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+            final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+            final TestObserver<Observable<Object>> to = new Observable<Object>() {
+                 @Override
+                 protected void subscribeActual(Observer<? super Object> observer) {
+                     observer.onSubscribe(Disposables.empty());
+                     refMain.set(observer);
+                 }
+             }
+             .window(Functions.justCallable(new Observable<Object>() {
+                 @Override
+                 protected void subscribeActual(Observer<? super Object> observer) {
+                     final AtomicInteger counter = new AtomicInteger();
+                     observer.onSubscribe(new Disposable() {
+
+                         @Override
+                         public void dispose() {
+                             // about a microsecond
+                             for (int i = 0; i < 100; i++) {
+                                 counter.incrementAndGet();
+                             }
+                         }
+
+                         @Override
+                         public boolean isDisposed() {
+                             return false;
+                         }
+                      });
+                     ref.set(observer);
+                 }
+             }))
+             .test();
+
+             Runnable r1 = new Runnable() {
+                 @Override
+                 public void run() {
+                     to.cancel();
+                 }
+             };
+             Runnable r2 = new Runnable() {
+                 @Override
+                 public void run() {
+                     Observer<Object> o = ref.get();
+                     o.onNext(1);
+                     o.onComplete();
+                 }
+             };
+
+             TestHelper.race(r1, r2);
+        }
+    }
+
+    @Test
+    public void supplierDisposeMainBoundaryErrorRace() {
+        final TestException ex = new TestException();
+
+        for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
+            List<Throwable> errors = TestHelper.trackPluginErrors();
+            try {
+                final AtomicReference<Observer<? super Object>> refMain = new AtomicReference<Observer<? super Object>>();
+                final AtomicReference<Observer<? super Object>> ref = new AtomicReference<Observer<? super Object>>();
+
+                final TestObserver<Observable<Object>> to = new Observable<Object>() {
+                    @Override
+                    protected void subscribeActual(Observer<? super Object> observer) {
+                        observer.onSubscribe(Disposables.empty());
+                        refMain.set(observer);
+                    }
+                }
+                .window(new Callable<ObservableSource<Object>>() {
+                    int count;
+                    @Override
+                    public ObservableSource<Object> call() throws Exception {
+                        if (++count > 1) {
+                            return Observable.never();
+                        }
+                        return (new Observable<Object>() {
+                            @Override
+                            protected void subscribeActual(Observer<? super Object> observer) {
+                                final AtomicInteger counter = new AtomicInteger();
+                                observer.onSubscribe(new Disposable() {
+
+                                    @Override
+                                    public void dispose() {
+                                        // about a microsecond
+                                        for (int i = 0; i < 100; i++) {
+                                            counter.incrementAndGet();
+                                        }
+                                    }
+
+                                    @Override
+                                    public boolean isDisposed() {
+                                        return false;
+                                    }
+                                });
+                                ref.set(observer);
+                            }
+                        });
+                    }
+                })
+                .test();
+
+                Runnable r1 = new Runnable() {
+                    @Override
+                    public void run() {
+                        to.cancel();
+                    }
+                };
+                Runnable r2 = new Runnable() {
+                    @Override
+                    public void run() {
+                        Observer<Object> o = ref.get();
+                        o.onNext(1);
+                        o.onError(ex);
+                    }
+                };
+
+                TestHelper.race(r1, r2);
+
+                if (!errors.isEmpty()) {
+                    TestHelper.assertUndeliverable(errors, 0, TestException.class);
+                }
+            } finally {
+                RxJavaPlugins.reset();
+            }
+        }
+    }
 }