prevent calling subscriber to FuncWithErrors with events after terminal event occurred (race condition)

Author: davidmoten

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

@@ -410,7 +410,7 @@ public void request(long n) {
                         }
                         return;
                     }
-                    if (n > 0 && req.getAndAdd(n) == 0 && (path.get() == 2 || path.compareAndSet(0, 2))) {
+                    if (n > 0 && req.getAndAdd(n) == 0 && (path.get() == 2 || path.compareAndSet(0, 2)) && !o.isUnsubscribed()) {
                         int i = count.getAndIncrement();
                         if (i < numFailures) {
                             o.onNext("beginningEveryTime");
@@ -686,25 +686,28 @@ public void testTimeoutWithRetry() {
 
     @Test//(timeout = 15000)
     public void testRetryWithBackpressure() throws InterruptedException {
-        final int NUM_RETRIES = RxRingBuffer.SIZE * 2;
-        for (int i = 0; i < 400; i++) {
-            @SuppressWarnings("unchecked")
-            Observer<String> observer = mock(Observer.class);
-            Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
-            TestSubscriber<String> ts = new TestSubscriber<String>(observer);
-            origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
-            ts.awaitTerminalEvent(5, TimeUnit.SECONDS);
-            
-            InOrder inOrder = inOrder(observer);
-            // should have no errors
-            verify(observer, never()).onError(any(Throwable.class));
-            // should show NUM_RETRIES attempts
-            inOrder.verify(observer, times(NUM_RETRIES + 1)).onNext("beginningEveryTime");
-            // should have a single success
-            inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
-            // should have a single successful onCompleted
-            inOrder.verify(observer, times(1)).onCompleted();
-            inOrder.verifyNoMoreInteractions();
+        final int NUM_LOOPS = 1;
+        for (int j=0;j<NUM_LOOPS;j++) {
+            final int NUM_RETRIES = RxRingBuffer.SIZE * 2;
+            for (int i = 0; i < 400; i++) {
+                @SuppressWarnings("unchecked")
+                Observer<String> observer = mock(Observer.class);
+                Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
+                TestSubscriber<String> ts = new TestSubscriber<String>(observer);
+                origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
+                ts.awaitTerminalEvent(5, TimeUnit.SECONDS);
+                
+                InOrder inOrder = inOrder(observer);
+                // should have no errors
+                verify(observer, never()).onError(any(Throwable.class));
+                // should show NUM_RETRIES attempts
+                inOrder.verify(observer, times(NUM_RETRIES + 1)).onNext("beginningEveryTime");
+                // should have a single success
+                inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
+                // should have a single successful onCompleted
+                inOrder.verify(observer, times(1)).onCompleted();
+                inOrder.verifyNoMoreInteractions();
+            }
         }
     }