prevent request overflow in OperatorObserveOn and add unit test that fails on original codebase but passes with change

Author: davidmoten

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

@@ -16,15 +16,22 @@
 package rx.internal.operators;
 
 import java.util.Queue;
-import java.util.concurrent.atomic.*;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 import rx.Observable.Operator;
-import rx.*;
+import rx.Producer;
+import rx.Scheduler;
+import rx.Subscriber;
+import rx.Subscription;
 import rx.exceptions.MissingBackpressureException;
 import rx.functions.Action0;
-import rx.internal.util.*;
-import rx.internal.util.unsafe.*;
-import rx.schedulers.*;
+import rx.internal.util.RxRingBuffer;
+import rx.internal.util.SynchronizedQueue;
+import rx.internal.util.unsafe.SpscArrayQueue;
+import rx.internal.util.unsafe.UnsafeAccess;
+import rx.schedulers.ImmediateScheduler;
+import rx.schedulers.TrampolineScheduler;
 
 /**
  * Delivers events on the specified {@code Scheduler} asynchronously via an unbounded buffer.
@@ -96,7 +103,7 @@ public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child) {
 
                 @Override
                 public void request(long n) {
-                    REQUESTED.getAndAdd(ObserveOnSubscriber.this, n);
+                    BackpressureUtils.getAndAddRequest(REQUESTED, ObserveOnSubscriber.this, n);
                     schedule();
                 }
 

src/test/java/rx/internal/operators/OperatorObserveOnTest.java

@@ -724,4 +724,45 @@ public Long call(Long t1, Integer t2) {
         assertEquals(MissingBackpressureException.class, ts.getOnErrorEvents().get(0).getClass());
     }
 
+    @Test
+    public void testRequestOverflow() throws InterruptedException {
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        final AtomicInteger count = new AtomicInteger();
+        Observable.range(1, 100).observeOn(Schedulers.computation())
+                .subscribe(new Subscriber<Integer>() {
+
+                    boolean first = true;
+                    
+                    @Override
+                    public void onStart() {
+                        request(2);
+                    }
+
+                    @Override
+                    public void onCompleted() {
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+
+                    }
+
+                    @Override
+                    public void onNext(Integer t) {
+                        count.incrementAndGet();
+                        if (first) {
+                            request(Long.MAX_VALUE - 1);
+                            request(Long.MAX_VALUE - 1);
+                            request(10);
+                            first = false;
+                        }
+                    }
+                });
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+        assertEquals(100, count.get());
+
+    }
+    
 }