generalized BlockingObservable and the execution hook further

Author: jmhofer

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

@@ -75,7 +75,7 @@ private BlockingObservable() {
     /**
      * Convert an Observable into a BlockingObservable.
      */
-    public static <T> BlockingObservable<T> from(final Observable<T> o) {
+    public static <T> BlockingObservable<T> from(final Observable<? extends T> o) {
         return new BlockingObservable<T>(new Func1<Observer<? super T>, Subscription>() {
 
             @Override
@@ -97,7 +97,7 @@ public Subscription call(Observer<? super T> observer) {
      *            the type of items emitted by the source {@link Observable}
      * @return an {@link Iterator} that can iterate over the items emitted by the {@link Observable}
      */
-    public static <T> Iterator<T> toIterator(Observable<T> source) {
+    public static <T> Iterator<T> toIterator(Observable<? extends T> source) {
         return OperationToIterator.toIterator(source);
     }
 
@@ -110,7 +110,7 @@ public static <T> Iterator<T> toIterator(Observable<T> source) {
      *            the source {@link Observable}
      * @return the last item emitted by the source {@link Observable}
      */
-    public static <T> T last(final Observable<T> source) {
+    public static <T> T last(final Observable<? extends T> source) {
         return from(source).last();
     }
 
@@ -126,7 +126,7 @@ public static <T> T last(final Observable<T> source) {
      * @return the last item emitted by the {@link Observable} for which the predicate function
      *         returns <code>true</code>
      */
-    public static <T> T last(final Observable<T> source, final Func1<? super T, Boolean> predicate) {
+    public static <T> T last(final Observable<? extends T> source, final Func1<? super T, Boolean> predicate) {
         return last(source.filter(predicate));
     }
 
@@ -145,7 +145,7 @@ public static <T> T last(final Observable<T> source, final Func1<? super T, Bool
      * @return the last item emitted by an {@link Observable}, or the default value if no item is
      *         emitted
      */
-    public static <T> T lastOrDefault(Observable<T> source, T defaultValue) {
+    public static <T> T lastOrDefault(Observable<? extends T> source, T defaultValue) {
         return from(source).lastOrDefault(defaultValue);
     }
 
@@ -166,7 +166,7 @@ public static <T> T lastOrDefault(Observable<T> source, T defaultValue) {
      * @return the last item emitted by an {@link Observable} that matches the predicate, or the
      *         default value if no matching item is emitted
      */
-    public static <T> T lastOrDefault(Observable<T> source, T defaultValue, Func1<? super T, Boolean> predicate) {
+    public static <T> T lastOrDefault(Observable<? extends T> source, T defaultValue, Func1<? super T, Boolean> predicate) {
         return lastOrDefault(source.filter(predicate), defaultValue);
     }
 
@@ -186,7 +186,7 @@ public static <T> T lastOrDefault(Observable<T> source, T defaultValue, Func1<?
      * @return an {@link Iterable} that on each iteration returns the item that the
      *         {@link Observable} has most recently emitted
      */
-    public static <T> Iterable<T> mostRecent(Observable<T> source, T initialValue) {
+    public static <T> Iterable<T> mostRecent(Observable<? extends T> source, T initialValue) {
         return OperationMostRecent.mostRecent(source, initialValue);
     }
 
@@ -203,12 +203,12 @@ public static <T> Iterable<T> mostRecent(Observable<T> source, T initialValue) {
      * @return an {@link Iterable} that blocks upon each iteration until the {@link Observable}
      *         emits a new item, whereupon the Iterable returns that item
      */
-    public static <T> Iterable<T> next(Observable<T> items) {
+    public static <T> Iterable<T> next(Observable<? extends T> items) {
         return OperationNext.next(items);
     }
 
-    private static <T> T _singleOrDefault(BlockingObservable<T> source, boolean hasDefault, T defaultValue) {
-        Iterator<T> it = source.toIterable().iterator();
+    private static <T> T _singleOrDefault(BlockingObservable<? extends T> source, boolean hasDefault, T defaultValue) {
+        Iterator<? extends T> it = source.toIterable().iterator();
 
         if (!it.hasNext()) {
             if (hasDefault) {
@@ -238,7 +238,7 @@ private static <T> T _singleOrDefault(BlockingObservable<T> source, boolean hasD
      * @throws IllegalStateException
      *             if the {@link Observable} does not emit exactly one item
      */
-    public static <T> T single(Observable<T> source) {
+    public static <T> T single(Observable<? extends T> source) {
         return from(source).single();
     }
 
@@ -257,7 +257,7 @@ public static <T> T single(Observable<T> source) {
      *             if the {@link Observable} does not emit exactly one item that matches the
      *             predicate
      */
-    public static <T> T single(Observable<T> source, Func1<? super T, Boolean> predicate) {
+    public static <T> T single(Observable<? extends T> source, Func1<? super T, Boolean> predicate) {
         return from(source).single(predicate);
     }
 
@@ -274,7 +274,7 @@ public static <T> T single(Observable<T> source, Func1<? super T, Boolean> predi
      * @return the single item emitted by the source {@link Observable}, or a default value if no
      *         value is emitted
      */
-    public static <T> T singleOrDefault(Observable<T> source, T defaultValue) {
+    public static <T> T singleOrDefault(Observable<? extends T> source, T defaultValue) {
         return from(source).singleOrDefault(defaultValue);
     }
 
@@ -293,7 +293,7 @@ public static <T> T singleOrDefault(Observable<T> source, T defaultValue) {
      * @return the single item emitted by the source {@link Observable} that matches the predicate,
      *         or a default value if no such value is emitted
      */
-    public static <T> T singleOrDefault(Observable<T> source, T defaultValue, Func1<? super T, Boolean> predicate) {
+    public static <T> T singleOrDefault(Observable<? extends T> source, T defaultValue, Func1<? super T, Boolean> predicate) {
         return from(source).singleOrDefault(defaultValue, predicate);
     }
 
@@ -310,7 +310,7 @@ public static <T> T singleOrDefault(Observable<T> source, T defaultValue, Func1<
      *            the source {@link Observable}
      * @return a Future that expects a single item to be emitted by the source {@link Observable}
      */
-    public static <T> Future<T> toFuture(final Observable<T> source) {
+    public static <T> Future<T> toFuture(final Observable<? extends T> source) {
         return OperationToFuture.toFuture(source);
     }
 
@@ -323,7 +323,7 @@ public static <T> Future<T> toFuture(final Observable<T> source) {
      *            the source {@link Observable}
      * @return an {@link Iterable} version of the underlying {@link Observable}
      */
-    public static <T> Iterable<T> toIterable(final Observable<T> source) {
+    public static <T> Iterable<T> toIterable(final Observable<? extends T> source) {
         return from(source).toIterable();
     }
 
@@ -334,7 +334,7 @@ public static <T> Iterable<T> toIterable(final Observable<T> source) {
      * 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)));
     }

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

@@ -63,7 +63,7 @@ public abstract class RxJavaObservableExecutionHook {
      *            original {@link Subscription}
      * @return {@link Subscription} subscription that can be modified, decorated, replaced or just returned as a pass-thru.
      */
-    public <T> Subscription onSubscribeReturn(Observable<T> observableInstance, Subscription subscription) {
+    public <T> Subscription onSubscribeReturn(Observable<? extends T> observableInstance, Subscription subscription) {
         // pass-thru by default
         return subscription;
     }
@@ -80,7 +80,7 @@ public <T> Subscription onSubscribeReturn(Observable<T> observableInstance, Subs
      *            Throwable thrown by {@link Observable#subscribe(Observer)}
      * @return Throwable that can be decorated, replaced or just returned as a pass-thru.
      */
-    public <T> Throwable onSubscribeError(Observable<T> observableInstance, Throwable e) {
+    public <T> Throwable onSubscribeError(Observable<? extends T> observableInstance, Throwable e) {
         // pass-thru by default
         return e;
     }