Check child unsubscription status more eagerly.

Author: akarnokd

rxjava-core/src/main/java/rx/operators/BufferUntilSubscriber.java

@@ -84,22 +84,23 @@ public void enterPassthroughMode() {
                 if (!passthroughMode) {
                     while (!queue.isEmpty()) {
                         Object o = queue.poll();
-                        
-                        if (o == NULL_SENTINEL) {
-                            actual.onNext(null);
-                        } else
-                        if (o == COMPLETE_SENTINEL) {
-                            actual.onCompleted();
-                        } else
-                        if (o instanceof ErrorSentinel) {
-                            actual.onError(((ErrorSentinel)o).t);
-                        } else
-                        if (o != null) {
-                            @SuppressWarnings("unchecked")
-                            T v = (T)o;
-                            actual.onNext(v);
-                        } else {
-                            throw new NullPointerException();
+                        if (!actual.isUnsubscribed()) {
+                            if (o == NULL_SENTINEL) {
+                                actual.onNext(null);
+                            } else
+                            if (o == COMPLETE_SENTINEL) {
+                                actual.onCompleted();
+                            } else
+                            if (o instanceof ErrorSentinel) {
+                                actual.onError(((ErrorSentinel)o).t);
+                            } else
+                            if (o != null) {
+                                @SuppressWarnings("unchecked")
+                                T v = (T)o;
+                                actual.onNext(v);
+                            } else {
+                                throw new NullPointerException();
+                            }
                         }
                     }
                     passthroughMode = true;
@@ -121,6 +122,9 @@ public void onNext(T t) {
                         while (!passthroughMode) {
                             gate.wait();
                         }
+                        if (actual.isUnsubscribed()) {
+                            return;
+                        }
                     } catch (InterruptedException ex) {
                         Thread.currentThread().interrupt();
                         actual.onError(ex);
@@ -145,6 +149,9 @@ public void onError(Throwable e) {
                         while (!passthroughMode) {
                             gate.wait();
                         }
+                        if (actual.isUnsubscribed()) {
+                            return;
+                        }
                     } catch (InterruptedException ex) {
                         Thread.currentThread().interrupt();
                         actual.onError(ex);
@@ -169,6 +176,9 @@ public void onCompleted() {
                         while (!passthroughMode) {
                             gate.wait();
                         }
+                        if (actual.isUnsubscribed()) {
+                            return;
+                        }
                     } catch (InterruptedException ex) {
                         Thread.currentThread().interrupt();
                         actual.onError(ex);