emit MissingBackpressureException

Author: davidmoten

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

@@ -236,21 +236,29 @@ public void onError(Throwable e) {
                     @Override
                     public void onNext(T v) {
                         if (!done) {
-                            child.onNext(v);
-                            while (true) {
-                                long cc = consumerCapacity.get();
-                                if (cc == 0) {
-                                    child.onError(new MissingBackpressureException(
-                                            "an item has arrived to this operator that was not requested, "
-                                                    + "please use backpressure aware operators upstream (or append .onBackpressureBuffer() or similar)"));
-                                    break;
-                                } else if (cc != Long.MAX_VALUE) {
-                                    if (consumerCapacity.compareAndSet(cc, cc - 1)) {
-                                        break;
-                                    }
-                                } else {
+                            if (consumerCapacity.get() == 0) {
+                                child.onError(new MissingBackpressureException(
+                                        "an item has arrived to this operator that was not requested, "
+                                                + "please use backpressure aware operators upstream (or insert .onBackpressureBuffer() or similar)"));
+                                return;
+                            } else {
+                                child.onNext(v);
+                                decrementConsumerCapacity();
+                            }
+                        }
+                    }
+
+                    private void decrementConsumerCapacity() {
+                        // use a CAS loop because we don't want to decrement the value 
+                        // if it is Long.MAX_VALUE
+                        while (true) {
+                            long cc = consumerCapacity.get();
+                            if (cc != Long.MAX_VALUE) {
+                                if (consumerCapacity.compareAndSet(cc, cc - 1)) {
                                     break;
                                 }
+                            } else {
+                                break;
                             }
                         }
                     }