Timestamped, Notification and Future are now also treated as covariant

Author: jmhofer

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

@@ -850,7 +850,7 @@ public Observable<Timestamped<T>> timestamp() {
      *            be emitted by the resulting Observable
      * @return an Observable that emits the item from the source Future
      */
-    public static <T> Observable<T> from(Future<T> future) {
+    public static <T> Observable<T> from(Future<? extends T> future) {
         return create(OperationToObservableFuture.toObservableFuture(future));
     }
 
@@ -873,7 +873,7 @@ public static <T> Observable<T> from(Future<T> future) {
      *            be emitted by the resulting Observable
      * @return an Observable that emits the item from the source Future
      */
-    public static <T> Observable<T> from(Future<T> future, Scheduler scheduler) {
+    public static <T> Observable<T> from(Future<? extends T> future, Scheduler scheduler) {
         return create(OperationToObservableFuture.toObservableFuture(future)).subscribeOn(scheduler);
     }
 
@@ -899,7 +899,7 @@ public static <T> Observable<T> from(Future<T> future, Scheduler scheduler) {
      *            be emitted by the resulting Observable
      * @return an Observable that emits the item from the source {@link Future}
      */
-    public static <T> Observable<T> from(Future<T> future, long timeout, TimeUnit unit) {
+    public static <T> Observable<T> from(Future<? extends T> future, long timeout, TimeUnit unit) {
         return create(OperationToObservableFuture.toObservableFuture(future, timeout, unit));
     }
 
@@ -1703,7 +1703,7 @@ public Observable<T> observeOn(Scheduler scheduler) {
      */
     @SuppressWarnings("unchecked")
     public <T2> Observable<T2> dematerialize() {
-        return create(OperationDematerialize.dematerialize((Observable<? extends Notification<T2>>) this));
+        return create(OperationDematerialize.dematerialize((Observable<? extends Notification<? extends T2>>) this));
     }
 
     /**

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

@@ -45,21 +45,21 @@ public final class OperationDematerialize {
      * @return An observable sequence exhibiting the behavior corresponding to the source sequence's notification values.
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229047(v=vs.103).aspx">Observable.Dematerialize(TSource) Method </a>
      */
-    public static <T> Func1<Observer<? super T>, Subscription> dematerialize(final Observable<? extends Notification<T>> sequence) {
+    public static <T> Func1<Observer<? super T>, Subscription> dematerialize(final Observable<? extends Notification<? extends T>> sequence) {
         return new DematerializeObservable<T>(sequence);
     }
 
     private static class DematerializeObservable<T> implements Func1<Observer<? super T>, Subscription> {
 
-        private final Observable<? extends Notification<T>> sequence;
+        private final Observable<? extends Notification<? extends T>> sequence;
 
-        public DematerializeObservable(Observable<? extends Notification<T>> sequence) {
+        public DematerializeObservable(Observable<? extends Notification<? extends T>> sequence) {
             this.sequence = sequence;
         }
 
         @Override
         public Subscription call(final Observer<? super T> observer) {
-            return sequence.subscribe(new Observer<Notification<T>>() {
+            return sequence.subscribe(new Observer<Notification<? extends T>>() {
                 @Override
                 public void onCompleted() {
                 }
@@ -69,7 +69,7 @@ public void onError(Throwable e) {
                 }
 
                 @Override
-                public void onNext(Notification<T> value) {
+                public void onNext(Notification<? extends T> value) {
                     switch (value.getKind()) {
                     case OnNext:
                         observer.onNext(value.getValue());

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

@@ -47,15 +47,15 @@ public final class OperationMaterialize {
      * @return An observable sequence whose elements are the result of materializing the notifications of the given sequence.
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229453(v=VS.103).aspx">Observable.Materialize(TSource) Method </a>
      */
-    public static <T> Func1<Observer<? super Notification<T>>, Subscription> materialize(final Observable<T> sequence) {
+    public static <T> Func1<Observer<? super Notification<T>>, Subscription> materialize(final Observable<? extends T> sequence) {
         return new MaterializeObservable<T>(sequence);
     }
 
     private static class MaterializeObservable<T> implements Func1<Observer<? super Notification<T>>, Subscription> {
 
-        private final Observable<T> sequence;
+        private final Observable<? extends T> sequence;
 
-        public MaterializeObservable(Observable<T> sequence) {
+        public MaterializeObservable(Observable<? extends T> sequence) {
             this.sequence = sequence;
         }
 

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

@@ -37,7 +37,7 @@ public final class OperationTimestamp {
      *            the type of the input sequence.
      * @return a sequence of timestamped values created by adding timestamps to each item in the input sequence.
      */
-    public static <T> Func1<Observer<? super Timestamped<T>>, Subscription> timestamp(Observable<T> sequence) {
+    public static <T> Func1<Observer<? super Timestamped<T>>, Subscription> timestamp(Observable<? extends T> sequence) {
         return OperationMap.map(sequence, new Func1<T, Timestamped<T>>() {
             @Override
             public Timestamped<T> call(T value) {

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

@@ -40,17 +40,17 @@
  */
 public class OperationToObservableFuture {
     private static class ToObservableFuture<T> implements Func1<Observer<? super T>, Subscription> {
-        private final Future<T> that;
+        private final Future<? extends T> that;
         private final Long time;
         private final TimeUnit unit;
 
-        public ToObservableFuture(Future<T> that) {
+        public ToObservableFuture(Future<? extends T> that) {
             this.that = that;
             this.time = null;
             this.unit = null;
         }
 
-        public ToObservableFuture(Future<T> that, long time, TimeUnit unit) {
+        public ToObservableFuture(Future<? extends T> that, long time, TimeUnit unit) {
             this.that = that;
             this.time = time;
             this.unit = unit;
@@ -75,11 +75,11 @@ public Subscription call(Observer<? super T> observer) {
         }
     }
 
-    public static <T> Func1<Observer<? super T>, Subscription> toObservableFuture(final Future<T> that) {
+    public static <T> Func1<Observer<? super T>, Subscription> toObservableFuture(final Future<? extends T> that) {
         return new ToObservableFuture<T>(that);
     }
 
-    public static <T> Func1<Observer<? super T>, Subscription> toObservableFuture(final Future<T> that, long time, TimeUnit unit) {
+    public static <T> Func1<Observer<? super T>, Subscription> toObservableFuture(final Future<? extends T> that, long time, TimeUnit unit) {
         return new ToObservableFuture<T>(that, time, unit);
     }
 

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

@@ -27,7 +27,7 @@
     private final Observer<? super T> underlying;
     private final Scheduler scheduler;
 
-    private final ConcurrentLinkedQueue<Notification<T>> queue = new ConcurrentLinkedQueue<Notification<T>>();
+    private final ConcurrentLinkedQueue<Notification<? extends T>> queue = new ConcurrentLinkedQueue<Notification<? extends T>>();
     private final AtomicInteger counter = new AtomicInteger(0);
 
     public ScheduledObserver(Observer<? super T> underlying, Scheduler scheduler) {
@@ -50,7 +50,7 @@ public void onNext(final T args) {
         enqueue(new Notification<T>(args));
     }
 
-    private void enqueue(Notification<T> notification) {
+    private void enqueue(Notification<? extends T> notification) {
         // this must happen before 'counter' is used to provide synchronization between threads
         queue.offer(notification);
 
@@ -66,7 +66,7 @@ private void processQueue() {
         scheduler.schedule(new Action0() {
             @Override
             public void call() {
-                Notification<T> not = queue.poll();
+                Notification<? extends T> not = queue.poll();
 
                 switch (not.getKind()) {
                 case OnNext:

rxjava-core/src/main/java/rx/subjects/PublishSubject.java

@@ -68,7 +68,7 @@
 public class PublishSubject<T> extends Subject<T, T> {
     public static <T> PublishSubject<T> create() {
         final ConcurrentHashMap<Subscription, Observer<? super T>> observers = new ConcurrentHashMap<Subscription, Observer<? super T>>();
-        final AtomicReference<Notification<T>> terminalState = new AtomicReference<Notification<T>>();
+        final AtomicReference<Notification<? extends T>> terminalState = new AtomicReference<Notification<? extends T>>();
 
         Func1<Observer<? super T>, Subscription> onSubscribe = new Func1<Observer<? super T>, Subscription>() {
             @Override
@@ -111,7 +111,7 @@ public void unsubscribe() {
             }
 
             private Subscription checkTerminalState(Observer<? super T> observer) {
-                Notification<T> n = terminalState.get();
+                Notification<? extends T> n = terminalState.get();
                 if (n != null) {
                     // we are terminated to immediately emit and don't continue with subscription
                     if (n.isOnCompleted()) {
@@ -130,9 +130,9 @@ private Subscription checkTerminalState(Observer<? super T> observer) {
     }
 
     private final ConcurrentHashMap<Subscription, Observer<? super T>> observers;
-    private final AtomicReference<Notification<T>> terminalState;
+    private final AtomicReference<Notification<? extends T>> terminalState;
 
-    protected PublishSubject(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe, ConcurrentHashMap<Subscription, Observer<? super T>> observers, AtomicReference<Notification<T>> terminalState) {
+    protected PublishSubject(Func1<? super Observer<? super T>, ? extends Subscription> onSubscribe, ConcurrentHashMap<Subscription, Observer<? super T>> observers, AtomicReference<Notification<? extends T>> terminalState) {
         super(onSubscribe);
         this.observers = observers;
         this.terminalState = terminalState;

rxjava-core/src/test/java/rx/CovarianceTest.java

@@ -28,52 +28,41 @@ public ExtendedResult call(Media m, Rating r) {
         };
 
         Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(new Action1<Result>() {
-
             @Override
             public void call(Result t1) {
                 System.out.println("Result: " + t1);
             }
-
         });
 
         Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(new Action1<Result>() {
-
             @Override
             public void call(Result t1) {
                 System.out.println("Result: " + t1);
             }
-
         });
 
         Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(new Action1<ExtendedResult>() {
-
             @Override
             public void call(ExtendedResult t1) {
                 System.out.println("Result: " + t1);
             }
-
         });
 
         Observable.<Media, Rating, Result> zip(horrors, ratings, combine).toBlockingObservable().forEach(new Action1<Result>() {
-
             @Override
             public void call(Result t1) {
                 System.out.println("Result: " + t1);
             }
-
         });
 
         Observable.<Media, Rating, ExtendedResult> zip(horrors, ratings, combine).toBlockingObservable().forEach(new Action1<Result>() {
-
             @Override
             public void call(Result t1) {
                 System.out.println("Result: " + t1);
             }
-
         });
 
         Observable.<Movie, CoolRating, Result> zip(horrors, ratings, combine);
-
     }
 
     static class Media {