fix OperatorObserveOn race condition where onComplete could be emitted despite onError being called

Author: davidmoten

src/main/java/rx/internal/operators/OperatorObserveOn.java

@@ -75,8 +75,15 @@ private static final class ObserveOnSubscriber<T> extends Subscriber<T> {
         final NotificationLite<T> on = NotificationLite.instance();
 
         final Queue<Object> queue;
-        volatile boolean completed = false;
-        volatile boolean failure = false;
+        
+        // These are the states for the status field. The transitions are 
+        // ACTIVE -> COMPLETED and ACTIVE -> ERRORED
+        private static final byte ACTIVE = 0;
+        private static final byte COMPLETED = 1;
+        private static final byte ERRORED = 2;
+        
+        // the current status of the incoming stream, possible values listed above
+        volatile byte status = ACTIVE;
 
         volatile long requested = 0;
         @SuppressWarnings("rawtypes")
@@ -127,7 +134,7 @@ public void onStart() {
 
         @Override
         public void onNext(final T t) {
-            if (isUnsubscribed() || completed) {
+            if (isUnsubscribed() || status !=ACTIVE) {
                 return;
             }
             if (!queue.offer(on.next(t))) {
@@ -139,30 +146,23 @@ public void onNext(final T t) {
 
         @Override
         public void onCompleted() {
-            if (isUnsubscribed() || completed) {
-                return;
-            }
-            if (error != null) {
+            if (isUnsubscribed() || status != ACTIVE) {
                 return;
             }
-            completed = true;
+            status = COMPLETED;
             schedule();
         }
 
         @Override
         public void onError(final Throwable e) {
-            if (isUnsubscribed() || completed) {
-                return;
-            }
-            if (error != null) {
+            if (isUnsubscribed() || status != ACTIVE) {
                 return;
             }
             error = e;
             // unsubscribe eagerly since time will pass before the scheduled onError results in an unsubscribe event
             unsubscribe();
             // mark failure so the polling thread will skip onNext still in the queue
-            completed = true;
-            failure = true;
+            status = ERRORED;
             schedule();
         }
 
@@ -191,41 +191,37 @@ void pollQueue() {
                  */
                 counter = 1;
 
-//                middle:
                 while (!scheduledUnsubscribe.isUnsubscribed()) {
-                    if (failure) {
+                    if (status == ERRORED) {
                         child.onError(error);
                         return;
-                    } else {
-                        if (requested == 0 && completed && queue.isEmpty()) {
-                            child.onCompleted();
-                            return;
-                        }
-                        if (REQUESTED.getAndDecrement(this) != 0) {
-                            Object o = queue.poll();
-                            if (o == null) {
-                                if (completed) {
-                                    if (failure) {
-                                        child.onError(error);
-                                    } else {
-                                        child.onCompleted();
-                                    }
-                                    return;
-                                }
-                                // nothing in queue
+                    } else if (requested == 0 && status == COMPLETED && queue.isEmpty()) {
+                        child.onCompleted();
+                        return;
+                    } else if (REQUESTED.getAndDecrement(this) != 0) {
+                        Object o = queue.poll();
+                        if (o == null) {
+                            // nothing in queue
+                            if (status == ERRORED) {
+                                child.onError(error);
+                                return;
+                            } else if (status == COMPLETED) {
+                                child.onCompleted();
+                                return;
+                            } else {
                                 REQUESTED.incrementAndGet(this);
                                 break;
-                            } else {
-                                if (!on.accept(child, o)) {
-                                    // non-terminal event so let's increment count
-                                    emitted++;
-                                }
                             }
                         } else {
-                            // we hit the end ... so increment back to 0 again
-                            REQUESTED.incrementAndGet(this);
-                            break;
+                            if (!on.accept(child, o)) {
+                                // non-terminal event so let's increment count
+                                emitted++;
+                            }
                         }
+                    } else {
+                        // we hit the end ... so increment back to 0 again
+                        REQUESTED.incrementAndGet(this);
+                        break;
                     }
                 }
             } while (COUNTER_UPDATER.decrementAndGet(this) > 0);