Refactoring for consistent implementation approach.

Combined ObservableExtensions into Observable to match how Rx works (Observable.* methods)

Author: benjchristensen

rxjava-core/src/main/java/org/rx/operations/AtomicWatchableSubscription.java renamed to rxjava-core/src/main/java/org/rx/operations/AtomicObservableSubscription.java

@@ -11,7 +11,7 @@
  * Thread-safe wrapper around ObservableSubscription that ensures unsubscribe can be called only once.
  */
 @ThreadSafe
-/* package */class AtomicObservableSubscription implements Subscription {
+/* package */final class AtomicObservableSubscription implements Subscription {
 
     private AtomicReference<Subscription> actualSubscription = new AtomicReference<Subscription>();
     private AtomicBoolean unsubscribed = new AtomicBoolean(false);

rxjava-core/src/main/java/org/rx/operations/AtomicWatcher.java renamed to rxjava-core/src/main/java/org/rx/operations/AtomicObserver.java

@@ -29,7 +29,7 @@
  * @param <T>
  */
 @ThreadSafe
-/* package */class AtomicObserver<T> implements Observer<T> {
+/* package */final class AtomicObserver<T> implements Observer<T> {
 
     /** Allow changing between forcing single or allowing multi-threaded execution of onNext */
     private static boolean allowMultiThreaded = true;

rxjava-core/src/main/java/org/rx/operations/AtomicWatcherMultiThreaded.java renamed to rxjava-core/src/main/java/org/rx/operations/AtomicObserverMultiThreaded.java

@@ -37,7 +37,7 @@
  * @param <T>
  */
 @ThreadSafe
-/* package */class AtomicObserverMultiThreaded<T> implements Observer<T> {
+/* package */final class AtomicObserverMultiThreaded<T> implements Observer<T> {
 
     private final Observer<T> Observer;
     private final AtomicObservableSubscription subscription;

rxjava-core/src/main/java/org/rx/operations/AtomicWatcherSingleThreaded.java renamed to rxjava-core/src/main/java/org/rx/operations/AtomicObserverSingleThreaded.java

@@ -34,7 +34,7 @@
  * @param <T>
  */
 @ThreadSafe
-/* package */class AtomicObserverSingleThreaded<T> implements Observer<T> {
+/* package */final class AtomicObserverSingleThreaded<T> implements Observer<T> {
 
     /**
      * Intrinsic synchronized locking with double-check short-circuiting was chosen after testing several other implementations.

rxjava-core/src/main/java/org/rx/operations/ObservableExtensions.java

@@ -22,7 +22,7 @@
 import org.rx.reactive.Observer;
 import org.rx.reactive.Subscription;
 
-class OperationCombineLatest {
+public class OperationCombineLatest {
 
     public static <R, T0, T1> Observable<R> combineLatest(Observable<T0> w0, Observable<T1> w1, Func2<R, T0, T1> combineLatestFunction) {
         Aggregator<R> a = new Aggregator<R>(Functions.fromFunc(combineLatestFunction));
@@ -652,7 +652,7 @@ public void testCombineLatest2Types() {
             /* define a Observer to receive aggregated events */
             Observer<String> aObserver = mock(Observer.class);
 
-            Observable<String> w = combineLatest(ObservableExtensions.toObservable("one", "two"), ObservableExtensions.toObservable(2, 3, 4), combineLatestFunction);
+            Observable<String> w = combineLatest(Observable.toObservable("one", "two"), Observable.toObservable(2, 3, 4), combineLatestFunction);
             w.subscribe(aObserver);
 
             verify(aObserver, never()).onError(any(Exception.class));
@@ -671,7 +671,7 @@ public void testCombineLatest3TypesA() {
             /* define a Observer to receive aggregated events */
             Observer<String> aObserver = mock(Observer.class);
 
-            Observable<String> w = combineLatest(ObservableExtensions.toObservable("one", "two"), ObservableExtensions.toObservable(2), ObservableExtensions.toObservable(new int[] { 4, 5, 6 }), combineLatestFunction);
+            Observable<String> w = combineLatest(Observable.toObservable("one", "two"), Observable.toObservable(2), Observable.toObservable(new int[] { 4, 5, 6 }), combineLatestFunction);
             w.subscribe(aObserver);
 
             verify(aObserver, never()).onError(any(Exception.class));
@@ -688,7 +688,7 @@ public void testCombineLatest3TypesB() {
             /* define a Observer to receive aggregated events */
             Observer<String> aObserver = mock(Observer.class);
 
-            Observable<String> w = combineLatest(ObservableExtensions.toObservable("one"), ObservableExtensions.toObservable(2), ObservableExtensions.toObservable(new int[] { 4, 5, 6 }, new int[] { 7, 8 }), combineLatestFunction);
+            Observable<String> w = combineLatest(Observable.toObservable("one"), Observable.toObservable(2), Observable.toObservable(new int[] { 4, 5, 6 }, new int[] { 7, 8 }), combineLatestFunction);
             w.subscribe(aObserver);
 
             verify(aObserver, never()).onError(any(Exception.class));
@@ -781,7 +781,7 @@ private static class TestObservable extends Observable<String> {
             public Subscription subscribe(Observer<String> Observer) {
                 // just store the variable where it can be accessed so we can manually trigger it
                 this.Observer = Observer;
-                return ObservableExtensions.noOpSubscription();
+                return Observable.noOpSubscription();
             }
 
         }

rxjava-core/src/main/java/org/rx/operations/OperationCombineLatest.java

@@ -4,54 +4,63 @@
 import static org.mockito.Mockito.*;
 
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.rx.functions.Func1;
 import org.rx.reactive.Observable;
 import org.rx.reactive.Observer;
 import org.rx.reactive.Subscription;
 
-/* package */final class OperationFilter<T> extends Observable<T> {
-    private final Observable<T> that;
-    private final Func1<Boolean, T> predicate;
+public final class OperationFilter<T> {
 
-    OperationFilter(Observable<T> that, Func1<Boolean, T> predicate) {
-        this.that = that;
-        this.predicate = predicate;
+    public static <T> Observable<T> filter(Observable<T> that, Func1<Boolean, T> predicate) {
+        return new Filter<T>(that, predicate);
     }
 
-    public Subscription subscribe(Observer<T> Observer) {
-        final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
-        final Observer<T> observer = new AtomicObserver<T>(Observer, subscription);
+    private static class Filter<T> extends Observable<T> {
 
-        subscription.setActual(that.subscribe(new Observer<T>() {
-            public void onNext(T value) {
-                try {
-                    if ((boolean) predicate.call(value)) {
-                        observer.onNext(value);
+        private final Observable<T> that;
+        private final Func1<Boolean, T> predicate;
+
+        public Filter(Observable<T> that, Func1<Boolean, T> predicate) {
+            this.that = that;
+            this.predicate = predicate;
+        }
+
+        public Subscription subscribe(Observer<T> Observer) {
+            final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
+            final Observer<T> observer = new AtomicObserver<T>(Observer, subscription);
+
+            subscription.setActual(that.subscribe(new Observer<T>() {
+                public void onNext(T value) {
+                    try {
+                        if ((boolean) predicate.call(value)) {
+                            observer.onNext(value);
+                        }
+                    } catch (Exception ex) {
+                        observer.onError(ex);
+                        subscription.unsubscribe();
                     }
-                } catch (Exception ex) {
-                    observer.onError(ex);
-                    subscription.unsubscribe();
                 }
-            }
 
-            public void onError(Exception ex) {
-                observer.onError(ex);
-            }
+                public void onError(Exception ex) {
+                    observer.onError(ex);
+                }
 
-            public void onCompleted() {
-                observer.onCompleted();
-            }
-        }));
+                public void onCompleted() {
+                    observer.onCompleted();
+                }
+            }));
 
-        return subscription;
+            return subscription;
+        }
     }
 
     public static class UnitTest {
 
         @Test
         public void testFilter() {
-            Observable<String> w = ObservableExtensions.toObservable("one", "two", "three");
-            Observable<String> Observable = new OperationFilter<String>(w, new Func1<Boolean, String>() {
+            Observable<String> w = Observable.toObservable("one", "two", "three");
+            Observable<String> Observable = filter(w, new Func1<Boolean, String>() {
 
                 @Override
                 public Boolean call(String t1) {
@@ -65,10 +74,10 @@ public Boolean call(String t1) {
             @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             Observable.subscribe(aObserver);
-            verify(aObserver, never()).onNext("one");
+            verify(aObserver, Mockito.never()).onNext("one");
             verify(aObserver, times(1)).onNext("two");
-            verify(aObserver, never()).onNext("three");
-            verify(aObserver, never()).onError(any(Exception.class));
+            verify(aObserver, Mockito.never()).onNext("three");
+            verify(aObserver, Mockito.never()).onError(any(Exception.class));
             verify(aObserver, times(1)).onCompleted();
         }
     }

rxjava-core/src/main/java/org/rx/operations/OperationFilter.java

@@ -7,6 +7,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
+import org.mockito.Mockito;
 import org.rx.reactive.Observable;
 import org.rx.reactive.Observer;
 import org.rx.reactive.Subscription;
@@ -16,54 +17,62 @@
  * 
  * @param <T>
  */
-/* package */final class OperationLast<T> extends Observable<T> {
-    private final AtomicReference<T> lastValue = new AtomicReference<T>();
-    private final Observable<T> that;
-    private final AtomicBoolean onNextCalled = new AtomicBoolean(false);
+public final class OperationLast<T> {
 
-    OperationLast(Observable<T> that) {
-        this.that = that;
+    public static <T> Observable<T> last(Observable<T> observable) {
+        return new Last<T>(observable);
     }
 
-    public Subscription subscribe(final Observer<T> Observer) {
-        final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
-        final Observer<T> observer = new AtomicObserver<T>(Observer, subscription);
+    private static class Last<T> extends Observable<T> {
 
-        subscription.setActual(that.subscribe(new Observer<T>() {
-            public void onNext(T value) {
-                onNextCalled.set(true);
-                lastValue.set(value);
-            }
+        private final AtomicReference<T> lastValue = new AtomicReference<T>();
+        private final Observable<T> that;
+        private final AtomicBoolean onNextCalled = new AtomicBoolean(false);
 
-            public void onError(Exception ex) {
-                observer.onError(ex);
-            }
+        public Last(Observable<T> that) {
+            this.that = that;
+        }
+
+        public Subscription subscribe(final Observer<T> Observer) {
+            final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
+            final Observer<T> observer = new AtomicObserver<T>(Observer, subscription);
+
+            subscription.setActual(that.subscribe(new Observer<T>() {
+                public void onNext(T value) {
+                    onNextCalled.set(true);
+                    lastValue.set(value);
+                }
 
-            public void onCompleted() {
-                if (onNextCalled.get()) {
-                    observer.onNext(lastValue.get());
+                public void onError(Exception ex) {
+                    observer.onError(ex);
                 }
-                observer.onCompleted();
-            }
-        }));
 
-        return subscription;
+                public void onCompleted() {
+                    if (onNextCalled.get()) {
+                        observer.onNext(lastValue.get());
+                    }
+                    observer.onCompleted();
+                }
+            }));
+
+            return subscription;
+        }
     }
 
     public static class UnitTest {
 
         @Test
         public void testLast() {
-            Observable<String> w = ObservableExtensions.toObservable("one", "two", "three");
-            Observable<String> Observable = new OperationLast<String>(w);
+            Observable<String> w = Observable.toObservable("one", "two", "three");
+            Observable<String> Observable = last(w);
 
             @SuppressWarnings("unchecked")
             Observer<String> aObserver = mock(Observer.class);
             Observable.subscribe(aObserver);
-            verify(aObserver, never()).onNext("one");
-            verify(aObserver, never()).onNext("two");
+            verify(aObserver, Mockito.never()).onNext("one");
+            verify(aObserver, Mockito.never()).onNext("two");
             verify(aObserver, times(1)).onNext("three");
-            verify(aObserver, never()).onError(any(Exception.class));
+            verify(aObserver, Mockito.never()).onError(any(Exception.class));
             verify(aObserver, times(1)).onCompleted();
         }
     }

rxjava-core/src/main/java/org/rx/operations/OperationLast.java

@@ -15,7 +15,7 @@
 import org.rx.reactive.Observer;
 import org.rx.reactive.Subscription;
 
-/* package */class OperationMap {
+public final class OperationMap {
 
     /**
      * Accepts a sequence and a transformation function. Returns a sequence that is the result of
@@ -130,7 +130,7 @@ public void testMap() {
             Map<String, String> m1 = getMap("One");
             Map<String, String> m2 = getMap("Two");
             @SuppressWarnings("unchecked")
-            Observable<Map<String, String>> observable = ObservableExtensions.toObservable(m1, m2);
+            Observable<Map<String, String>> observable = Observable.toObservable(m1, m2);
 
             Observable<String> m = map(observable, new Func1<String, Map<String, String>>() {
 
@@ -152,7 +152,7 @@ public String call(Map<String, String> map) {
         @Test
         public void testMapMany() {
             /* simulate a top-level async call which returns IDs */
-            Observable<Integer> ids = ObservableExtensions.toObservable(1, 2);
+            Observable<Integer> ids = Observable.toObservable(1, 2);
 
             /* now simulate the behavior to take those IDs and perform nested async calls based on them */
             Observable<String> m = mapMany(ids, new Func1<Observable<String>, Integer>() {
@@ -165,11 +165,11 @@ public Observable<String> call(Integer id) {
                     if (id == 1) {
                         Map<String, String> m1 = getMap("One");
                         Map<String, String> m2 = getMap("Two");
-                        subObservable = ObservableExtensions.toObservable(m1, m2);
+                        subObservable = Observable.toObservable(m1, m2);
                     } else {
                         Map<String, String> m3 = getMap("Three");
                         Map<String, String> m4 = getMap("Four");
-                        subObservable = ObservableExtensions.toObservable(m3, m4);
+                        subObservable = Observable.toObservable(m3, m4);
                     }
 
                     /* simulate kicking off the async call and performing a select on it to transform the data */
@@ -197,15 +197,15 @@ public void testMapMany2() {
             Map<String, String> m1 = getMap("One");
             Map<String, String> m2 = getMap("Two");
             @SuppressWarnings("unchecked")
-            Observable<Map<String, String>> observable1 = ObservableExtensions.toObservable(m1, m2);
+            Observable<Map<String, String>> observable1 = Observable.toObservable(m1, m2);
 
             Map<String, String> m3 = getMap("Three");
             Map<String, String> m4 = getMap("Four");
             @SuppressWarnings("unchecked")
-            Observable<Map<String, String>> observable2 = ObservableExtensions.toObservable(m3, m4);
+            Observable<Map<String, String>> observable2 = Observable.toObservable(m3, m4);
 
             @SuppressWarnings("unchecked")
-            Observable<Observable<Map<String, String>>> observable = ObservableExtensions.toObservable(observable1, observable2);
+            Observable<Observable<Map<String, String>>> observable = Observable.toObservable(observable1, observable2);
 
             Observable<String> m = mapMany(observable, new Func1<Observable<String>, Observable<Map<String, String>>>() {
 

rxjava-core/src/main/java/org/rx/operations/OperationMap.java

@@ -18,7 +18,7 @@
  * <p>
  * See http://msdn.microsoft.com/en-us/library/hh229453(v=VS.103).aspx for the Microsoft Rx equivalent.
  */
-public class OperationMaterialize {
+public final class OperationMaterialize {
 
     /**
      * Materializes the implicit notifications of an observable sequence as explicit notification values.

rxjava-core/src/main/java/org/rx/operations/OperationMaterialize.java

@@ -17,7 +17,7 @@
 import org.rx.reactive.Observer;
 import org.rx.reactive.Subscription;
 
-/* package */class OperationMerge {
+public final class OperationMerge {
 
     /**
      * Flattens the observable sequences from the list of Observables into one observable sequence without any transformation.

rxjava-core/src/main/java/org/rx/operations/OperationMerge.java

@@ -27,7 +27,7 @@
  * <p>
  * NOTE: If this is used on an infinite stream it will never call onError and effectively will swallow errors.
  */
-/* package */class OperationMergeDelayError {
+public final class OperationMergeDelayError {
 
     /**
      * Flattens the observable sequences from the list of Observables into one observable sequence without any transformation and delays any onError calls until after all sequences have called