enhance generics doOnError doOnRequest (#4555)

Author: davidmoten

src/main/java/rx/Observable.java

@@ -5879,7 +5879,7 @@ public final Observable<T> doOnEach(Observer<? super T> observer) {
      * @return the source Observable with the side-effecting behavior applied
      * @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
      */
-    public final Observable<T> doOnError(final Action1<Throwable> onError) {
+    public final Observable<T> doOnError(final Action1<? super Throwable> onError) {
         Action1<T> onNext = Actions.empty();
         Action0 onCompleted = Actions.empty();
         Observer<T> observer = new ActionObserver<T>(onNext, onError, onCompleted);
@@ -5934,7 +5934,7 @@ public final Observable<T> doOnNext(final Action1<? super T> onNext) {
      *      documentation: Do</a>
      * @since 1.2
      */
-    public final Observable<T> doOnRequest(final Action1<Long> onRequest) {
+    public final Observable<T> doOnRequest(final Action1<? super Long> onRequest) {
         return lift(new OperatorDoOnRequest<T>(onRequest));
     }
 

src/main/java/rx/internal/operators/OperatorDoOnRequest.java

@@ -28,9 +28,9 @@
  */
 public class OperatorDoOnRequest<T> implements Operator<T, T> {
 
-    final Action1<Long> request;
+    final Action1<? super Long> request;
 
-    public OperatorDoOnRequest(Action1<Long> request) {
+    public OperatorDoOnRequest(Action1<? super Long> request) {
         this.request = request;
     }
 

src/main/java/rx/internal/util/ActionObserver.java

@@ -25,10 +25,10 @@
 public final class ActionObserver<T> implements Observer<T> {
 
     final Action1<? super T> onNext;
-    final Action1<Throwable> onError;
+    final Action1<? super Throwable> onError;
     final Action0 onCompleted;
 
-    public ActionObserver(Action1<? super T> onNext, Action1<Throwable> onError, Action0 onCompleted) {
+    public ActionObserver(Action1<? super T> onNext, Action1<? super Throwable> onError, Action0 onCompleted) {
         this.onNext = onNext;
         this.onError = onError;
         this.onCompleted = onCompleted;

src/test/java/rx/ObservableDoOnTest.java

@@ -27,6 +27,7 @@
 
 import rx.functions.Action0;
 import rx.functions.Action1;
+import rx.observers.TestSubscriber;
 
 public class ObservableDoOnTest {
 
@@ -65,6 +66,21 @@ public void call(Throwable v) {
         assertNotNull(t);
         assertEquals(t, r.get());
     }
+    
+    @Test
+    public void testDoOnErrorWithActionOfTypeObject() {
+        final AtomicReference<Boolean> r = new AtomicReference<Boolean>();
+        TestSubscriber<String> ts = TestSubscriber.create();
+        Observable.<String> error(new RuntimeException("an error"))
+            .doOnError(new Action1<Object>() {
+
+                @Override
+                public void call(Object v) {
+                    r.set(true);
+                }
+            }).subscribe(ts);
+        assertTrue(r.get());
+    }
 
     @Test
     public void testDoOnCompleted() {

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

@@ -27,6 +27,7 @@
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.functions.*;
+import rx.observers.TestSubscriber;
 
 public class OperatorDoOnRequestTest {
 
@@ -140,4 +141,19 @@ public void setProducer(Producer p) {
             Assert.assertEquals(Arrays.asList(0L, 1L), requested);
         }
     }
+    
+    @Test
+    public void canCallDoOnRequestWithActionOfTypeObject() {
+        final AtomicReference<Boolean> r = new AtomicReference<Boolean>();
+        TestSubscriber<String> ts = TestSubscriber.create();
+        Observable.just("a").doOnRequest(
+            new Action1<Object>() {
+
+                @Override
+                public void call(Object v) {
+                    r.set(true);
+                }
+            }).subscribe(ts);
+        assertTrue(r.get());
+    }
 }