Generalized all the operators, too

jmhofer · jmhofer · commit 1ca5900c0fc5 · 2013-09-01T10:59:49.000+02:00
diff --git a/rxjava-core/src/main/java/rx/operators/OperationAll.java b/rxjava-core/src/main/java/rx/operators/OperationAll.java
@@ -34,18 +34,18 @@
  */
 public class OperationAll {
 
-    public static <T> Func1<Observer<? super Boolean>, Subscription> all(Observable<T> sequence, Func1<? super T, Boolean> predicate) {
+    public static <T> Func1<Observer<? super Boolean>, Subscription> all(Observable<? extends T> sequence, Func1<? super T, Boolean> predicate) {
         return new AllObservable<T>(sequence, predicate);
     }
 
     private static class AllObservable<T> implements Func1<Observer<? super Boolean>, Subscription> {
-        private final Observable<T> sequence;
+        private final Observable<? extends T> sequence;
         private final Func1<? super T, Boolean> predicate;
 
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
 
 
-        private AllObservable(Observable<T> sequence, Func1<? super T, Boolean> predicate) {
+        private AllObservable(Observable<? extends T> sequence, Func1<? super T, Boolean> predicate) {
             this.sequence = sequence;
             this.predicate = predicate;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationCache.java b/rxjava-core/src/main/java/rx/operators/OperationCache.java
@@ -47,7 +47,7 @@
  */
 public class OperationCache {
 
-    public static <T> Func1<Observer<? super T>, Subscription> cache(final Observable<T> source) {
+    public static <T> Func1<Observer<? super T>, Subscription> cache(final Observable<? extends T> source) {
         return new Func1<Observer<? super T>, Subscription>() {
 
             final AtomicBoolean subscribed = new AtomicBoolean(false);
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDefer.java b/rxjava-core/src/main/java/rx/operators/OperationDefer.java
@@ -37,12 +37,12 @@
  */
 public final class OperationDefer {
 
-    public static <T> Func1<Observer<? super T>, Subscription> defer(final Func0<? extends Observable<T>> observableFactory) {
+    public static <T> Func1<Observer<? super T>, Subscription> defer(final Func0<? extends Observable<? extends T>> observableFactory) {
 
         return new Func1<Observer<? super T>, Subscription>() {
             @Override
             public Subscription call(Observer<? super T> observer) {
-                Observable<T> obs = observableFactory.call();
+                Observable<? extends T> obs = observableFactory.call();
                 return obs.subscribe(observer);
             }
         };
diff --git a/rxjava-core/src/main/java/rx/operators/OperationFilter.java b/rxjava-core/src/main/java/rx/operators/OperationFilter.java
@@ -33,16 +33,16 @@
  */
 public final class OperationFilter<T> {
 
-    public static <T> Func1<Observer<? super T>, Subscription> filter(Observable<T> that, Func1<? super T, Boolean> predicate) {
+    public static <T> Func1<Observer<? super T>, Subscription> filter(Observable<? extends T> that, Func1<? super T, Boolean> predicate) {
         return new Filter<T>(that, predicate);
     }
 
     private static class Filter<T> implements Func1<Observer<? super T>, Subscription> {
 
-        private final Observable<T> that;
+        private final Observable<? extends T> that;
         private final Func1<? super T, Boolean> predicate;
 
-        public Filter(Observable<T> that, Func1<? super T, Boolean> predicate) {
+        public Filter(Observable<? extends T> that, Func1<? super T, Boolean> predicate) {
             this.that = that;
             this.predicate = predicate;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationFinally.java b/rxjava-core/src/main/java/rx/operators/OperationFinally.java
@@ -52,7 +52,7 @@ public final class OperationFinally {
      *         the given action will be called.
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh212133(v=vs.103).aspx">MSDN Observable.Finally method</a>
      */
-    public static <T> Func1<Observer<? super T>, Subscription> finallyDo(final Observable<T> sequence, final Action0 action) {
+    public static <T> Func1<Observer<? super T>, Subscription> finallyDo(final Observable<? extends T> sequence, final Action0 action) {
         return new Func1<Observer<? super T>, Subscription>() {
             @Override
             public Subscription call(Observer<? super T> observer) {
@@ -62,10 +62,10 @@ public Subscription call(Observer<? super T> observer) {
     }
 
     private static class Finally<T> implements Func1<Observer<? super T>, Subscription> {
-        private final Observable<T> sequence;
+        private final Observable<? extends T> sequence;
         private final Action0 finalAction;
 
-        Finally(final Observable<T> sequence, Action0 finalAction) {
+        Finally(final Observable<? extends T> sequence, Action0 finalAction) {
             this.sequence = sequence;
             this.finalAction = finalAction;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationGroupBy.java b/rxjava-core/src/main/java/rx/operators/OperationGroupBy.java
@@ -48,7 +48,7 @@
  */
 public final class OperationGroupBy {
 
-    public static <K, T, R> Func1<Observer<? super GroupedObservable<K, R>>, Subscription> groupBy(Observable<T> source, final Func1<? super T, ? extends K> keySelector, final Func1<? super T, ? extends R> elementSelector) {
+    public static <K, T, R> Func1<Observer<? super GroupedObservable<K, R>>, Subscription> groupBy(Observable<? extends T> source, final Func1<? super T, ? extends K> keySelector, final Func1<? super T, ? extends R> elementSelector) {
 
         final Observable<KeyValue<K, R>> keyval = source.map(new Func1<T, KeyValue<K, R>>() {
             @Override
@@ -63,7 +63,7 @@ public KeyValue<K, R> call(T t) {
         return new GroupBy<K, R>(keyval);
     }
 
-    public static <K, T> Func1<Observer<? super GroupedObservable<K, T>>, Subscription> groupBy(Observable<T> source, final Func1<? super T, ? extends K> keySelector) {
+    public static <K, T> Func1<Observer<? super GroupedObservable<K, T>>, Subscription> groupBy(Observable<? extends T> source, final Func1<? super T, ? extends K> keySelector) {
         return groupBy(source, keySelector, Functions.<T> identity());
     }
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationMostRecent.java b/rxjava-core/src/main/java/rx/operators/OperationMostRecent.java
@@ -37,7 +37,7 @@
  */
 public final class OperationMostRecent {
 
-    public static <T> Iterable<T> mostRecent(final Observable<T> source, T initialValue) {
+    public static <T> Iterable<T> mostRecent(final Observable<? extends T> source, T initialValue) {
 
         MostRecentObserver<T> mostRecentObserver = new MostRecentObserver<T>(initialValue);
         final MostRecentIterator<T> nextIterator = new MostRecentIterator<T>(mostRecentObserver);
diff --git a/rxjava-core/src/main/java/rx/operators/OperationMulticast.java b/rxjava-core/src/main/java/rx/operators/OperationMulticast.java
@@ -28,19 +28,19 @@
 import rx.util.functions.Func1;
 
 public class OperationMulticast {
-    public static <T, R> ConnectableObservable<R> multicast(Observable<T> source, final Subject<T, R> subject) {
+    public static <T, R> ConnectableObservable<R> multicast(Observable<? extends T> source, final Subject<T, R> subject) {
         return new MulticastConnectableObservable<T, R>(source, subject);
     }
 
     private static class MulticastConnectableObservable<T, R> extends ConnectableObservable<R> {
         private final Object lock = new Object();
 
-        private final Observable<T> source;
+        private final Observable<? extends T> source;
         private final Subject<T, R> subject;
 
         private Subscription subscription;
 
-        public MulticastConnectableObservable(Observable<T> source, final Subject<T, R> subject) {
+        public MulticastConnectableObservable(Observable<? extends T> source, final Subject<T, R> subject) {
             super(new Func1<Observer<? super R>, Subscription>() {
                 @Override
                 public Subscription call(Observer<? super R> observer) {
diff --git a/rxjava-core/src/main/java/rx/operators/OperationNext.java b/rxjava-core/src/main/java/rx/operators/OperationNext.java
@@ -47,7 +47,7 @@
  */
 public final class OperationNext {
 
-    public static <T> Iterable<T> next(final Observable<T> items) {
+    public static <T> Iterable<T> next(final Observable<? extends T> items) {
 
         NextObserver<T> nextObserver = new NextObserver<T>();
         final NextIterator<T> nextIterator = new NextIterator<T>(nextObserver);
@@ -65,9 +65,9 @@ public Iterator<T> iterator() {
 
     private static class NextIterator<T> implements Iterator<T> {
 
-        private final NextObserver<T> observer;
+        private final NextObserver<? extends T> observer;
 
-        private NextIterator(NextObserver<T> observer) {
+        private NextIterator(NextObserver<? extends T> observer) {
             this.observer = observer;
         }
 
@@ -99,8 +99,8 @@ public void remove() {
         }
     }
 
-    private static class NextObserver<T> implements Observer<Notification<T>> {
-        private final BlockingQueue<Notification<T>> buf = new ArrayBlockingQueue<Notification<T>>(1);
+    private static class NextObserver<T> implements Observer<Notification<? extends T>> {
+        private final BlockingQueue<Notification<? extends T>> buf = new ArrayBlockingQueue<Notification<? extends T>>(1);
         private final AtomicBoolean waiting = new AtomicBoolean(false);
 
         @Override
@@ -114,12 +114,12 @@ public void onError(Throwable e) {
         }
 
         @Override
-        public void onNext(Notification<T> args) {
+        public void onNext(Notification<? extends T> args) {
 
             if (waiting.getAndSet(false) || !args.isOnNext()) {
-                Notification<T> toOffer = args;
+                Notification<? extends T> toOffer = args;
                 while (!buf.offer(toOffer)) {
-                    Notification<T> concurrentItem = buf.poll();
+                    Notification<? extends T> concurrentItem = buf.poll();
 
                     // in case if we won race condition with onComplete/onError method
                     if (!concurrentItem.isOnNext()) {
@@ -135,7 +135,7 @@ public void await() {
         }
 
         public boolean isCompleted(boolean rethrowExceptionIfExists) {
-            Notification<T> lastItem = buf.peek();
+            Notification<? extends T> lastItem = buf.peek();
             if (lastItem == null) {
                 return false;
             }
@@ -152,7 +152,7 @@ public boolean isCompleted(boolean rethrowExceptionIfExists) {
         }
 
         public T takeNext() throws InterruptedException {
-            Notification<T> next = buf.take();
+            Notification<? extends T> next = buf.take();
 
             if (next.isOnError()) {
                 throw Exceptions.propagate(next.getThrowable());
diff --git a/rxjava-core/src/main/java/rx/operators/OperationObserveOn.java b/rxjava-core/src/main/java/rx/operators/OperationObserveOn.java
@@ -41,15 +41,15 @@
  */
 public class OperationObserveOn {
 
-    public static <T> Func1<Observer<? super T>, Subscription> observeOn(Observable<T> source, Scheduler scheduler) {
+    public static <T> Func1<Observer<? super T>, Subscription> observeOn(Observable<? extends T> source, Scheduler scheduler) {
         return new ObserveOn<T>(source, scheduler);
     }
 
     private static class ObserveOn<T> implements Func1<Observer<? super T>, Subscription> {
-        private final Observable<T> source;
+        private final Observable<? extends T> source;
         private final Scheduler scheduler;
 
-        public ObserveOn(Observable<T> source, Scheduler scheduler) {
+        public ObserveOn(Observable<? extends T> source, Scheduler scheduler) {
             this.source = source;
             this.scheduler = scheduler;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationOnErrorReturn.java b/rxjava-core/src/main/java/rx/operators/OperationOnErrorReturn.java
@@ -50,15 +50,15 @@
  */
 public final class OperationOnErrorReturn<T> {
 
-    public static <T> Func1<Observer<? super T>, Subscription> onErrorReturn(Observable<T> originalSequence, Func1<? super Throwable, ? extends T> resumeFunction) {
+    public static <T> Func1<Observer<? super T>, Subscription> onErrorReturn(Observable<? extends T> originalSequence, Func1<? super Throwable, ? extends T> resumeFunction) {
         return new OnErrorReturn<T>(originalSequence, resumeFunction);
     }
 
     private static class OnErrorReturn<T> implements Func1<Observer<? super T>, Subscription> {
         private final Func1<? super Throwable, ? extends T> resumeFunction;
-        private final Observable<T> originalSequence;
+        private final Observable<? extends T> originalSequence;
 
-        public OnErrorReturn(Observable<T> originalSequence, Func1<? super Throwable, ? extends T> resumeFunction) {
+        public OnErrorReturn(Observable<? extends T> originalSequence, Func1<? super Throwable, ? extends T> resumeFunction) {
             this.resumeFunction = resumeFunction;
             this.originalSequence = originalSequence;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationSample.java b/rxjava-core/src/main/java/rx/operators/OperationSample.java
@@ -48,27 +48,27 @@ public final class OperationSample {
     /**
      * Samples the observable sequence at each interval.
      */
-    public static <T> Func1<Observer<? super T>, Subscription> sample(final Observable<T> source, long period, TimeUnit unit) {
+    public static <T> Func1<Observer<? super T>, Subscription> sample(final Observable<? extends T> source, long period, TimeUnit unit) {
         return new Sample<T>(source, period, unit, Schedulers.executor(Executors.newSingleThreadScheduledExecutor()));
     }
 
     /**
      * Samples the observable sequence at each interval.
      */
-    public static <T> Func1<Observer<? super T>, Subscription> sample(final Observable<T> source, long period, TimeUnit unit, Scheduler scheduler) {
+    public static <T> Func1<Observer<? super T>, Subscription> sample(final Observable<? extends T> source, long period, TimeUnit unit, Scheduler scheduler) {
         return new Sample<T>(source, period, unit, scheduler);
     }
     
     private static class Sample<T> implements Func1<Observer<? super T>, Subscription> {
-        private final Observable<T> source;
+        private final Observable<? extends T> source;
         private final long period;
         private final TimeUnit unit;
         private final Scheduler scheduler;
         
         private final AtomicBoolean hasValue = new AtomicBoolean();
         private final AtomicReference<T> latestValue = new AtomicReference<T>();
         
-        private Sample(Observable<T> source, long interval, TimeUnit unit, Scheduler scheduler) {
+        private Sample(Observable<? extends T> source, long interval, TimeUnit unit, Scheduler scheduler) {
             this.source = source;
             this.period = interval;
             this.unit = unit;
diff --git a/rxjava-core/src/main/java/rx/operators/OperationScan.java b/rxjava-core/src/main/java/rx/operators/OperationScan.java
@@ -55,7 +55,7 @@ public final class OperationScan {
      * @return An observable sequence whose elements are the result of accumulating the output from the list of Observables.
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh212007%28v=vs.103%29.aspx">Observable.Scan(TSource, TAccumulate) Method (IObservable(TSource), TAccumulate, Func(TAccumulate, TSource, TAccumulate))</a>
      */
-    public static <T, R> Func1<Observer<? super R>, Subscription> scan(Observable<T> sequence, R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
+    public static <T, R> Func1<Observer<? super R>, Subscription> scan(Observable<? extends T> sequence, R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
         return new Accumulator<T, R>(sequence, initialValue, accumulator);
     }
 
@@ -70,17 +70,17 @@ public static <T, R> Func1<Observer<? super R>, Subscription> scan(Observable<T>
      * @return An observable sequence whose elements are the result of accumulating the output from the list of Observables.
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v=vs.103).aspx">Observable.Scan(TSource) Method (IObservable(TSource), Func(TSource, TSource, TSource))</a>
      */
-    public static <T> Func1<Observer<? super T>, Subscription> scan(Observable<T> sequence, Func2<? super T, ? super T, ? extends T> accumulator) {
+    public static <T> Func1<Observer<? super T>, Subscription> scan(Observable<? extends T> sequence, Func2<? super T, ? super T, ? extends T> accumulator) {
         return new AccuWithoutInitialValue<T>(sequence, accumulator);
     }
 
     private static class AccuWithoutInitialValue<T> implements Func1<Observer<? super T>, Subscription> {
-        private final Observable<T> sequence;
+        private final Observable<? extends T> sequence;
         private final Func2<? super T, ? super T, ? extends T> accumulatorFunction;
         
         private AccumulatingObserver<T, T> accumulatingObserver;
         
-        private AccuWithoutInitialValue(Observable<T> sequence, Func2<? super T, ? super T, ? extends T> accumulator) {
+        private AccuWithoutInitialValue(Observable<? extends T> sequence, Func2<? super T, ? super T, ? extends T> accumulator) {
             this.sequence = sequence;
             this.accumulatorFunction = accumulator;
         }
@@ -114,11 +114,11 @@ public void onCompleted() {
     }
     
     private static class Accumulator<T, R> implements Func1<Observer<? super R>, Subscription> {
-        private final Observable<T> sequence;
+        private final Observable<? extends T> sequence;
         private final R initialValue;
         private final Func2<? super R, ? super T, ? extends R> accumulatorFunction;
 
-        private Accumulator(Observable<T> sequence, R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
+        private Accumulator(Observable<? extends T> sequence, R initialValue, Func2<? super R, ? super T, ? extends R> accumulator) {
             this.sequence = sequence;
             this.initialValue = initialValue;
             this.accumulatorFunction = accumulator;
diff --git a/rxjava-core/src/main/java/rx/operators/OperationSkip.java b/rxjava-core/src/main/java/rx/operators/OperationSkip.java
@@ -47,7 +47,7 @@ public final class OperationSkip {
      * 
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229847(v=vs.103).aspx">Observable.Skip(TSource) Method</a>
      */
-    public static <T> Func1<Observer<? super T>, Subscription> skip(final Observable<T> items, final int num) {
+    public static <T> Func1<Observer<? super T>, Subscription> skip(final Observable<? extends T> items, final int num) {
         // wrap in a Observable so that if a chain is built up, then asynchronously subscribed to twice we will have 2 instances of Take<T> rather than 1 handing both, which is not thread-safe.
         return new Func1<Observer<? super T>, Subscription>() {
 
@@ -68,9 +68,9 @@ public Subscription call(Observer<? super T> observer) {
      */
     private static class Skip<T> implements Func1<Observer<? super T>, Subscription> {
         private final int num;
-        private final Observable<T> items;
+        private final Observable<? extends T> items;
 
-        Skip(final Observable<T> items, final int num) {
+        Skip(final Observable<? extends T> items, final int num) {
             this.num = num;
             this.items = items;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationSubscribeOn.java b/rxjava-core/src/main/java/rx/operators/OperationSubscribeOn.java
@@ -36,15 +36,15 @@
  */
 public class OperationSubscribeOn {
 
-    public static <T> Func1<Observer<? super T>, Subscription> subscribeOn(Observable<T> source, Scheduler scheduler) {
+    public static <T> Func1<Observer<? super T>, Subscription> subscribeOn(Observable<? extends T> source, Scheduler scheduler) {
         return new SubscribeOn<T>(source, scheduler);
     }
 
     private static class SubscribeOn<T> implements Func1<Observer<? super T>, Subscription> {
-        private final Observable<T> source;
+        private final Observable<? extends T> source;
         private final Scheduler scheduler;
 
-        public SubscribeOn(Observable<T> source, Scheduler scheduler) {
+        public SubscribeOn(Observable<? extends T> source, Scheduler scheduler) {
             this.source = source;
             this.scheduler = scheduler;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTake.java b/rxjava-core/src/main/java/rx/operators/OperationTake.java
@@ -50,7 +50,7 @@ public final class OperationTake {
      * @param num
      * @return the specified number of contiguous values from the start of the given observable sequence
      */
-    public static <T> Func1<Observer<? super T>, Subscription> take(final Observable<T> items, final int num) {
+    public static <T> Func1<Observer<? super T>, Subscription> take(final Observable<? extends T> items, final int num) {
         // wrap in a Func so that if a chain is built up, then asynchronously subscribed to twice we will have 2 instances of Take<T> rather than 1 handing both, which is not thread-safe.
         return new Func1<Observer<? super T>, Subscription>() {
 
@@ -74,11 +74,11 @@ public Subscription call(Observer<? super T> observer) {
      * @param <T>
      */
     private static class Take<T> implements Func1<Observer<? super T>, Subscription> {
-        private final Observable<T> items;
+        private final Observable<? extends T> items;
         private final int num;
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
 
-        private Take(Observable<T> items, int num) {
+        private Take(Observable<? extends T> items, int num) {
             this.items = items;
             this.num = num;
         }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTakeLast.java b/rxjava-core/src/main/java/rx/operators/OperationTakeLast.java
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTakeWhile.java b/rxjava-core/src/main/java/rx/operators/OperationTakeWhile.java
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToFuture.java b/rxjava-core/src/main/java/rx/operators/OperationToFuture.java
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToIterator.java b/rxjava-core/src/main/java/rx/operators/OperationToIterator.java
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToObservableIterable.java b/rxjava-core/src/main/java/rx/operators/OperationToObservableIterable.java
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToObservableList.java b/rxjava-core/src/main/java/rx/operators/OperationToObservableList.java
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToObservableSortedList.java b/rxjava-core/src/main/java/rx/operators/OperationToObservableSortedList.java

Original file line number	Diff line number	Diff line change
`@@ -48,7 +48,7 @@`
`48`	`48`	`*/`
`49`	`49`	`public final class OperationGroupBy {`
`50`	`50`
`51`		`- public static <K, T, R> Func1<Observer<? super GroupedObservable<K, R>>, Subscription> groupBy(Observable<T> source, final Func1<? super T, ? extends K> keySelector, final Func1<? super T, ? extends R> elementSelector) {`
	`51`	`+ public static <K, T, R> Func1<Observer<? super GroupedObservable<K, R>>, Subscription> groupBy(Observable<? extends T> source, final Func1<? super T, ? extends K> keySelector, final Func1<? super T, ? extends R> elementSelector) {`
`52`	`52`
`53`	`53`	`final Observable<KeyValue<K, R>> keyval = source.map(new Func1<T, KeyValue<K, R>>() {`
`54`	`54`	`@Override`
`@@ -63,7 +63,7 @@ public KeyValue<K, R> call(T t) {`
`63`	`63`	`return new GroupBy<K, R>(keyval);`
`64`	`64`	`}`
`65`	`65`
`66`		`- public static <K, T> Func1<Observer<? super GroupedObservable<K, T>>, Subscription> groupBy(Observable<T> source, final Func1<? super T, ? extends K> keySelector) {`
	`66`	`+ public static <K, T> Func1<Observer<? super GroupedObservable<K, T>>, Subscription> groupBy(Observable<? extends T> source, final Func1<? super T, ? extends K> keySelector) {`
`67`	`67`	`return groupBy(source, keySelector, Functions.<T> identity());`
`68`	`68`	`}`
`69`	`69`