From d1282eec387fb140b3695ac016f828a7e181a95c Mon Sep 17 00:00:00 2001
From: George Campbell <abersnaze@gmail.com>
Date: Fri, 3 Oct 2014 14:48:04 -0700
Subject: [PATCH 1/2] Add a shim to make it so the public api for repeatWhen
 and retryWhen don't expose Notification<?>

---
 src/main/java/rx/Observable.java | 60 +++++++++++++++++++++++++++-----
 1 file changed, 52 insertions(+), 8 deletions(-)

diff --git a/src/main/java/rx/Observable.java b/src/main/java/rx/Observable.java
index 88335ef03e..20ef13bdbd 100644
--- a/src/main/java/rx/Observable.java
+++ b/src/main/java/rx/Observable.java
@@ -5801,8 +5801,19 @@ public final Observable<T> repeat(final long count, Scheduler scheduler) {
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Creating-Observables#repeatwhen">RxJava Wiki: repeatWhen()</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428.aspx">MSDN: Observable.Repeat</a>
      */
-    public final Observable<T> repeatWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler, Scheduler scheduler) {
-        return OnSubscribeRedo.repeat(this, notificationHandler, scheduler);
+    public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable<?>> notificationHandler, Scheduler scheduler) {
+        Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> dematerializedNotificationHandler = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
+            @Override
+            public Observable<?> call(Observable<? extends Notification<?>> notifications) {
+                return notificationHandler.call(notifications.map(new Func1<Notification<?>, Void>() {
+                    @Override
+                    public Void call(Notification<?> notification) {
+                        return null;
+                    }
+                }));
+            }
+        };
+        return OnSubscribeRedo.repeat(this, dematerializedNotificationHandler, scheduler);
     }
 
     /**
@@ -5825,8 +5836,19 @@ public final Observable<T> repeatWhen(Func1<? super Observable<? extends Notific
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Creating-Observables#repeatwhen">RxJava Wiki: repeatWhen()</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229428.aspx">MSDN: Observable.Repeat</a>
      */
-    public final Observable<T> repeatWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler) {
-        return OnSubscribeRedo.repeat(this, notificationHandler);
+    public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable<?>> notificationHandler) {
+        Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> dematerializedNotificationHandler = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
+            @Override
+            public Observable<?> call(Observable<? extends Notification<?>> notifications) {
+                return notificationHandler.call(notifications.map(new Func1<Notification<?>, Void>() {
+                    @Override
+                    public Void call(Notification<?> notification) {
+                        return null;
+                    }
+                }));
+            }
+        };
+        return OnSubscribeRedo.repeat(this, dematerializedNotificationHandler);
     }
 
     /**
@@ -6541,8 +6563,19 @@ public final Observable<T> retry(Func2<Integer, Throwable, Boolean> predicate) {
      * @return the source Observable modified with retry logic
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators#retrywhen">RxJava Wiki: retryWhen()</a>
      */
-    public final Observable<T> retryWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler) {
-        return OnSubscribeRedo.<T> retry(this, notificationHandler);
+    public final Observable<T> retryWhen(final Func1<? super Observable<? extends Throwable>, ? extends Observable<?>> notificationHandler) {
+        Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> dematerializedNotificationHandler = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
+            @Override
+            public Observable<?> call(Observable<? extends Notification<?>> notifications) {
+                return notificationHandler.call(notifications.map(new Func1<Notification<?>, Throwable>() {
+                    @Override
+                    public Throwable call(Notification<?> notification) {
+                        return notification.getThrowable();
+                    }
+                }));
+            }
+        };
+        return OnSubscribeRedo.<T> retry(this, dematerializedNotificationHandler);
     }
 
     /**
@@ -6566,8 +6599,19 @@ public final Observable<T> retryWhen(Func1<? super Observable<? extends Notifica
      * @return the source Observable modified with retry logic
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Error-Handling-Operators#retrywhen">RxJava Wiki: retryWhen()</a>
      */
-    public final Observable<T> retryWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler, Scheduler scheduler) {
-        return OnSubscribeRedo.<T> retry(this, notificationHandler, scheduler);
+    public final Observable<T> retryWhen(final Func1<? super Observable<? extends Throwable>, ? extends Observable<?>> notificationHandler, Scheduler scheduler) {
+        Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> dematerializedNotificationHandler = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
+            @Override
+            public Observable<?> call(Observable<? extends Notification<?>> notifications) {
+                return notificationHandler.call(notifications.map(new Func1<Notification<?>, Throwable>() {
+                    @Override
+                    public Throwable call(Notification<?> notification) {
+                        return notification.getThrowable();
+                    }
+                }));
+            }
+        };
+        return OnSubscribeRedo.<T> retry(this, dematerializedNotificationHandler, scheduler);
     }
 
     /**

From bb8be53f1058db3d94b3dcad6fcc75047e347e00 Mon Sep 17 00:00:00 2001
From: George Campbell <abersnaze@gmail.com>
Date: Fri, 3 Oct 2014 15:01:45 -0700
Subject: [PATCH 2/2] forgot to commit the changes to the tests.

---
 .../internal/operators/OperatorRetryTest.java | 57 +++++++++----------
 1 file changed, 28 insertions(+), 29 deletions(-)

diff --git a/src/test/java/rx/internal/operators/OperatorRetryTest.java b/src/test/java/rx/internal/operators/OperatorRetryTest.java
index dbfa504d45..716a0bfd8a 100644
--- a/src/test/java/rx/internal/operators/OperatorRetryTest.java
+++ b/src/test/java/rx/internal/operators/OperatorRetryTest.java
@@ -24,18 +24,12 @@
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
 import org.junit.Test;
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 
 import rx.Observable;
 import rx.Observable.OnSubscribe;
-import rx.Notification;
 import rx.Observer;
 import rx.Subscriber;
 import rx.Subscription;
@@ -49,6 +43,11 @@
 import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
 public class OperatorRetryTest {
 
     @Test
@@ -73,15 +72,15 @@ public void call(Subscriber<? super String> t1) {
             
         });
         TestSubscriber<String> ts = new TestSubscriber<String>(consumer);
-        producer.retryWhen(new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
+        producer.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
 
             @Override
-            public Observable<?> call(Observable<? extends Notification<?>> attempts) {
+            public Observable<?> call(Observable<? extends Throwable> attempts) {
                 // Worker w = Schedulers.computation().createWorker();
                 return attempts
-                    .map(new Func1<Notification<?>, Tuple>() {
+                    .map(new Func1<Throwable, Tuple>() {
                         @Override
-                        public Tuple call(Notification<?> n) {
+                        public Tuple call(Throwable n) {
                             return new Tuple(new Long(1), n);
                         }})
                     .scan(new Func2<Tuple, Tuple, Tuple>(){
@@ -94,7 +93,7 @@ public Tuple call(Tuple t, Tuple n) {
                         public Observable<Long> call(Tuple t) {
                             System.out.println("Retry # "+t.count);
                             return t.count > 20 ? 
-                                Observable.<Long>error(t.n.getThrowable()) :
+                                Observable.<Long>error(t.n) :
                                 Observable.timer(t.count *1L, TimeUnit.MILLISECONDS);
                     }});
             }
@@ -112,9 +111,9 @@ public Observable<Long> call(Tuple t) {
 
     public static class Tuple {
         Long count;
-        Notification<?> n;
+        Throwable n;
 
-        Tuple(Long c, Notification<?> n) {
+        Tuple(Long c, Throwable n) {
             count = c;
             this.n = n;
         }
@@ -147,15 +146,15 @@ public void testSchedulingNotificationHandler() {
         int NUM_RETRIES = 2;
         Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
         TestSubscriber<String> subscriber = new TestSubscriber<String>(observer);
-        origin.retryWhen(new Func1<Observable<? extends Notification<?>>, Observable<? extends Notification<?>>>() {
+        origin.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
             @Override
-            public Observable<? extends Notification<?>> call(Observable<? extends Notification<?>> t1) {
-                return t1.observeOn(Schedulers.computation()).map(new Func1<Notification<?>, Notification<?>>() {
+            public Observable<?> call(Observable<? extends Throwable> t1) {
+                return t1.observeOn(Schedulers.computation()).map(new Func1<Throwable, Void>() {
                     @Override
-                    public Notification<?> call(Notification<?> t1) {
-                        return Notification.createOnNext(null);
+                    public Void call(Throwable t1) {
+                        return null;
                     }
-                }).startWith(Notification.createOnNext(null));
+                }).startWith((Void) null);
             }
         }).subscribe(subscriber);
 
@@ -178,16 +177,16 @@ public void testOnNextFromNotificationHandler() {
         Observer<String> observer = mock(Observer.class);
         int NUM_RETRIES = 2;
         Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
-        origin.retryWhen(new Func1<Observable<? extends Notification<?>>, Observable<? extends Notification<?>>>() {
+        origin.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
             @Override
-            public Observable<? extends Notification<?>> call(Observable<? extends Notification<?>> t1) {
-                return t1.map(new Func1<Notification<?>, Notification<?>>() {
+            public Observable<?> call(Observable<? extends Throwable> t1) {
+                return t1.map(new Func1<Throwable, Void>() {
 
                     @Override
-                    public Notification<?> call(Notification<?> t1) {
-                        return Notification.createOnNext(null);
+                    public Void call(Throwable t1) {
+                        return null;
                     }
-                }).startWith(Notification.createOnNext(null));
+                }).startWith((Void) null);
             }
         }).subscribe(observer);
 
@@ -209,9 +208,9 @@ public void testOnCompletedFromNotificationHandler() {
         Observer<String> observer = mock(Observer.class);
         Observable<String> origin = Observable.create(new FuncWithErrors(1));
         TestSubscriber<String> subscriber = new TestSubscriber<String>(observer);
-        origin.retryWhen(new Func1<Observable<? extends Notification<?>>, Observable<? extends Notification<?>>>() {
+        origin.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
             @Override
-            public Observable<? extends Notification<?>> call(Observable<? extends Notification<?>> t1) {
+            public Observable<?> call(Observable<? extends Throwable> t1) {
                 return Observable.empty();
             }
         }).subscribe(subscriber);
@@ -229,9 +228,9 @@ public void testOnErrorFromNotificationHandler() {
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
         Observable<String> origin = Observable.create(new FuncWithErrors(2));
-        origin.retryWhen(new Func1<Observable<? extends Notification<?>>, Observable<? extends Notification<?>>>() {
+        origin.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
             @Override
-            public Observable<? extends Notification<?>> call(Observable<? extends Notification<?>> t1) {
+            public Observable<?> call(Observable<? extends Throwable> t1) {
                 return Observable.error(new RuntimeException());
             }
         }).subscribe(observer);