diff --git a/build.gradle b/build.gradle
index 12a5b04978..9da4ae5757 100644
--- a/build.gradle
+++ b/build.gradle
@@ -57,6 +57,7 @@ def mockitoVersion = "2.1.0"
 def jmhLibVersion = "1.19"
 def testNgVersion = "6.11"
 def guavaVersion = "24.0-jre"
+def jacocoVersion = "0.8.0"
 // --------------------------------------
 
 repositories {
@@ -257,7 +258,7 @@ task testng(type: Test) {
 check.dependsOn testng 
 
 jacoco {
-    toolVersion = "0.7.9" // See http://www.eclemma.org/jacoco/.
+    toolVersion = jacocoVersion // See http://www.eclemma.org/jacoco/.
 }
 
 task GCandMem(dependsOn: "check") doLast {
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundary.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundary.java
index 58cec89232..8c8164f76f 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundary.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundary.java
@@ -17,186 +17,210 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.*;
 import io.reactivex.exceptions.MissingBackpressureException;
-import io.reactivex.internal.disposables.DisposableHelper;
-import io.reactivex.internal.fuseable.SimplePlainQueue;
 import io.reactivex.internal.queue.MpscLinkedQueue;
-import io.reactivex.internal.subscribers.QueueDrainSubscriber;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
-import io.reactivex.internal.util.NotificationLite;
+import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.UnicastProcessor;
-import io.reactivex.subscribers.*;
+import io.reactivex.subscribers.DisposableSubscriber;
 
 public final class FlowableWindowBoundary<T, B> extends AbstractFlowableWithUpstream<T, Flowable<T>> {
     final Publisher<B> other;
-    final int bufferSize;
+    final int capacityHint;
 
-    public FlowableWindowBoundary(Flowable<T> source, Publisher<B> other, int bufferSize) {
+    public FlowableWindowBoundary(Flowable<T> source, Publisher<B> other, int capacityHint) {
         super(source);
         this.other = other;
-        this.bufferSize = bufferSize;
+        this.capacityHint = capacityHint;
     }
 
     @Override
-    protected void subscribeActual(Subscriber<? super Flowable<T>> s) {
-        source.subscribe(
-                new WindowBoundaryMainSubscriber<T, B>(
-                        new SerializedSubscriber<Flowable<T>>(s), other, bufferSize));
+    protected void subscribeActual(Subscriber<? super Flowable<T>> subscriber) {
+        WindowBoundaryMainSubscriber<T, B> parent = new WindowBoundaryMainSubscriber<T, B>(subscriber, capacityHint);
+
+        subscriber.onSubscribe(parent);
+
+        parent.innerNext();
+
+        other.subscribe(parent.boundarySubscriber);
+
+        source.subscribe(parent);
     }
 
     static final class WindowBoundaryMainSubscriber<T, B>
-    extends QueueDrainSubscriber<T, Object, Flowable<T>>
-    implements Subscription {
+    extends AtomicInteger
+    implements FlowableSubscriber<T>, Subscription, Runnable {
 
-        final Publisher<B> other;
-        final int bufferSize;
+        private static final long serialVersionUID = 2233020065421370272L;
 
-        Subscription s;
+        final Subscriber<? super Flowable<T>> downstream;
 
-        final AtomicReference<Disposable> boundary = new AtomicReference<Disposable>();
+        final int capacityHint;
 
-        UnicastProcessor<T> window;
+        final WindowBoundaryInnerSubscriber<T, B> boundarySubscriber;
 
-        static final Object NEXT = new Object();
+        final AtomicReference<Subscription> upstream;
 
-        final AtomicLong windows = new AtomicLong();
+        final AtomicInteger windows;
 
-        WindowBoundaryMainSubscriber(Subscriber<? super Flowable<T>> actual, Publisher<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(Subscription s) {
-            if (SubscriptionHelper.validate(this.s, s)) {
-                this.s = s;
+        final AtomicThrowable errors;
 
-                Subscriber<? super Flowable<T>> a = actual;
-                a.onSubscribe(this);
+        final AtomicBoolean stopWindows;
 
-                if (cancelled) {
-                    return;
-                }
+        final AtomicLong requested;
 
-                UnicastProcessor<T> w = UnicastProcessor.<T>create(bufferSize);
+        static final Object NEXT_WINDOW = new Object();
 
-                long r = requested();
-                if (r != 0L) {
-                    a.onNext(w);
-                    if (r != Long.MAX_VALUE) {
-                        produced(1);
-                    }
-                } else {
-                    s.cancel();
-                    a.onError(new MissingBackpressureException("Could not deliver first window due to lack of requests"));
-                    return;
-                }
+        volatile boolean done;
 
-                window = w;
+        UnicastProcessor<T> window;
 
-                WindowBoundaryInnerSubscriber<T, B> inner = new WindowBoundaryInnerSubscriber<T, B>(this);
+        long emitted;
+
+        WindowBoundaryMainSubscriber(Subscriber<? super Flowable<T>> downstream, int capacityHint) {
+            this.downstream = downstream;
+            this.capacityHint = capacityHint;
+            this.boundarySubscriber = new WindowBoundaryInnerSubscriber<T, B>(this);
+            this.upstream = new AtomicReference<Subscription>();
+            this.windows = new AtomicInteger(1);
+            this.queue = new MpscLinkedQueue<Object>();
+            this.errors = new AtomicThrowable();
+            this.stopWindows = new AtomicBoolean();
+            this.requested = new AtomicLong();
+        }
 
-                if (boundary.compareAndSet(null, inner)) {
-                    windows.getAndIncrement();
-                    s.request(Long.MAX_VALUE);
-                    other.subscribe(inner);
-                }
+        @Override
+        public void onSubscribe(Subscription d) {
+            if (SubscriptionHelper.setOnce(upstream, d)) {
+                d.request(Long.MAX_VALUE);
             }
         }
 
         @Override
         public void onNext(T t) {
-            if (fastEnter()) {
-                UnicastProcessor<T> w = window;
-
-                w.onNext(t);
+            queue.offer(t);
+            drain();
+        }
 
-                if (leave(-1) == 0) {
-                    return;
-                }
+        @Override
+        public void onError(Throwable e) {
+            boundarySubscriber.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() {
+            boundarySubscriber.dispose();
             done = true;
-            if (enter()) {
-                drainLoop();
-            }
+            drain();
+        }
 
-            if (windows.decrementAndGet() == 0) {
-                DisposableHelper.dispose(boundary);
+        @Override
+        public void cancel() {
+            if (stopWindows.compareAndSet(false, true)) {
+                boundarySubscriber.dispose();
+                if (windows.decrementAndGet() == 0) {
+                    SubscriptionHelper.cancel(upstream);
+                }
             }
-
-            actual.onError(t);
         }
 
         @Override
-        public void onComplete() {
-            if (done) {
-                return;
-            }
-            done = true;
-            if (enter()) {
-                drainLoop();
-            }
+        public void request(long n) {
+            BackpressureHelper.add(requested, n);
+        }
 
+        @Override
+        public void run() {
             if (windows.decrementAndGet() == 0) {
-                DisposableHelper.dispose(boundary);
+                SubscriptionHelper.cancel(upstream);
             }
+        }
 
-            actual.onComplete();
-
+        void innerNext() {
+            queue.offer(NEXT_WINDOW);
+            drain();
         }
 
-        @Override
-        public void request(long n) {
-            requested(n);
+        void innerError(Throwable e) {
+            SubscriptionHelper.cancel(upstream);
+            if (errors.addThrowable(e)) {
+                done = true;
+                drain();
+            } else {
+                RxJavaPlugins.onError(e);
+            }
         }
 
-        @Override
-        public void cancel() {
-            cancelled = true;
+        void innerComplete() {
+            SubscriptionHelper.cancel(upstream);
+            done = true;
+            drain();
         }
 
-        void drainLoop() {
-            final SimplePlainQueue<Object> q = queue;
-            final Subscriber<? super Flowable<T>> a = actual;
+        @SuppressWarnings("unchecked")
+        void drain() {
+            if (getAndIncrement() != 0) {
+                return;
+            }
+
             int missed = 1;
-            UnicastProcessor<T> w = window;
+            Subscriber<? super Flowable<T>> downstream = this.downstream;
+            MpscLinkedQueue<Object> queue = this.queue;
+            AtomicThrowable errors = this.errors;
+            long emitted = this.emitted;
+
             for (;;) {
 
                 for (;;) {
+                    if (windows.get() == 0) {
+                        queue.clear();
+                        window = null;
+                        return;
+                    }
+
+                    UnicastProcessor<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;
+                    }
 
-                    boolean empty = o == null;
+                    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;
                     }
@@ -205,59 +229,44 @@ void drainLoop() {
                         break;
                     }
 
-                    if (o == NEXT) {
-                        w.onComplete();
-
-                        if (windows.decrementAndGet() == 0) {
-                            DisposableHelper.dispose(boundary);
-                            return;
-                        }
-
-                        if (cancelled) {
-                            continue;
-                        }
+                    if (v != NEXT_WINDOW) {
+                        w.onNext((T)v);
+                        continue;
+                    }
 
-                        w = UnicastProcessor.<T>create(bufferSize);
+                    if (w != null) {
+                        window = null;
+                        w.onComplete();
+                    }
 
-                        long r = requested();
-                        if (r != 0L) {
-                            windows.getAndIncrement();
+                    if (!stopWindows.get()) {
+                        w = UnicastProcessor.create(capacityHint, this);
+                        window = w;
+                        windows.getAndIncrement();
 
-                            a.onNext(w);
-                            if (r != Long.MAX_VALUE) {
-                                produced(1);
-                            }
+                        if (emitted != requested.get()) {
+                            emitted++;
+                            downstream.onNext(w);
                         } else {
-                            // don't emit new windows
-                            cancelled = true;
-                            a.onError(new MissingBackpressureException("Could not deliver new window due to lack of requests"));
-                            continue;
+                            SubscriptionHelper.cancel(upstream);
+                            boundarySubscriber.dispose();
+                            errors.addThrowable(new MissingBackpressureException("Could not deliver a window due to lack of requests"));
+                            done = true;
                         }
-
-                        window = w;
-                        continue;
                     }
-
-                    w.onNext(NotificationLite.<T>getValue(o));
                 }
 
-                missed = leave(-missed);
+                this.emitted = emitted;
+                missed = addAndGet(-missed);
                 if (missed == 0) {
-                    return;
+                    break;
                 }
             }
         }
-
-        void next() {
-            queue.offer(NEXT);
-            if (enter()) {
-                drainLoop();
-            }
-        }
-
     }
 
     static final class WindowBoundaryInnerSubscriber<T, B> extends DisposableSubscriber<B> {
+
         final WindowBoundaryMainSubscriber<T, B> parent;
 
         boolean done;
@@ -271,7 +280,7 @@ public void onNext(B t) {
             if (done) {
                 return;
             }
-            parent.next();
+            parent.innerNext();
         }
 
         @Override
@@ -281,7 +290,7 @@ public void onError(Throwable t) {
                 return;
             }
             done = true;
-            parent.onError(t);
+            parent.innerError(t);
         }
 
         @Override
@@ -290,7 +299,7 @@ public void onComplete() {
                 return;
             }
             done = true;
-            parent.onComplete();
+            parent.innerComplete();
         }
     }
 }
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundarySupplier.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundarySupplier.java
index f90d74ad2a..faaf1e7384 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundarySupplier.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundarySupplier.java
@@ -18,201 +18,224 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.Flowable;
+import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.*;
-import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.functions.ObjectHelper;
-import io.reactivex.internal.fuseable.SimplePlainQueue;
 import io.reactivex.internal.queue.MpscLinkedQueue;
-import io.reactivex.internal.subscribers.QueueDrainSubscriber;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
-import io.reactivex.internal.util.NotificationLite;
+import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.UnicastProcessor;
-import io.reactivex.subscribers.*;
+import io.reactivex.subscribers.DisposableSubscriber;
 
 public final class FlowableWindowBoundarySupplier<T, B> extends AbstractFlowableWithUpstream<T, Flowable<T>> {
     final Callable<? extends Publisher<B>> other;
-    final int bufferSize;
+    final int capacityHint;
 
     public FlowableWindowBoundarySupplier(Flowable<T> source,
-            Callable<? extends Publisher<B>> other, int bufferSize) {
+            Callable<? extends Publisher<B>> other, int capacityHint) {
         super(source);
         this.other = other;
-        this.bufferSize = bufferSize;
+        this.capacityHint = capacityHint;
     }
 
     @Override
-    protected void subscribeActual(Subscriber<? super Flowable<T>> s) {
-        source.subscribe(new WindowBoundaryMainSubscriber<T, B>(
-                new SerializedSubscriber<Flowable<T>>(s), other, bufferSize));
+    protected void subscribeActual(Subscriber<? super Flowable<T>> subscriber) {
+        WindowBoundaryMainSubscriber<T, B> parent = new WindowBoundaryMainSubscriber<T, B>(subscriber, capacityHint, other);
+
+        source.subscribe(parent);
     }
 
     static final class WindowBoundaryMainSubscriber<T, B>
-    extends QueueDrainSubscriber<T, Object, Flowable<T>>
-    implements Subscription {
+    extends AtomicInteger
+    implements FlowableSubscriber<T>, Subscription, Runnable {
 
-        final Callable<? extends Publisher<B>> other;
-        final int bufferSize;
+        private static final long serialVersionUID = 2233020065421370272L;
 
-        Subscription s;
+        final Subscriber<? super Flowable<T>> downstream;
 
-        final AtomicReference<Disposable> boundary = new AtomicReference<Disposable>();
+        final int capacityHint;
 
-        UnicastProcessor<T> window;
+        final AtomicReference<WindowBoundaryInnerSubscriber<T, B>> boundarySubscriber;
 
-        static final Object NEXT = new Object();
+        static final WindowBoundaryInnerSubscriber<Object, Object> BOUNDARY_DISPOSED = new WindowBoundaryInnerSubscriber<Object, Object>(null);
 
-        final AtomicLong windows = new AtomicLong();
+        final AtomicInteger windows;
 
-        WindowBoundaryMainSubscriber(Subscriber<? super Flowable<T>> actual, Callable<? extends Publisher<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(Subscription s) {
-            if (SubscriptionHelper.validate(this.s, s)) {
-                this.s = s;
+        final AtomicThrowable errors;
 
-                Subscriber<? super Flowable<T>> a = actual;
-                a.onSubscribe(this);
+        final AtomicBoolean stopWindows;
 
-                if (cancelled) {
-                    return;
-                }
+        final Callable<? extends Publisher<B>> other;
 
-                Publisher<B> p;
+        static final Object NEXT_WINDOW = new Object();
 
-                try {
-                    p = ObjectHelper.requireNonNull(other.call(), "The first window publisher supplied is null");
-                } catch (Throwable e) {
-                    Exceptions.throwIfFatal(e);
-                    s.cancel();
-                    a.onError(e);
-                    return;
-                }
+        final AtomicLong requested;
 
-                UnicastProcessor<T> w = UnicastProcessor.<T>create(bufferSize);
+        Subscription upstream;
 
-                long r = requested();
-                if (r != 0L) {
-                    a.onNext(w);
-                    if (r != Long.MAX_VALUE) {
-                        produced(1);
-                    }
-                } else {
-                    s.cancel();
-                    a.onError(new MissingBackpressureException("Could not deliver first window due to lack of requests"));
-                    return;
-                }
+        volatile boolean done;
 
-                window = w;
+        UnicastProcessor<T> window;
 
-                WindowBoundaryInnerSubscriber<T, B> inner = new WindowBoundaryInnerSubscriber<T, B>(this);
+        long emitted;
 
-                if (boundary.compareAndSet(null, inner)) {
-                    windows.getAndIncrement();
-                    s.request(Long.MAX_VALUE);
-                    p.subscribe(inner);
-                }
-            }
+        WindowBoundaryMainSubscriber(Subscriber<? super Flowable<T>> downstream, int capacityHint, Callable<? extends Publisher<B>> other) {
+            this.downstream = downstream;
+            this.capacityHint = capacityHint;
+            this.boundarySubscriber = new AtomicReference<WindowBoundaryInnerSubscriber<T, B>>();
+            this.windows = new AtomicInteger(1);
+            this.queue = new MpscLinkedQueue<Object>();
+            this.errors = new AtomicThrowable();
+            this.stopWindows = new AtomicBoolean();
+            this.other = other;
+            this.requested = new AtomicLong();
         }
 
         @Override
-        public void onNext(T t) {
-            if (done) {
-                return;
+        public void onSubscribe(Subscription d) {
+            if (SubscriptionHelper.validate(upstream, d)) {
+                upstream = d;
+                downstream.onSubscribe(this);
+                queue.offer(NEXT_WINDOW);
+                drain();
+                d.request(Long.MAX_VALUE);
             }
-            if (fastEnter()) {
-                UnicastProcessor<T> w = window;
+        }
 
-                w.onNext(t);
+        @Override
+        public void onNext(T 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 cancel() {
+            if (stopWindows.compareAndSet(false, true)) {
+                disposeBoundary();
+                if (windows.decrementAndGet() == 0) {
+                    upstream.cancel();
+                }
             }
-
-            actual.onError(t);
         }
 
         @Override
-        public void onComplete() {
-            if (done) {
-                return;
-            }
-            done = true;
-            if (enter()) {
-                drainLoop();
+        public void request(long n) {
+            BackpressureHelper.add(requested, n);
+        }
+
+        @SuppressWarnings({ "rawtypes", "unchecked" })
+        void disposeBoundary() {
+            Disposable d = boundarySubscriber.getAndSet((WindowBoundaryInnerSubscriber)BOUNDARY_DISPOSED);
+            if (d != null && d != BOUNDARY_DISPOSED) {
+                d.dispose();
             }
+        }
 
+        @Override
+        public void run() {
             if (windows.decrementAndGet() == 0) {
-                DisposableHelper.dispose(boundary);
+                upstream.cancel();
             }
+        }
 
-            actual.onComplete();
-
+        void innerNext(WindowBoundaryInnerSubscriber<T, B> sender) {
+            boundarySubscriber.compareAndSet(sender, null);
+            queue.offer(NEXT_WINDOW);
+            drain();
         }
 
-        @Override
-        public void request(long n) {
-            requested(n);
+        void innerError(Throwable e) {
+            upstream.cancel();
+            if (errors.addThrowable(e)) {
+                done = true;
+                drain();
+            } else {
+                RxJavaPlugins.onError(e);
+            }
         }
 
-        @Override
-        public void cancel() {
-            cancelled = true;
+        void innerComplete() {
+            upstream.cancel();
+            done = true;
+            drain();
         }
 
-        void drainLoop() {
-            final SimplePlainQueue<Object> q = queue;
-            final Subscriber<? super Flowable<T>> a = actual;
+        @SuppressWarnings("unchecked")
+        void drain() {
+            if (getAndIncrement() != 0) {
+                return;
+            }
+
             int missed = 1;
-            UnicastProcessor<T> w = window;
+            Subscriber<? super Flowable<T>> downstream = this.downstream;
+            MpscLinkedQueue<Object> queue = this.queue;
+            AtomicThrowable errors = this.errors;
+            long emitted = this.emitted;
+
             for (;;) {
 
                 for (;;) {
+                    if (windows.get() == 0) {
+                        queue.clear();
+                        window = null;
+                        return;
+                    }
+
+                    UnicastProcessor<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;
                     }
@@ -221,73 +244,57 @@ void drainLoop() {
                         break;
                     }
 
-                    if (o == NEXT) {
+                    if (v != NEXT_WINDOW) {
+                        w.onNext((T)v);
+                        continue;
+                    }
+
+                    if (w != null) {
+                        window = null;
                         w.onComplete();
+                    }
 
-                        if (windows.decrementAndGet() == 0) {
-                            DisposableHelper.dispose(boundary);
-                            return;
-                        }
+                    if (!stopWindows.get()) {
+                        if (emitted != requested.get()) {
+                            w = UnicastProcessor.create(capacityHint, this);
+                            window = w;
+                            windows.getAndIncrement();
 
-                        if (cancelled) {
-                            continue;
-                        }
+                            Publisher<B> otherSource;
 
-                        Publisher<B> p;
+                            try {
+                                otherSource = ObjectHelper.requireNonNull(other.call(), "The other Callable returned a null Publisher");
+                            } catch (Throwable ex) {
+                                Exceptions.throwIfFatal(ex);
+                                errors.addThrowable(ex);
+                                done = true;
+                                continue;
+                            }
 
-                        try {
-                            p = ObjectHelper.requireNonNull(other.call(), "The publisher supplied is null");
-                        } catch (Throwable e) {
-                            Exceptions.throwIfFatal(e);
-                            DisposableHelper.dispose(boundary);
-                            a.onError(e);
-                            return;
-                        }
+                            WindowBoundaryInnerSubscriber<T, B> bo = new WindowBoundaryInnerSubscriber<T, B>(this);
 
-                        w = UnicastProcessor.<T>create(bufferSize);
+                            if (boundarySubscriber.compareAndSet(null, bo)) {
+                                otherSource.subscribe(bo);
 
-                        long r = requested();
-                        if (r != 0L) {
-                            windows.getAndIncrement();
-
-                            a.onNext(w);
-                            if (r != Long.MAX_VALUE) {
-                                produced(1);
+                                emitted++;
+                                downstream.onNext(w);
                             }
                         } else {
-                            // don't emit new windows
-                            cancelled = true;
-                            a.onError(new MissingBackpressureException("Could not deliver new window due to lack of requests"));
-                            continue;
-                        }
-
-                        window = w;
-
-                        WindowBoundaryInnerSubscriber<T, B> b = new WindowBoundaryInnerSubscriber<T, B>(this);
-
-                        if (boundary.compareAndSet(boundary.get(), b)) {
-                            p.subscribe(b);
+                            upstream.cancel();
+                            disposeBoundary();
+                            errors.addThrowable(new MissingBackpressureException("Could not deliver a window due to lack of requests"));
+                            done = true;
                         }
-
-                        continue;
                     }
-
-                    w.onNext(NotificationLite.<T>getValue(o));
                 }
 
-                missed = leave(-missed);
+                this.emitted = emitted;
+                missed = addAndGet(-missed);
                 if (missed == 0) {
-                    return;
+                    break;
                 }
             }
         }
-
-        void next() {
-            queue.offer(NEXT);
-            if (enter()) {
-                drainLoop();
-            }
-        }
     }
 
     static final class WindowBoundaryInnerSubscriber<T, B> extends DisposableSubscriber<B> {
@@ -305,8 +312,8 @@ public void onNext(B t) {
                 return;
             }
             done = true;
-            cancel();
-            parent.next();
+            dispose();
+            parent.innerNext(this);
         }
 
         @Override
@@ -316,7 +323,7 @@ public void onError(Throwable t) {
                 return;
             }
             done = true;
-            parent.onError(t);
+            parent.innerError(t);
         }
 
         @Override
@@ -325,7 +332,7 @@ public void onComplete() {
                 return;
             }
             done = true;
-            parent.onComplete();
+            parent.innerComplete();
         }
     }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableWindowWithFlowableTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableWindowWithFlowableTest.java
index 7ceccf678f..d9912f0b01 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableWindowWithFlowableTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableWindowWithFlowableTest.java
@@ -19,7 +19,7 @@
 
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.*;
 
 import org.junit.Test;
 import org.reactivestreams.*;
@@ -28,6 +28,8 @@
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.functions.Functions;
+import io.reactivex.internal.subscriptions.BooleanSubscription;
+import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.*;
 import io.reactivex.subscribers.*;
 
@@ -198,7 +200,7 @@ public void onComplete() {
     }
 
     @Test
-    public void testWindowViaFlowableSourceThrows() {
+    public void testWindowViaFlowableThrows() {
         PublishProcessor<Integer> source = PublishProcessor.create();
         PublishProcessor<Integer> boundary = PublishProcessor.create();
 
@@ -343,8 +345,8 @@ public Flowable<Integer> call() {
 
         boundary.onComplete();
 
-        // FIXME source still active because the open window
-        assertTrue(source.hasSubscribers());
+
+        assertFalse(source.hasSubscribers());
         assertFalse(boundary.hasSubscribers());
 
         ts.assertComplete();
@@ -371,10 +373,14 @@ public Flowable<Integer> call() {
 
         ts.dispose();
 
-        // FIXME source has subscribers because the open window
+
         assertTrue(source.hasSubscribers());
-        // FIXME boundary has subscribers because the open window
-        assertTrue(boundary.hasSubscribers());
+
+        assertFalse(boundary.hasSubscribers());
+
+        ts.values().get(0).test().cancel();
+
+        assertFalse(source.hasSubscribers());
 
         ts.assertNotComplete();
         ts.assertNoErrors();
@@ -503,8 +509,18 @@ public Flowable<Integer> apply(Flowable<Integer> v) throws Exception {
 
         TestHelper.checkBadSourceFlowable(new Function<Flowable<Integer>, Object>() {
             @Override
-            public Object apply(Flowable<Integer> o) throws Exception {
-                return Flowable.just(1).window(Functions.justCallable(o)).flatMap(new Function<Flowable<Integer>, Flowable<Integer>>() {
+            public Object apply(final Flowable<Integer> o) throws Exception {
+                return Flowable.just(1).window(new Callable<Publisher<Integer>>() {
+                    int count;
+                    @Override
+                    public Publisher<Integer> call() throws Exception {
+                        if (++count > 1) {
+                            return Flowable.never();
+                        }
+                        return o;
+                    }
+                })
+                        .flatMap(new Function<Flowable<Integer>, Flowable<Integer>>() {
                     @Override
                     public Flowable<Integer> apply(Flowable<Integer> v) throws Exception {
                         return v;
@@ -716,4 +732,625 @@ public Publisher<Flowable<Object>> apply(Flowable<Object> f)
             }
         });
     }
+
+    @Test
+    public void upstreamDisposedWhenOutputsDisposed() {
+        PublishProcessor<Integer> source = PublishProcessor.create();
+        PublishProcessor<Integer> boundary = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = source.window(boundary)
+        .take(1)
+        .flatMap(new Function<Flowable<Integer>, Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> apply(
+                    Flowable<Integer> w) throws Exception {
+                return w.take(1);
+            }
+        })
+        .test();
+
+        source.onNext(1);
+
+        assertFalse("source not disposed", source.hasSubscribers());
+        assertFalse("boundary not disposed", boundary.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
+
+    @Test
+    public void mainAndBoundaryBothError() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+            TestSubscriber<Flowable<Object>> to = Flowable.error(new TestException("main"))
+            .window(new Flowable<Object>() {
+                @Override
+                protected void subscribeActual(Subscriber<? super Object> observer) {
+                    observer.onSubscribe(new BooleanSubscription());
+                    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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+                final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+                TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+                    @Override
+                    protected void subscribeActual(Subscriber<? super Object> observer) {
+                        observer.onSubscribe(new BooleanSubscription());
+                        refMain.set(observer);
+                    }
+                }
+                .window(new Flowable<Object>() {
+                    @Override
+                    protected void subscribeActual(Subscriber<? super Object> observer) {
+                        observer.onSubscribe(new BooleanSubscription());
+                        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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+            final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+            TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+                @Override
+                protected void subscribeActual(Subscriber<? super Object> observer) {
+                    observer.onSubscribe(new BooleanSubscription());
+                    refMain.set(observer);
+                }
+            }
+            .window(new Flowable<Object>() {
+                @Override
+                protected void subscribeActual(Subscriber<? super Object> observer) {
+                    observer.onSubscribe(new BooleanSubscription());
+                    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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+        final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+        TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+            @Override
+            protected void subscribeActual(Subscriber<? super Object> observer) {
+                observer.onSubscribe(new BooleanSubscription());
+                refMain.set(observer);
+            }
+        }
+        .window(new Flowable<Object>() {
+            @Override
+            protected void subscribeActual(Subscriber<? super Object> observer) {
+                observer.onSubscribe(new BooleanSubscription());
+                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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+            final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+            final TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+                 @Override
+                 protected void subscribeActual(Subscriber<? super Object> observer) {
+                     observer.onSubscribe(new BooleanSubscription());
+                     refMain.set(observer);
+                 }
+             }
+             .window(new Flowable<Object>() {
+                 @Override
+                 protected void subscribeActual(Subscriber<? super Object> observer) {
+                     final AtomicInteger counter = new AtomicInteger();
+                     observer.onSubscribe(new Subscription() {
+
+                         @Override
+                         public void cancel() {
+                             // about a microsecond
+                             for (int i = 0; i < 100; i++) {
+                                 counter.incrementAndGet();
+                             }
+                         }
+
+                         @Override
+                        public void request(long n) {
+                        }
+                     });
+                     ref.set(observer);
+                 }
+             })
+             .test();
+
+             Runnable r1 = new Runnable() {
+                 @Override
+                 public void run() {
+                     to.cancel();
+                 }
+             };
+             Runnable r2 = new Runnable() {
+                 @Override
+                 public void run() {
+                     Subscriber<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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+           final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+           final TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+               @Override
+               protected void subscribeActual(Subscriber<? super Object> observer) {
+                   observer.onSubscribe(new BooleanSubscription());
+                   refMain.set(observer);
+               }
+           }
+           .window(new Flowable<Object>() {
+               @Override
+               protected void subscribeActual(Subscriber<? super Object> observer) {
+                   final AtomicInteger counter = new AtomicInteger();
+                   observer.onSubscribe(new Subscription() {
+
+                       @Override
+                       public void cancel() {
+                           // about a microsecond
+                           for (int i = 0; i < 100; i++) {
+                               counter.incrementAndGet();
+                           }
+                       }
+
+                       @Override
+                      public void request(long n) {
+                      }
+                   });
+                   ref.set(observer);
+               }
+           })
+           .test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    to.cancel();
+                }
+            };
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    Subscriber<Object> o = ref.get();
+                    o.onNext(1);
+                    o.onError(ex);
+                }
+            };
+
+            TestHelper.race(r1, r2);
+        }
+    }
+
+    @Test
+    public void boundarySupplierDoubleOnSubscribe() {
+        TestHelper.checkDoubleOnSubscribeFlowable(new Function<Flowable<Object>, Flowable<Flowable<Object>>>() {
+            @Override
+            public Flowable<Flowable<Object>> apply(Flowable<Object> f)
+                    throws Exception {
+                return f.window(Functions.justCallable(Flowable.never())).takeLast(1);
+            }
+        });
+    }
+
+    @Test
+    public void selectorUpstreamDisposedWhenOutputsDisposed() {
+        PublishProcessor<Integer> source = PublishProcessor.create();
+        PublishProcessor<Integer> boundary = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = source.window(Functions.justCallable(boundary))
+        .take(1)
+        .flatMap(new Function<Flowable<Integer>, Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> apply(
+                    Flowable<Integer> w) throws Exception {
+                return w.take(1);
+            }
+        })
+        .test();
+
+        source.onNext(1);
+
+        assertFalse("source not disposed", source.hasSubscribers());
+        assertFalse("boundary not disposed", boundary.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void supplierMainAndBoundaryBothError() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+            TestSubscriber<Flowable<Object>> to = Flowable.error(new TestException("main"))
+            .window(Functions.justCallable(new Flowable<Object>() {
+                @Override
+                protected void subscribeActual(Subscriber<? super Object> observer) {
+                    observer.onSubscribe(new BooleanSubscription());
+                    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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+                final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+                TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+                    @Override
+                    protected void subscribeActual(Subscriber<? super Object> observer) {
+                        observer.onSubscribe(new BooleanSubscription());
+                        refMain.set(observer);
+                    }
+                }
+                .window(Functions.justCallable(new Flowable<Object>() {
+                    @Override
+                    protected void subscribeActual(Subscriber<? super Object> observer) {
+                        observer.onSubscribe(new BooleanSubscription());
+                        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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+            final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+            TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+                @Override
+                protected void subscribeActual(Subscriber<? super Object> observer) {
+                    observer.onSubscribe(new BooleanSubscription());
+                    refMain.set(observer);
+                }
+            }
+            .window(Functions.justCallable(new Flowable<Object>() {
+                @Override
+                protected void subscribeActual(Subscriber<? super Object> observer) {
+                    observer.onSubscribe(new BooleanSubscription());
+                    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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+        final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+        TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+            @Override
+            protected void subscribeActual(Subscriber<? super Object> observer) {
+                observer.onSubscribe(new BooleanSubscription());
+                refMain.set(observer);
+            }
+        }
+        .window(Functions.justCallable(new Flowable<Object>() {
+            @Override
+            protected void subscribeActual(Subscriber<? super Object> observer) {
+                observer.onSubscribe(new BooleanSubscription());
+                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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+            final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+            final TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+                 @Override
+                 protected void subscribeActual(Subscriber<? super Object> observer) {
+                     observer.onSubscribe(new BooleanSubscription());
+                     refMain.set(observer);
+                 }
+             }
+             .window(Functions.justCallable(new Flowable<Object>() {
+                 @Override
+                 protected void subscribeActual(Subscriber<? super Object> observer) {
+                     final AtomicInteger counter = new AtomicInteger();
+                     observer.onSubscribe(new Subscription() {
+
+                         @Override
+                         public void cancel() {
+                             // about a microsecond
+                             for (int i = 0; i < 100; i++) {
+                                 counter.incrementAndGet();
+                             }
+                         }
+
+                         @Override
+                         public void request(long n) {
+                         }
+                      });
+                     ref.set(observer);
+                 }
+             }))
+             .test();
+
+             Runnable r1 = new Runnable() {
+                 @Override
+                 public void run() {
+                     to.cancel();
+                 }
+             };
+             Runnable r2 = new Runnable() {
+                 @Override
+                 public void run() {
+                     Subscriber<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<Subscriber<? super Object>> refMain = new AtomicReference<Subscriber<? super Object>>();
+                final AtomicReference<Subscriber<? super Object>> ref = new AtomicReference<Subscriber<? super Object>>();
+
+                final TestSubscriber<Flowable<Object>> to = new Flowable<Object>() {
+                    @Override
+                    protected void subscribeActual(Subscriber<? super Object> observer) {
+                        observer.onSubscribe(new BooleanSubscription());
+                        refMain.set(observer);
+                    }
+                }
+                .window(new Callable<Flowable<Object>>() {
+                    int count;
+                    @Override
+                    public Flowable<Object> call() throws Exception {
+                        if (++count > 1) {
+                            return Flowable.never();
+                        }
+                        return (new Flowable<Object>() {
+                            @Override
+                            protected void subscribeActual(Subscriber<? super Object> observer) {
+                                final AtomicInteger counter = new AtomicInteger();
+                                observer.onSubscribe(new Subscription() {
+
+                                    @Override
+                                    public void cancel() {
+                                        // about a microsecond
+                                        for (int i = 0; i < 100; i++) {
+                                            counter.incrementAndGet();
+                                        }
+                                    }
+
+                                    @Override
+                                    public void request(long n) {
+                                    }
+                                });
+                                ref.set(observer);
+                            }
+                        });
+                    }
+                })
+                .test();
+
+                Runnable r1 = new Runnable() {
+                    @Override
+                    public void run() {
+                        to.cancel();
+                    }
+                };
+                Runnable r2 = new Runnable() {
+                    @Override
+                    public void run() {
+                        Subscriber<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();
+            }
+        }
+    }
+
 }