Fixing the generics for merge and lift

rxjava-contrib/rxjava-debug/src/main/java/rx/operators/DebugSubscriber.java

@@ -11,15 +11,15 @@ public final class DebugSubscriber<T> extends Subscriber<T> {
     private final Func1<T, T> onNextHook;
     final Action1<DebugNotification> events;
     final Observer<? super T> o;
-    Operator<T, ?> from = null;
-    Operator<?, T> to = null;
+    Operator<? extends T, ?> from = null;
+    Operator<?, ? super T> to = null;
 
     public DebugSubscriber(
             Func1<T, T> onNextHook,
             Action1<DebugNotification> _events,
             Subscriber<? super T> _o,
-            Operator<T, ?> _out,
-            Operator<?, T> _in) {
+            Operator<? extends T, ?> _out,
+            Operator<?, ? super T> _in) {
         super(_o);
         this.events = _events;
         this.o = _o;
@@ -47,19 +47,19 @@ public void onNext(T t) {
         o.onNext(onNextHook.call(t));
     }
 
-    public Operator<T, ?> getFrom() {
+    public Operator<? extends T, ?> getFrom() {
         return from;
     }
 
-    public void setFrom(Operator<T, ?> op) {
-        this.from = op;
+    public void setFrom(Operator<? extends T, ?> bind) {
+        this.from = bind;
     }
 
-    public Operator<?, T> getTo() {
+    public Operator<?, ? super T> getTo() {
         return to;
     }
 
-    public void setTo(Operator<?, T> op) {
+    public void setTo(Operator<?, ? super T> op) {
         this.to = op;
     }
 

rxjava-contrib/rxjava-debug/src/main/java/rx/plugins/DebugHook.java

@@ -63,7 +63,7 @@ public void call(Subscriber<? super T> o) {
     }
 
     @Override
-    public <T, R> Operator<R, T> onLift(final Operator<R, T> bind) {
+    public <T, R> Operator<? extends R, ? super T> onLift(final Operator<? extends R, ? super T> bind) {
         return new Operator<R, T>() {
             @Override
             public Subscriber<? super T> call(final Subscriber<? super R> o) {
@@ -78,7 +78,7 @@ public <T> Subscription onAdd(Subscriber<T> subscriber, Subscription s) {
     }
 
     @SuppressWarnings("unchecked")
-    private <R> Subscriber<? super R> wrapOutbound(Operator<R, ?> bind, Subscriber<? super R> o) {
+    private <R> Subscriber<? super R> wrapOutbound(Operator<? extends R, ?> bind, Subscriber<? super R> o) {
         if (o instanceof DebugSubscriber) {
             if (bind != null)
                 ((DebugSubscriber<R>) o).setFrom(bind);
@@ -88,7 +88,7 @@ private <R> Subscriber<? super R> wrapOutbound(Operator<R, ?> bind, Subscriber<?
     }
 
     @SuppressWarnings("unchecked")
-    private <T> Subscriber<? super T> wrapInbound(Operator<?, T> bind, Subscriber<? super T> o) {
+    private <T> Subscriber<? super T> wrapInbound(Operator<?, ? super T> bind, Subscriber<? super T> o) {
         if (o instanceof DebugSubscriber) {
             if (bind != null)
                 ((DebugSubscriber<T>) o).setTo(bind);

rxjava-contrib/rxjava-debug/src/main/java/rx/plugins/DebugNotification.java

@@ -13,17 +13,17 @@ public static enum Kind {
     }
 
     private final OnSubscribe<T> source;
-    private final Operator<T, ?> from;
+    private final Operator<? extends T, ?> from;
     private final Kind kind;
     private final Notification<T> notification;
-    private final Operator<?, T> to;
+    private final Operator<?, ? super T> to;
     private final long nanoTime;
     private final long threadId;
     private Observer o;
 
     public static <T> DebugNotification<T> createSubscribe(Observer<? super T> o, OnSubscribe<T> source) {
-        Operator<?, T> to = null;
-        Operator<T, ?> from = null;
+        Operator<?, ? super T> to = null;
+        Operator<? extends T, ?> from = null;
         if (o instanceof DebugSubscriber) {
             to = ((DebugSubscriber<T>) o).getTo();
             from = ((DebugSubscriber<T>) o).getFrom();
@@ -32,23 +32,23 @@ public static <T> DebugNotification<T> createSubscribe(Observer<? super T> o, On
         return new DebugNotification<T>(o, from, Kind.Subscribe, null, to, source);
     }
 
-    public static <T> DebugNotification<T> createOnNext(Observer<? super T> o, Operator<T, ?> from, T t, Operator<?, T> to) {
+    public static <T> DebugNotification<T> createOnNext(Observer<? super T> o, Operator<? extends T, ?> from, T t, Operator<?, ? super T> to) {
         return new DebugNotification<T>(o, from, Kind.OnNext, Notification.createOnNext(t), to, null);
     }
 
-    public static <T> DebugNotification<T> createOnError(Observer<? super T> o, Operator<T, ?> from, Throwable e, Operator<?, T> to) {
+    public static <T> DebugNotification<T> createOnError(Observer<? super T> o, Operator<? extends T, ?> from, Throwable e, Operator<?, ? super T> to) {
         return new DebugNotification<T>(o, from, Kind.OnError, Notification.<T> createOnError(e), to, null);
     }
 
-    public static <T> DebugNotification<T> createOnCompleted(Observer<? super T> o, Operator<T, ?> from, Operator<?, T> to) {
+    public static <T> DebugNotification<T> createOnCompleted(Observer<? super T> o, Operator<? extends T, ?> from, Operator<?, ? super T> to) {
         return new DebugNotification<T>(o, from, Kind.OnCompleted, Notification.<T> createOnCompleted(), to, null);
     }
 
-    public static <T> DebugNotification<T> createUnsubscribe(Observer<? super T> o, Operator<T, ?> from, Operator<?, T> to) {
+    public static <T> DebugNotification<T> createUnsubscribe(Observer<? super T> o, Operator<? extends T, ?> from, Operator<?, ? super T> to) {
         return new DebugNotification<T>(o, from, Kind.Unsubscribe, null, to, null);
     }
 
-    private DebugNotification(Observer o, Operator<T, ?> from, Kind kind, Notification<T> notification, Operator<?, T> to, OnSubscribe<T> source) {
+    private DebugNotification(Observer o, Operator<? extends T, ?> from, Kind kind, Notification<T> notification, Operator<?, ? super T> to, OnSubscribe<T> source) {
         this.o = (o instanceof SafeSubscriber) ? ((SafeSubscriber) o).getActual() : o;
         this.from = from;
         this.kind = kind;
@@ -59,15 +59,15 @@ private DebugNotification(Observer o, Operator<T, ?> from, Kind kind, Notificati
         this.threadId = Thread.currentThread().getId();
     }
 
-    public Operator<T, ?> getFrom() {
+    public Operator<? extends T, ?> getFrom() {
         return from;
     }
 
     public Notification<T> getNotification() {
         return notification;
     }
 
-    public Operator<?, T> getTo() {
+    public Operator<?, ? super T> getTo() {
         return to;
     }
 

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

@@ -261,7 +261,7 @@ public static interface OnSubscribeFunc<T> extends Function {
      * @param bind
      * @return an Observable that emits values that are the result of applying the bind function to the values of the current Observable
      */
-    public <R> Observable<R> lift(final Operator<R, T> bind) {
+    public <R> Observable<R> lift(final Operator<? extends R, ? super T> bind) {
         return new Observable<R>(new OnSubscribe<R>() {
             @Override
             public void call(Subscriber<? super R> o) {
@@ -1777,7 +1777,7 @@ public final static <T> Observable<T> merge(Iterable<? extends Observable<? exte
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
-        return source.lift(new OperatorMerge()); // any idea how to get these generics working?!
+        return source.lift(new OperatorMerge<T>());
     }
 
     /**
@@ -1821,7 +1821,7 @@ public final static <T> Observable<T> merge(Observable<? extends Observable<? ex
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2) {
-        return merge(from(t1, t2));
+        return merge(from(Arrays.asList(t1, t2)));
     }
 
     /**
@@ -1843,7 +1843,7 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3) {
-        return merge(from(t1, t2, t3));
+        return merge(from(Arrays.asList(t1, t2, t3)));
     }
 
     /**
@@ -1867,7 +1867,7 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4) {
-        return merge(from(t1, t2, t3, t4));
+        return merge(from(Arrays.asList(t1, t2, t3, t4)));
     }
 
     /**
@@ -1893,7 +1893,7 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5) {
-        return merge(from(t1, t2, t3, t4, t5));
+        return merge(from(Arrays.asList(t1, t2, t3, t4, t5)));
     }
 
     /**
@@ -1921,7 +1921,7 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6) {
-        return merge(from(t1, t2, t3, t4, t5, t6));
+        return merge(from(Arrays.asList(t1, t2, t3, t4, t5, t6)));
     }
 
     /**
@@ -1951,7 +1951,7 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7) {
-        return merge(from(t1, t2, t3, t4, t5, t6, t7));
+        return merge(from(Arrays.asList(t1, t2, t3, t4, t5, t6, t7)));
     }
 
     /**
@@ -1984,7 +1984,7 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
      */
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7, Observable<? extends T> t8) {
-        return merge(from(t1, t2, t3, t4, t5, t6, t7, t8));
+        return merge(from(Arrays.asList(t1, t2, t3, t4, t5, t6, t7, t8)));
     }
 
     /**
@@ -2019,7 +2019,7 @@ public final static <T> Observable<T> merge(Observable<? extends T> t1, Observab
      */
     // suppress because the types are checked by the method signature before using a vararg
     public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7, Observable<? extends T> t8, Observable<? extends T> t9) {
-        return merge(from(t1, t2, t3, t4, t5, t6, t7, t8, t9));
+        return merge(from(Arrays.asList(t1, t2, t3, t4, t5, t6, t7, t8, t9)));
     }
 
     /**
@@ -4863,7 +4863,7 @@ public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
      * @deprecated use {@link #flatMap(Func1)}
      */
     @Deprecated
-    public final <R> Observable<R> mapMany(Func1<? super T, ? extends Observable<? extends R>> func) {
+    public final <R> Observable<R> mapMany(Func1<? super T, ? extends Observable<R>> func) {
         return mergeMap(func);
     }
 

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

@@ -30,13 +30,13 @@
  * You can combine the items emitted by multiple Observables so that they act like a single
  * Observable, by using the merge operation.
  */
-public final class OperatorMerge<T> implements Operator<T, Observable<T>> {
+public final class OperatorMerge<T> implements Operator<T, Observable<? extends T>> {
 
     @Override
-    public Subscriber<Observable<T>> call(final Subscriber<? super T> outerOperation) {
+    public Subscriber<Observable<? extends T>> call(final Subscriber<? super T> outerOperation) {
 
         final Subscriber<T> o = new SynchronizedSubscriber<T>(outerOperation);
-        return new Subscriber<Observable<T>>(outerOperation) {
+        return new Subscriber<Observable<? extends T>>(outerOperation) {
 
             private volatile boolean completed = false;
             private final AtomicInteger runningCount = new AtomicInteger();
@@ -55,7 +55,7 @@ public void onError(Throwable e) {
             }
 
             @Override
-            public void onNext(Observable<T> innerObservable) {
+            public void onNext(Observable<? extends T> innerObservable) {
                 runningCount.incrementAndGet();
                 innerObservable.subscribe(new InnerObserver());
             }

rxjava-core/src/main/java/rx/plugins/RxJavaObservableExecutionHook.java

@@ -97,7 +97,7 @@ public <T> OnSubscribe<T> onCreate(OnSubscribe<T> f) {
         return f;
     }
 
-    public <T, R> Operator<R, T> onLift(final Operator<R, T> bind) {
+    public <T, R> Operator<? extends R, ? super T> onLift(final Operator<? extends R, ? super T> bind) {
         return bind;
     }