OperationRetry -> OperatorRetry

Updated to use "lift" and Subscriber.
Fixes #943 Observable.retry() does not unsubscribe from source

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -5985,7 +5985,7 @@ public final ConnectableObservable<T> replay(Scheduler scheduler) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Error-Handling-Operators#wiki-retry">RxJava Wiki: retry()</a>
      */
     public final Observable<T> retry() {
-        return create(OperationRetry.retry(this));
+        return nest().lift(new OperatorRetry<T>());
     }
 
     /**
@@ -6009,7 +6009,7 @@ public final Observable<T> retry() {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Error-Handling-Operators#wiki-retry">RxJava Wiki: retry()</a>
      */
     public final Observable<T> retry(int retryCount) {
-        return create(OperationRetry.retry(this, retryCount));
+        return nest().lift(new OperatorRetry<T>(retryCount));
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationRetry.java

@@ -0,0 +1,110 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import rx.Observable;
+import rx.Observable.Operator;
+import rx.Scheduler.Inner;
+import rx.Subscriber;
+import rx.functions.Action1;
+import rx.schedulers.Schedulers;
+
+public class OperatorRetry<T> implements Operator<T, Observable<T>> {
+
+    private static final int INFINITE_RETRY = -1;
+
+    private final int retryCount;
+    private final AtomicInteger attempts = new AtomicInteger(0);
+
+    public OperatorRetry(int retryCount) {
+        this.retryCount = retryCount;
+    }
+
+    public OperatorRetry() {
+        this(INFINITE_RETRY);
+    }
+
+    @Override
+    public Subscriber<? super Observable<T>> call(final Subscriber<? super T> s) {
+        return new Subscriber<Observable<T>>(s) {
+
+            @Override
+            public void onCompleted() {
+                // ignore as we expect a single nested Observable<T>
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                s.onError(e);
+            }
+
+            @Override
+            public void onNext(final Observable<T> o) {
+                Schedulers.trampoline().schedule(new Action1<Inner>() {
+
+                    @Override
+                    public void call(final Inner inner) {
+                        final Action1<Inner> _self = this;
+                        attempts.incrementAndGet();
+                        o.subscribe(new Subscriber<T>(s) {
+
+                            @Override
+                            public void onCompleted() {
+                                s.onCompleted();
+                            }
+
+                            @Override
+                            public void onError(Throwable e) {
+                                if ((retryCount == INFINITE_RETRY || attempts.get() <= retryCount) && !inner.isUnsubscribed()) {
+                                    // retry again
+                                    inner.schedule(_self);
+                                } else {
+                                    // give up and pass the failure
+                                    s.onError(e);
+                                }
+                            }
+
+                            @Override
+                            public void onNext(T v) {
+                                s.onNext(v);
+                            }
+
+                        });
+                    }
+                });
+            }
+
+        };
+    }
+}

rxjava-core/src/main/java/rx/operators/OperatorRetry.java

@@ -15,9 +15,10 @@
  */
 package rx.operators;
 
+import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
-import static rx.operators.OperationRetry.*;
+import static rx.operators.OperatorRetry.*;
 
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -27,9 +28,11 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.functions.Action1;
+import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
 
-public class OperationRetryTest {
+public class OperatorRetryTest {
 
     @Test
     public void testOriginFails() {
@@ -52,7 +55,7 @@ public void testRetryFail() {
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
         Observable<String> origin = Observable.create(new FuncWithErrors(NUM_FAILURES));
-        Observable.create(retry(origin, NUM_RETRIES)).subscribe(observer);
+        origin.nest().lift(new OperatorRetry<String>(NUM_RETRIES)).subscribe(observer);
 
         InOrder inOrder = inOrder(observer);
         // should show 2 attempts (first time fail, second time (1st retry) fail)
@@ -72,7 +75,7 @@ public void testRetrySuccess() {
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
         Observable<String> origin = Observable.create(new FuncWithErrors(NUM_FAILURES));
-        Observable.create(retry(origin, NUM_RETRIES)).subscribe(observer);
+        origin.nest().lift(new OperatorRetry<String>(NUM_RETRIES)).subscribe(observer);
 
         InOrder inOrder = inOrder(observer);
         // should show 3 attempts
@@ -92,7 +95,7 @@ public void testInfiniteRetry() {
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
         Observable<String> origin = Observable.create(new FuncWithErrors(NUM_FAILURES));
-        Observable.create(retry(origin)).subscribe(observer);
+        origin.nest().lift(new OperatorRetry<String>()).subscribe(observer);
 
         InOrder inOrder = inOrder(observer);
         // should show 3 attempts
@@ -127,4 +130,19 @@ public Subscription onSubscribe(Observer<? super String> o) {
             return Subscriptions.empty();
         }
     }
+    
+    @Test
+    public void testUnsubscribeFromRetry() {
+        PublishSubject<Integer> subject = PublishSubject.create();
+        final AtomicInteger count = new AtomicInteger(0);
+        Subscription sub = subject.retry().subscribe(new Action1<Integer>() {
+            @Override
+            public void call(Integer n) {
+                count.incrementAndGet();
+            }});
+        subject.onNext(1);
+        sub.unsubscribe();
+        subject.onNext(2);
+        assertEquals(1,count.get());
+    }
 }