Merge pull request ReactiveX#133 from benjchristensen/fixes

Various fixes

Author: benjchristensen

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

@@ -149,7 +149,7 @@ public boolean equals(Object obj) {
             return true;
         if (obj.getClass() != getClass())
             return false;
-        Notification notification = (Notification) obj;
+        Notification<?> notification = (Notification<?>) obj;
         if (notification.getKind() != getKind())
             return false;
         if (hasValue() && !getValue().equals(notification.getValue()))

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

@@ -413,6 +413,13 @@ public static <T> Observable<T> create(Func1<Observer<T>, Subscription> func) {
         return new Observable<T>(func);
     }
 
+    /*
+     * Private version that creates a 'trusted' Observable to allow performance optimizations.
+     */
+    private static <T> Observable<T> _create(Func1<Observer<T>, Subscription> func) {
+        return new Observable<T>(func, true);
+    }
+
     /**
      * Creates an Observable that will execute the given function when a {@link Observer} subscribes to it.
      * <p>
@@ -484,7 +491,7 @@ public static <T> Observable<T> error(Exception exception) {
      * @return an Observable that emits only those items in the original Observable that the filter evaluates as true
      */
     public static <T> Observable<T> filter(Observable<T> that, Func1<T, Boolean> predicate) {
-        return create(OperationFilter.filter(that, predicate));
+        return _create(OperationFilter.filter(that, predicate));
     }
 
     /**
@@ -577,7 +584,7 @@ public static <T> Observable<T> just(T value) {
      *         by the source Observable
      */
     public static <T> Observable<T> last(final Observable<T> that) {
-        return create(OperationLast.last(that));
+        return _create(OperationLast.last(that));
     }
 
     /**
@@ -598,7 +605,7 @@ public static <T> Observable<T> last(final Observable<T> that) {
      *         in the sequence emitted by the source Observable
      */
     public static <T, R> Observable<R> map(Observable<T> sequence, Func1<T, R> func) {
-        return create(OperationMap.map(sequence, func));
+        return _create(OperationMap.map(sequence, func));
     }
 
     /**
@@ -654,7 +661,7 @@ public R call(T t1) {
      *         the Observables obtained from this transformation
      */
     public static <T, R> Observable<R> mapMany(Observable<T> sequence, Func1<T, Observable<R>> func) {
-        return create(OperationMap.mapMany(sequence, func));
+        return _create(OperationMap.mapMany(sequence, func));
     }
 
     /**
@@ -703,7 +710,7 @@ public R call(T t1) {
      * @see http://msdn.microsoft.com/en-us/library/hh229453(v=VS.103).aspx
      */
     public static <T> Observable<Notification<T>> materialize(final Observable<T> sequence) {
-        return create(OperationMaterialize.materialize(sequence));
+        return _create(OperationMaterialize.materialize(sequence));
     }
 
     /**
@@ -720,7 +727,7 @@ public static <T> Observable<Notification<T>> materialize(final Observable<T> se
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> merge(List<Observable<T>> source) {
-        return create(OperationMerge.merge(source));
+        return _create(OperationMerge.merge(source));
     }
 
     /**
@@ -737,7 +744,7 @@ public static <T> Observable<T> merge(List<Observable<T>> source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> merge(Observable<Observable<T>> source) {
-        return create(OperationMerge.merge(source));
+        return _create(OperationMerge.merge(source));
     }
 
     /**
@@ -754,7 +761,7 @@ public static <T> Observable<T> merge(Observable<Observable<T>> source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> merge(Observable<T>... source) {
-        return create(OperationMerge.merge(source));
+        return _create(OperationMerge.merge(source));
     }
 
     /**
@@ -770,7 +777,7 @@ public static <T> Observable<T> merge(Observable<T>... source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.concat(v=vs.103).aspx">MSDN: Observable.Concat Method</a>
      */
     public static <T> Observable<T> concat(Observable<T>... source) {
-        return create(OperationConcat.concat(source));
+        return _create(OperationConcat.concat(source));
     }
 
     /**
@@ -789,7 +796,7 @@ public static <T> Observable<T> concat(Observable<T>... source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> mergeDelayError(List<Observable<T>> source) {
-        return create(OperationMergeDelayError.mergeDelayError(source));
+        return _create(OperationMergeDelayError.mergeDelayError(source));
     }
 
     /**
@@ -808,7 +815,7 @@ public static <T> Observable<T> mergeDelayError(List<Observable<T>> source) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> mergeDelayError(Observable<Observable<T>> source) {
-        return create(OperationMergeDelayError.mergeDelayError(source));
+        return _create(OperationMergeDelayError.mergeDelayError(source));
     }
 
     /**
@@ -827,7 +834,7 @@ public static <T> Observable<T> mergeDelayError(Observable<Observable<T>> source
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge Method</a>
      */
     public static <T> Observable<T> mergeDelayError(Observable<T>... source) {
-        return create(OperationMergeDelayError.mergeDelayError(source));
+        return _create(OperationMergeDelayError.mergeDelayError(source));
     }
 
     /**
@@ -916,7 +923,7 @@ public void unsubscribe() {
      * @return the source Observable, with its behavior modified as described
      */
     public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, final Func1<Exception, Observable<T>> resumeFunction) {
-        return create(OperationOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(that, resumeFunction));
+        return _create(OperationOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(that, resumeFunction));
     }
 
     /**
@@ -980,7 +987,7 @@ public Observable<T> call(Exception e) {
      * @return the source Observable, with its behavior modified as described
      */
     public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, final Observable<T> resumeSequence) {
-        return create(OperationOnErrorResumeNextViaObservable.onErrorResumeNextViaObservable(that, resumeSequence));
+        return _create(OperationOnErrorResumeNextViaObservable.onErrorResumeNextViaObservable(that, resumeSequence));
     }
 
     /**
@@ -1004,7 +1011,7 @@ public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, fina
      * @return the source Observable, with its behavior modified as described
      */
     public static <T> Observable<T> onErrorReturn(final Observable<T> that, Func1<Exception, T> resumeFunction) {
-        return create(OperationOnErrorReturn.onErrorReturn(that, resumeFunction));
+        return _create(OperationOnErrorReturn.onErrorReturn(that, resumeFunction));
     }
 
     /**
@@ -1034,7 +1041,7 @@ public static <T> Observable<T> onErrorReturn(final Observable<T> that, Func1<Ex
      * @see <a href="http://en.wikipedia.org/wiki/Fold_(higher-order_function)">Wikipedia: Fold (higher-order function)</a>
      */
     public static <T> Observable<T> reduce(Observable<T> sequence, Func2<T, T, T> accumulator) {
-        return last(create(OperationScan.scan(sequence, accumulator)));
+        return last(_create(OperationScan.scan(sequence, accumulator)));
     }
 
     /**
@@ -1106,7 +1113,7 @@ public T call(T t1, T t2) {
      * @see <a href="http://en.wikipedia.org/wiki/Fold_(higher-order_function)">Wikipedia: Fold (higher-order function)</a>
      */
     public static <T> Observable<T> reduce(Observable<T> sequence, T initialValue, Func2<T, T, T> accumulator) {
-        return last(create(OperationScan.scan(sequence, initialValue, accumulator)));
+        return last(_create(OperationScan.scan(sequence, initialValue, accumulator)));
     }
 
     /**
@@ -1170,7 +1177,7 @@ public T call(T t1, T t2) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v%3Dvs.103).aspx">MSDN: Observable.Scan</a>
      */
     public static <T> Observable<T> scan(Observable<T> sequence, Func2<T, T, T> accumulator) {
-        return create(OperationScan.scan(sequence, accumulator));
+        return _create(OperationScan.scan(sequence, accumulator));
     }
 
     /**
@@ -1228,7 +1235,7 @@ public T call(T t1, T t2) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211665(v%3Dvs.103).aspx">MSDN: Observable.Scan</a>
      */
     public static <T> Observable<T> scan(Observable<T> sequence, T initialValue, Func2<T, T, T> accumulator) {
-        return create(OperationScan.scan(sequence, initialValue, accumulator));
+        return _create(OperationScan.scan(sequence, initialValue, accumulator));
     }
 
     /**
@@ -1282,7 +1289,7 @@ public T call(T t1, T t2) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229847(v=vs.103).aspx">MSDN: Observable.Skip Method</a>
      */
     public static <T> Observable<T> skip(final Observable<T> items, int num) {
-        return create(OperationSkip.skip(items, num));
+        return _create(OperationSkip.skip(items, num));
     }
 
     /**
@@ -1300,7 +1307,7 @@ public static <T> Observable<T> skip(final Observable<T> items, int num) {
      * @return an Observable that is a chronologically well-behaved version of the source Observable
      */
     public static <T> Observable<T> synchronize(Observable<T> observable) {
-        return create(OperationSynchronize.synchronize(observable));
+        return _create(OperationSynchronize.synchronize(observable));
     }
 
     /**
@@ -1322,7 +1329,7 @@ public static <T> Observable<T> synchronize(Observable<T> observable) {
      *         Observable
      */
     public static <T> Observable<T> take(final Observable<T> items, final int num) {
-        return create(OperationTake.take(items, num));
+        return _create(OperationTake.take(items, num));
     }
 
     /**
@@ -1345,7 +1352,7 @@ public static <T> Observable<T> take(final Observable<T> items, final int num) {
      *         items emitted by the source Observable
      */
     public static <T> Observable<List<T>> toList(final Observable<T> that) {
-        return create(OperationToObservableList.toObservableList(that));
+        return _create(OperationToObservableList.toObservableList(that));
     }
 
     /**
@@ -1364,7 +1371,7 @@ public static <T> Observable<List<T>> toList(final Observable<T> that) {
      * @return an Observable that emits each item in the source Iterable sequence
      */
     public static <T> Observable<T> toObservable(Iterable<T> iterable) {
-        return create(OperationToObservableIterable.toObservableIterable(iterable));
+        return _create(OperationToObservableIterable.toObservableIterable(iterable));
     }
 
     /**
@@ -1382,7 +1389,7 @@ public static <T> Observable<T> toObservable(Iterable<T> iterable) {
      * @return an Observable that emits the item from the source Future
      */
     public static <T> Observable<T> toObservable(Future<T> future) {
-        return create(OperationToObservableFuture.toObservableFuture(future));
+        return _create(OperationToObservableFuture.toObservableFuture(future));
     }
 
     /**
@@ -1405,7 +1412,7 @@ public static <T> Observable<T> toObservable(Future<T> future) {
      * @return an Observable that emits the item from the source Future
      */
     public static <T> Observable<T> toObservable(Future<T> future, long time, TimeUnit unit) {
-        return create(OperationToObservableFuture.toObservableFuture(future, time, unit));
+        return _create(OperationToObservableFuture.toObservableFuture(future, time, unit));
     }
 
     /**
@@ -1438,7 +1445,7 @@ public static <T> Observable<T> toObservable(T... items) {
      * @return
      */
     public static <T> Observable<List<T>> toSortedList(Observable<T> sequence) {
-        return create(OperationToObservableSortedList.toSortedList(sequence));
+        return _create(OperationToObservableSortedList.toSortedList(sequence));
     }
 
     /**
@@ -1451,7 +1458,7 @@ public static <T> Observable<List<T>> toSortedList(Observable<T> sequence) {
      * @return
      */
     public static <T> Observable<List<T>> toSortedList(Observable<T> sequence, Func2<T, T, Integer> sortFunction) {
-        return create(OperationToObservableSortedList.toSortedList(sequence, sortFunction));
+        return _create(OperationToObservableSortedList.toSortedList(sequence, sortFunction));
     }
 
     /**
@@ -1466,7 +1473,7 @@ public static <T> Observable<List<T>> toSortedList(Observable<T> sequence, Func2
     public static <T> Observable<List<T>> toSortedList(Observable<T> sequence, final Object sortFunction) {
         @SuppressWarnings("rawtypes")
         final FuncN _f = Functions.from(sortFunction);
-        return create(OperationToObservableSortedList.toSortedList(sequence, new Func2<T, T, Integer>() {
+        return _create(OperationToObservableSortedList.toSortedList(sequence, new Func2<T, T, Integer>() {
 
             @Override
             public Integer call(T t1, T t2) {
@@ -1501,7 +1508,7 @@ public Integer call(T t1, T t2) {
      * @return an Observable that emits the zipped results
      */
     public static <R, T0, T1> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Func2<T0, T1, R> reduceFunction) {
-        return create(OperationZip.zip(w0, w1, reduceFunction));
+        return _create(OperationZip.zip(w0, w1, reduceFunction));
     }
 
     /**
@@ -1571,7 +1578,7 @@ public R call(T0 t0, T1 t1) {
      * @return an Observable that emits the zipped results
      */
     public static <R, T0, T1, T2> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Observable<T2> w2, Func3<T0, T1, T2, R> function) {
-        return create(OperationZip.zip(w0, w1, w2, function));
+        return _create(OperationZip.zip(w0, w1, w2, function));
     }
 
     /**
@@ -1646,7 +1653,7 @@ public R call(T0 t0, T1 t1, T2 t2) {
      * @return an Observable that emits the zipped results
      */
     public static <R, T0, T1, T2, T3> Observable<R> zip(Observable<T0> w0, Observable<T1> w1, Observable<T2> w2, Observable<T3> w3, Func4<T0, T1, T2, T3, R> reduceFunction) {
-        return create(OperationZip.zip(w0, w1, w2, w3, reduceFunction));
+        return _create(OperationZip.zip(w0, w1, w2, w3, reduceFunction));
     }
 
     /**