Merge pull request #1793 from benjchristensen/1791_first_with_retryWhen

Take/Redo Unsubscribe
ReactiveX · Oct 23, 2014 · 147c12d · 147c12d
2 parents bcf2a39 + d217209
commit 147c12d
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 2 deletions.
diff --git a/src/main/java/rx/internal/operators/OnSubscribeRedo.java b/src/main/java/rx/internal/operators/OnSubscribeRedo.java
@@ -206,6 +206,10 @@ public void call(final Subscriber<? super T> child) {
         final Action0 subscribeToSource = new Action0() {
             @Override
             public void call() {
+                if (child.isUnsubscribed()) {
+                    return;
+                }
+
                 Subscriber<T> terminalDelegatingSubscriber = new Subscriber<T>() {
                     @Override
                     public void onCompleted() {

diff --git a/src/main/java/rx/internal/operators/OperatorTake.java b/src/main/java/rx/internal/operators/OperatorTake.java
@@ -61,12 +61,16 @@ public void onError(Throwable e) {
             @Override
             public void onNext(T i) {
                 if (!isUnsubscribed()) {
-                    child.onNext(i);
                     if (++count >= limit) {
                         completed = true;
-                        child.onCompleted();
+                        // unsubscribe before emitting onNext so shutdown happens before possible effects
+                        // of onNext such as product.request(n) calls be sent upstream. 
                         unsubscribe();
                     }
+                    child.onNext(i);
+                    if (completed) {
+                        child.onCompleted();
+                    }
                 }
             }
 

diff --git a/src/test/java/rx/internal/operators/OperatorRetryTest.java b/src/test/java/rx/internal/operators/OperatorRetryTest.java
@@ -244,6 +244,37 @@ public Observable<?> call(Observable<? extends Throwable> t1) {
         inOrder.verifyNoMoreInteractions();
     }
 
+    @Test
+    public void testSingleSubscriptionOnFirst() throws Exception {
+        final AtomicInteger inc = new AtomicInteger(0);
+        Observable.OnSubscribe<Integer> onSubscribe = new OnSubscribe<Integer>() {
+            @Override
+            public void call(Subscriber<? super Integer> subscriber) {
+                final int emit = inc.incrementAndGet();
+                subscriber.onNext(emit);
+                subscriber.onCompleted();
+            }
+        };
+
+        int first = Observable.create(onSubscribe)
+                .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
+                    @Override
+                    public Observable<?> call(Observable<? extends Throwable> attempt) {
+                        return attempt.zipWith(Observable.just(1), new Func2<Throwable, Integer, Void>() {
+                            @Override
+                            public Void call(Throwable o, Integer integer) {
+                                return null;
+                            }
+                        });
+                    }
+                })
+                .toBlocking()
+                .first();
+
+        assertEquals("Observer did not receive the expected output", 1, first);
+        assertEquals("Subscribe was not called once", 1, inc.get());
+    }
+
     @Test
     public void testOriginFails() {
         @SuppressWarnings("unchecked")