making Func0 covariant in its return type, cleaning up a few warnings, removing a bit of unused code

Author: jmhofer

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

@@ -574,7 +574,7 @@ public static Observable<Integer> range(int start, int count) {
      * @return an Observable whose {@link Observer}s trigger an invocation of the given Observable
      *         factory function
      */
-    public static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {
+    public static <T> Observable<T> defer(Func0<? extends Observable<T>> observableFactory) {
         return create(OperationDefer.defer(observableFactory));
     }
 
@@ -1074,7 +1074,7 @@ public static <R, T0, T1, T2, T3> Observable<R> combineLatest(Observable<T0> w0,
      *         An {@link Observable} which produces connected non-overlapping buffers, which are emitted
      *         when the current {@link Observable} created with the {@link Func0} argument produces a {@link BufferClosing} object.
      */
-    public Observable<List<T>> buffer(Func0<Observable<BufferClosing>> bufferClosingSelector) {
+    public Observable<List<T>> buffer(Func0<? extends Observable<BufferClosing>> bufferClosingSelector) {
         return create(OperationBuffer.buffer(this, bufferClosingSelector));
     }
 

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

@@ -69,7 +69,7 @@ public final class OperationBuffer {
      * @return
      *            the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> Func1<Observer<List<T>>, Subscription> buffer(final Observable<T> source, final Func0<Observable<BufferClosing>> bufferClosingSelector) {
+    public static <T> Func1<Observer<List<T>>, Subscription> buffer(final Observable<T> source, final Func0<? extends Observable<BufferClosing>> bufferClosingSelector) {
         return new Func1<Observer<List<T>>, Subscription>() {
             @Override
             public Subscription call(final Observer<List<T>> observer) {
@@ -444,10 +444,10 @@ public void stop() {
     private static class ObservableBasedSingleBufferCreator<T> implements BufferCreator<T> {
 
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
-        private final Func0<Observable<BufferClosing>> bufferClosingSelector;
+        private final Func0<? extends Observable<BufferClosing>> bufferClosingSelector;
         private final NonOverlappingBuffers<T> buffers;
 
-        public ObservableBasedSingleBufferCreator(NonOverlappingBuffers<T> buffers, Func0<Observable<BufferClosing>> bufferClosingSelector) {
+        public ObservableBasedSingleBufferCreator(NonOverlappingBuffers<T> buffers, Func0<? extends Observable<BufferClosing>> bufferClosingSelector) {
             this.buffers = buffers;
             this.bufferClosingSelector = bufferClosingSelector;
 

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

@@ -37,7 +37,7 @@
  */
 public final class OperationDefer {
 
-    public static <T> Func1<Observer<T>, Subscription> defer(final Func0<Observable<T>> observableFactory) {
+    public static <T> Func1<Observer<T>, Subscription> defer(final Func0<? extends Observable<T>> observableFactory) {
 
         return new Func1<Observer<T>, Subscription>() {
             @Override

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

@@ -80,9 +80,9 @@ public static <T0, T1, T2, T3, R> Func1<Observer<R>, Subscription> zip(Observabl
 
     @SuppressWarnings("unchecked")
     public static <R> Func1<Observer<R>, Subscription> zip(Collection<Observable<?>> ws, FuncN<R> zipFunction) {
-        Aggregator a = new Aggregator(zipFunction);
-        for (Observable w : ws) {
-            ZipObserver zipObserver = new ZipObserver(a, w);
+        Aggregator<R> a = new Aggregator<R>(zipFunction);
+        for (@SuppressWarnings("rawtypes") Observable w : ws) {
+            ZipObserver<R, Object> zipObserver = new ZipObserver<R, Object>(a, w);
             a.addObserver(zipObserver);
         }
         return a;
@@ -299,11 +299,13 @@ public static class UnitTest {
         @SuppressWarnings("unchecked")
         @Test
         public void testCollectionSizeDifferentThanFunction() {
-            FuncN<String> zipr = Functions.from(getConcatStringIntegerIntArrayZipr());
+            FuncN<String> zipr = Functions.fromFunc(getConcatStringIntegerIntArrayZipr());
+            //Func3<String, Integer, int[], String>
 
             /* define a Observer to receive aggregated events */
             Observer<String> aObserver = mock(Observer.class);
 
+            @SuppressWarnings("rawtypes")
             Collection ws = java.util.Collections.singleton(Observable.from("one", "two"));
             Observable<String> w = Observable.create(zip(ws, zipr));
             w.subscribe(aObserver);

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

@@ -132,6 +132,7 @@ public static class UnitTest {
         public void testNeverCompleted() {
         	AsyncSubject<String> subject = AsyncSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             subject.subscribe(aObserver);
 
@@ -153,6 +154,7 @@ private void assertNeverCompletedObserver(Observer<String> aObserver)
         public void testCompleted() {
         	AsyncSubject<String> subject = AsyncSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             subject.subscribe(aObserver);
 
@@ -175,6 +177,7 @@ private void assertCompletedObserver(Observer<String> aObserver)
         public void testError() {
         	AsyncSubject<String> subject = AsyncSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             subject.subscribe(aObserver);
 

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

@@ -247,6 +247,7 @@ public void unsubscribe() {
         public void testCompleted() {
             PublishSubject<String> subject = PublishSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             subject.subscribe(aObserver);
 
@@ -255,6 +256,7 @@ public void testCompleted() {
             subject.onNext("three");
             subject.onCompleted();
 
+            @SuppressWarnings("unchecked")
             Observer<String> anotherObserver = mock(Observer.class);
             subject.subscribe(anotherObserver);
 
@@ -275,17 +277,11 @@ private void assertCompletedObserver(Observer<String> aObserver)
             verify(aObserver, times(1)).onCompleted();
         }
 
-        private void assertNeverObserver(Observer<String> aObserver)
-        {
-            verify(aObserver, Mockito.never()).onNext(any(String.class));
-            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
-            verify(aObserver, Mockito.never()).onCompleted();
-        }
-
         @Test
         public void testError() {
             PublishSubject<String> subject = PublishSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             subject.subscribe(aObserver);
 
@@ -294,6 +290,7 @@ public void testError() {
             subject.onNext("three");
             subject.onError(testException);
 
+            @SuppressWarnings("unchecked")
             Observer<String> anotherObserver = mock(Observer.class);
             subject.subscribe(anotherObserver);
 
@@ -318,6 +315,7 @@ private void assertErrorObserver(Observer<String> aObserver)
         public void testSubscribeMidSequence() {
             PublishSubject<String> subject = PublishSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             subject.subscribe(aObserver);
 
@@ -326,6 +324,7 @@ public void testSubscribeMidSequence() {
 
             assertObservedUntilTwo(aObserver);
 
+            @SuppressWarnings("unchecked")
             Observer<String> anotherObserver = mock(Observer.class);
             subject.subscribe(anotherObserver);
 
@@ -349,6 +348,7 @@ private void assertCompletedStartingWithThreeObserver(Observer<String> aObserver
         public void testUnsubscribeFirstObserver() {
             PublishSubject<String> subject = PublishSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             Subscription subscription = subject.subscribe(aObserver);
 
@@ -358,6 +358,7 @@ public void testUnsubscribeFirstObserver() {
             subscription.unsubscribe();
             assertObservedUntilTwo(aObserver);
 
+            @SuppressWarnings("unchecked")
             Observer<String> anotherObserver = mock(Observer.class);
             subject.subscribe(anotherObserver);
 
@@ -391,6 +392,7 @@ private void assertObservedUntilTwo(Observer<String> aObserver)
         public void testUnsubscribeAfterOnCompleted() {
             PublishSubject<String> subject = PublishSubject.create();
 
+            @SuppressWarnings("unchecked")
             Observer<String> anObserver = mock(Observer.class);
             subject.subscribe(anObserver);
 

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

@@ -54,7 +54,7 @@ public UnsubscribeTester() {
      * @param <T>
      *            The type of object passed by the Observable
      */
-    public static <T, O extends Observable<T>> void test(Func0<O> provider, Action1<? super O> generateOnCompleted, Action1<? super O> generateOnError, Action1<? super O> generateOnNext)
+    public static <T, O extends Observable<T>> void test(Func0<? extends O> provider, Action1<? super O> generateOnCompleted, Action1<? super O> generateOnError, Action1<? super O> generateOnNext)
     {
         if (generateOnCompleted != null) {
             O observable = provider.call();

rxjava-core/src/main/java/rx/util/functions/Action.java

@@ -20,6 +20,6 @@
  * <p>
  * Marker interface to allow instanceof checks.
  */
-public interface Action extends Function {
+public interface Action extends Function<Void> {
 
 }

rxjava-core/src/main/java/rx/util/functions/Action0.java

@@ -15,6 +15,6 @@
  */
 package rx.util.functions;
 
-public interface Action0 extends Function, Action {
+public interface Action0 extends Action {
     public void call();
 }

rxjava-core/src/main/java/rx/util/functions/Action1.java

@@ -15,6 +15,6 @@
  */
 package rx.util.functions;
 
-public interface Action1<T1> extends Function, Action {
+public interface Action1<T1> extends Action {
     public void call(T1 t1);
 }