lots of Observer<? super X>

Author: jmhofer

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/AbstractButtonSource.java

@@ -42,9 +42,9 @@ public enum AbstractButtonSource { ; // no instances
      * @see SwingObservable.fromButtonAction
      */
     public static Observable<ActionEvent> fromActionOf(final AbstractButton button) {
-        return Observable.create(new Func1<Observer<ActionEvent>, Subscription>() {
+        return Observable.create(new Func1<Observer<? super ActionEvent>, Subscription>() {
             @Override
-            public Subscription call(final Observer<ActionEvent> observer) {
+            public Subscription call(final Observer<? super ActionEvent> observer) {
                 final ActionListener listener = new ActionListener() {
                     @Override
                     public void actionPerformed(ActionEvent e) {

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/ComponentEventSource.java

@@ -36,9 +36,9 @@ public enum ComponentEventSource { ; // no instances
      * @see SwingObservable.fromComponentEvents
      */
     public static Observable<ComponentEvent> fromComponentEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<ComponentEvent>, Subscription>() {
+        return Observable.create(new Func1<Observer<? super ComponentEvent>, Subscription>() {
             @Override
-            public Subscription call(final Observer<ComponentEvent> observer) {
+            public Subscription call(final Observer<? super ComponentEvent> observer) {
                 final ComponentListener listener = new ComponentListener() {
                     @Override
                     public void componentHidden(ComponentEvent event) {

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/KeyEventSource.java

@@ -50,9 +50,9 @@ public enum KeyEventSource { ; // no instances
      * @see SwingObservable.fromKeyEvents(Component)
      */
     public static Observable<KeyEvent> fromKeyEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<KeyEvent>, Subscription>() {
+        return Observable.create(new Func1<Observer<? super KeyEvent>, Subscription>() {
             @Override
-            public Subscription call(final Observer<KeyEvent> observer) {
+            public Subscription call(final Observer<? super KeyEvent> observer) {
                 final KeyListener listener = new KeyListener() {
                     @Override
                     public void keyPressed(KeyEvent event) {

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/MouseEventSource.java

@@ -33,9 +33,9 @@ public enum MouseEventSource { ; // no instances
      * @see SwingObservable.fromMouseEvents
      */
     public static Observable<MouseEvent> fromMouseEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<MouseEvent>, Subscription>() {
+        return Observable.create(new Func1<Observer<? super MouseEvent>, Subscription>() {
             @Override
-            public Subscription call(final Observer<MouseEvent> observer) {
+            public Subscription call(final Observer<? super MouseEvent> observer) {
                 final MouseListener listener = new MouseListener() {
                     @Override
                     public void mouseClicked(MouseEvent event) {
@@ -78,9 +78,9 @@ public void call() {
      * @see SwingObservable.fromMouseMotionEvents
      */
     public static Observable<MouseEvent> fromMouseMotionEventsOf(final Component component) {
-        return Observable.create(new Func1<Observer<MouseEvent>, Subscription>() {
+        return Observable.create(new Func1<Observer<? super MouseEvent>, Subscription>() {
             @Override
-            public Subscription call(final Observer<MouseEvent> observer) {
+            public Subscription call(final Observer<? super MouseEvent> observer) {
                 final MouseMotionListener listener = new MouseMotionListener() {
                     @Override
                     public void mouseDragged(MouseEvent event) {

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

@@ -108,7 +108,7 @@ public class Observable<T> {
 
     private final static RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
 
-    private final Func1<? super Observer<T>, ? extends Subscription> onSubscribe;
+    private final Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe;
 
     /**
      * Observable with Function to execute when subscribed to.
@@ -119,7 +119,7 @@ public class Observable<T> {
      * @param onSubscribe
      *            {@link Func1} to be executed when {@link #subscribe(Observer)} is called.
      */
-    protected Observable(Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
+    protected Observable(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe) {
         this.onSubscribe = onSubscribe;
     }
 
@@ -155,9 +155,9 @@ protected Observable() {
      * @throws IllegalArgumentException
      *             if the {@link Observer} provided as the argument to {@code subscribe()} is {@code null}
      */
-    public Subscription subscribe(Observer<T> observer) {
+    public Subscription subscribe(Observer<? super T> observer) {
         // allow the hook to intercept and/or decorate
-        Func1<? super Observer<T>, ? extends Subscription> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
+        Func1<? super Observer<? super T>, ? extends Subscription> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
         // validate and proceed
         if (observer == null) {
             throw new IllegalArgumentException("observer can not be null");
@@ -233,7 +233,7 @@ public Subscription subscribe(Observer<T> observer) {
      * @throws IllegalArgumentException
      *             if an argument to {@code subscribe()} is {@code null}
      */
-    public Subscription subscribe(Observer<T> observer, Scheduler scheduler) {
+    public Subscription subscribe(Observer<? super T> observer, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(observer);
     }
 
@@ -242,7 +242,7 @@ public Subscription subscribe(Observer<T> observer, Scheduler scheduler) {
      * <p>
      * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
      */
-    private Subscription protectivelyWrapAndSubscribe(Observer<T> o) {
+    private Subscription protectivelyWrapAndSubscribe(Observer<? super T> o) {
         SafeObservableSubscription subscription = new SafeObservableSubscription();
         return subscription.wrap(subscribe(new SafeObserver<T>(subscription, o)));
     }
@@ -397,10 +397,10 @@ private void handleError(Throwable e) {
      */
     private static class NeverObservable<T> extends Observable<T> {
         public NeverObservable() {
-            super(new Func1<Observer<T>, Subscription>() {
+            super(new Func1<Observer<? super T>, Subscription>() {
 
                 @Override
-                public Subscription call(Observer<T> t1) {
+                public Subscription call(Observer<? super T> t1) {
                     return Subscriptions.empty();
                 }
 
@@ -417,7 +417,7 @@ public Subscription call(Observer<T> t1) {
     private static class ThrowObservable<T> extends Observable<T> {
 
         public ThrowObservable(final Throwable exception) {
-            super(new Func1<Observer<T>, Subscription>() {
+            super(new Func1<Observer<? super T>, Subscription>() {
 
                 /**
                  * Accepts an {@link Observer} and calls its {@link Observer#onError onError} method.
@@ -427,7 +427,7 @@ public ThrowObservable(final Throwable exception) {
                  * @return a reference to the subscription
                  */
                 @Override
-                public Subscription call(Observer<T> observer) {
+                public Subscription call(Observer<? super T> observer) {
                     observer.onError(exception);
                     return Subscriptions.empty();
                 }
@@ -461,7 +461,7 @@ public Subscription call(Observer<T> observer) {
      * @return an Observable that, when an {@link Observer} subscribes to it, will execute the given
      *         function
      */
-    public static <T> Observable<T> create(Func1<? super Observer<T>, ? extends Subscription> func) {
+    public static <T> Observable<T> create(Func1<? super Observer<? super T>, ? extends Subscription> func) {
         return new Observable<T>(func);
     }
 
@@ -918,7 +918,7 @@ public static <T> Observable<T> from(Future<T> future, long timeout, TimeUnit un
      *            Observable
      * @return an Observable that emits the zipped results
      */
-    public static <R, T0, T1> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Func2<? super T0, ? super T1, ? extends R> reduceFunction) {
+    public static <R, T0, T1> Observable<R> zip(Observable<? extends T0> w0, Observable<? extends T1> w1, Func2<? super T0, ? super T1, ? extends R> reduceFunction) {
         return create(OperationZip.zip(w0, w1, reduceFunction));
     }
 
@@ -992,7 +992,7 @@ public static <T> Observable<Boolean> sequenceEqual(Observable<T> first, Observa
      *            Observables, results in an item that will be emitted by the resulting Observable
      * @return an Observable that emits the zipped results
      */
-    public static <R, T0, T1, T2> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Observable<T2> w2, Func3<? super T0, ? super T1, ? super T2, ? extends R> function) {
+    public static <R, T0, T1, T2> Observable<R> zip(Observable<? extends T0> w0, Observable<? extends T1> w1, Observable<? extends T2> w2, Func3<? super T0, ? super T1, ? super T2, ? extends R> function) {
         return create(OperationZip.zip(w0, w1, w2, function));
     }
 

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -60,7 +60,7 @@
  */
 public class BlockingObservable<T> extends Observable<T> {
 
-    protected BlockingObservable(Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
+    protected BlockingObservable(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe) {
         super(onSubscribe);
     }
 
@@ -76,10 +76,10 @@ private BlockingObservable() {
      * Convert an Observable into a BlockingObservable.
      */
     public static <T> BlockingObservable<T> from(final Observable<T> o) {
-        return new BlockingObservable<T>(new Func1<Observer<T>, Subscription>() {
+        return new BlockingObservable<T>(new Func1<Observer<? super T>, Subscription>() {
 
             @Override
-            public Subscription call(Observer<T> observer) {
+            public Subscription call(Observer<? super T> observer) {
                 return o.subscribe(observer);
             }
         });
@@ -784,10 +784,10 @@ public void testToIterable() {
 
         @Test(expected = TestException.class)
         public void testToIterableWithException() {
-            BlockingObservable<String> obs = BlockingObservable.from(create(new Func1<Observer<String>, Subscription>() {
+            BlockingObservable<String> obs = BlockingObservable.from(create(new Func1<Observer<? super String>, Subscription>() {
 
                 @Override
-                public Subscription call(Observer<String> observer) {
+                public Subscription call(Observer<? super String> observer) {
                     observer.onNext("one");
                     observer.onError(new TestException());
                     return Subscriptions.empty();
@@ -807,10 +807,10 @@ public Subscription call(Observer<String> observer) {
         @Test
         public void testForEachWithError() {
             try {
-                BlockingObservable.from(Observable.create(new Func1<Observer<String>, Subscription>() {
+                BlockingObservable.from(Observable.create(new Func1<Observer<? super String>, Subscription>() {
 
                     @Override
-                    public Subscription call(final Observer<String> observer) {
+                    public Subscription call(final Observer<? super String> observer) {
                         final BooleanSubscription subscription = new BooleanSubscription();
                         new Thread(new Runnable() {
 

rxjava-core/src/main/java/rx/observables/ConnectableObservable.java

@@ -37,7 +37,7 @@
 
 public abstract class ConnectableObservable<T> extends Observable<T> {
 
-    protected ConnectableObservable(Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
+    protected ConnectableObservable(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe) {
         super(onSubscribe);
     }
 

rxjava-core/src/main/java/rx/observables/GroupedObservable.java

@@ -32,7 +32,7 @@
 public class GroupedObservable<K, T> extends Observable<T> {
     private final K key;
 
-    public GroupedObservable(K key, Func1<? super Observer<T>, ? extends Subscription> onSubscribe) {
+    public GroupedObservable(K key, Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe) {
         super(onSubscribe);
         this.key = key;
     }

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

@@ -34,11 +34,11 @@
  */
 public class OperationAll {
 
-    public static <T> Func1<Observer<Boolean>, Subscription> all(Observable<T> sequence, Func1<? super T, Boolean> predicate) {
+    public static <T> Func1<Observer<? super Boolean>, Subscription> all(Observable<T> sequence, Func1<? super T, Boolean> predicate) {
         return new AllObservable<T>(sequence, predicate);
     }
 
-    private static class AllObservable<T> implements Func1<Observer<Boolean>, Subscription> {
+    private static class AllObservable<T> implements Func1<Observer<? super Boolean>, Subscription> {
         private final Observable<T> sequence;
         private final Func1<? super T, Boolean> predicate;
 
@@ -52,17 +52,17 @@ private AllObservable(Observable<T> sequence, Func1<? super T, Boolean> predicat
 
 
         @Override
-        public Subscription call(final Observer<Boolean> observer) {
+        public Subscription call(final Observer<? super Boolean> observer) {
             return subscription.wrap(sequence.subscribe(new AllObserver(observer)));
 
         }
 
         private class AllObserver implements Observer<T> {
-            private final Observer<Boolean> underlying;
+            private final Observer<? super Boolean> underlying;
 
             private final AtomicBoolean status = new AtomicBoolean(true);
 
-            public AllObserver(Observer<Boolean> underlying) {
+            public AllObserver(Observer<? super Boolean> underlying) {
                 this.underlying = underlying;
             }