Merge pull request #2845 from akarnokd/Issue2844Fix

Fix for repeat: wrong target of request
ReactiveX · Mar 25, 2015 · 34cd1a6 · 34cd1a6
2 parents cfeb5eb + bbd4c37
commit 34cd1a6
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 1 deletion.
diff --git a/src/main/java/rx/internal/operators/OnSubscribeRedo.java b/src/main/java/rx/internal/operators/OnSubscribeRedo.java
@@ -213,12 +213,14 @@ public void call() {
                 Subscriber<T> terminalDelegatingSubscriber = new Subscriber<T>() {
                     @Override
                     public void onCompleted() {
+                        currentProducer.set(null);
                         unsubscribe();
                         terminals.onNext(Notification.createOnCompleted());
                     }
 
                     @Override
                     public void onError(Throwable e) {
+                        currentProducer.set(null);
                         unsubscribe();
                         terminals.onNext(Notification.createOnError(e));
                     }

diff --git a/src/test/java/rx/internal/operators/OperatorRepeatTest.java b/src/test/java/rx/internal/operators/OperatorRepeatTest.java
@@ -20,7 +20,8 @@
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
-import java.util.Arrays;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
@@ -174,4 +175,28 @@ public void testRepeatAndDistinctUnbounded() {
         ts.assertTerminalEvent();
         ts.assertReceivedOnNext(Arrays.asList(1, 2, 3));
     }
+    /** Issue #2844: wrong target of request. */
+    @Test(timeout = 3000)
+    public void testRepeatRetarget() {
+        final List<Integer> concatBase = new ArrayList<Integer>();
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.just(1, 2)
+        .repeat(5)
+        .concatMap(new Func1<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(Integer x) {
+                System.out.println("testRepeatRetarget -> " + x);
+                concatBase.add(x);
+                return Observable.<Integer>empty()
+                        .delay(200, TimeUnit.MILLISECONDS);
+            }
+        })
+        .subscribe(ts);
+
+        ts.awaitTerminalEvent();
+        ts.assertNoErrors();
+        ts.assertReceivedOnNext(Collections.<Integer>emptyList());
+
+        assertEquals(Arrays.asList(1, 2, 1, 2, 1, 2, 1, 2, 1, 2), concatBase);
+    }
 }