fix GroupBy and GroupByUntil which are actually Operators

davidmoten · davidmoten · commit 007ce5508e36 · 2014-05-30T17:28:13.000+10:00
diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
@@ -4053,7 +4053,7 @@ public final void forEach(final Action1<? super T> onNext, final Action1<Throwab
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Transforming-Observables#wiki-groupby-and-groupbyuntil">RxJava Wiki: groupBy</a>
      */
     public final <K> Observable<GroupedObservable<K, T>> groupBy(final Func1<? super T, ? extends K> keySelector) {
-        return lift(new OnSubscribeGroupBy<K, T>(keySelector));
+        return lift(new OperatorGroupBy<K, T>(keySelector));
     }
 
     /**
@@ -4106,7 +4106,7 @@ public final <TKey, TDuration> Observable<GroupedObservable<TKey, T>> groupByUnt
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229433.aspx">MSDN: Observable.GroupByUntil</a>
      */
     public final <TKey, TValue, TDuration> Observable<GroupedObservable<TKey, TValue>> groupByUntil(Func1<? super T, ? extends TKey> keySelector, Func1<? super T, ? extends TValue> valueSelector, Func1<? super GroupedObservable<TKey, TValue>, ? extends Observable<? extends TDuration>> durationSelector) {
-        return lift(new OnSubscribeGroupByUntil<T, TKey, TValue, TDuration>(keySelector, valueSelector, durationSelector));
+        return lift(new OperatorGroupByUntil<T, TKey, TValue, TDuration>(keySelector, valueSelector, durationSelector));
     }
 
     /**
diff --git a/rxjava-core/src/main/java/rx/operators/OperatorGroupBy.java b/rxjava-core/src/main/java/rx/operators/OperatorGroupBy.java
@@ -37,11 +37,11 @@
  * @param <K> the key type
  * @param <T> the source and group value type
  */
-public final class OnSubscribeGroupBy<K, T> implements Operator<GroupedObservable<K, T>, T> {
+public final class OperatorGroupBy<K, T> implements Operator<GroupedObservable<K, T>, T> {
     
     final Func1<? super T, ? extends K> keySelector;
     
-    public OnSubscribeGroupBy(final Func1<? super T, ? extends K> keySelector) {
+    public OperatorGroupBy(final Func1<? super T, ? extends K> keySelector) {
         this.keySelector = keySelector;
     }
     
diff --git a/rxjava-core/src/main/java/rx/operators/OperatorGroupByUntil.java b/rxjava-core/src/main/java/rx/operators/OperatorGroupByUntil.java
@@ -44,12 +44,12 @@
  * @param <R> the value type of the groups
  * @param <D> the type of the duration
  */
-public class OnSubscribeGroupByUntil<T, K, R, D> implements Operator<GroupedObservable<K, R>, T> {
+public class OperatorGroupByUntil<T, K, R, D> implements Operator<GroupedObservable<K, R>, T> {
     final Func1<? super T, ? extends K> keySelector;
     final Func1<? super T, ? extends R> valueSelector;
     final Func1<? super GroupedObservable<K, R>, ? extends Observable<? extends D>> durationSelector;
 
-    public OnSubscribeGroupByUntil(
+    public OperatorGroupByUntil(
             Func1<? super T, ? extends K> keySelector,
             Func1<? super T, ? extends R> valueSelector,
             Func1<? super GroupedObservable<K, R>, ? extends Observable<? extends D>> durationSelector) {
diff --git a/rxjava-core/src/main/java/rx/operators/OperatorParallel.java b/rxjava-core/src/main/java/rx/operators/OperatorParallel.java
@@ -41,7 +41,7 @@ public OperatorParallel(Func1<Observable<T>, Observable<R>> f, Scheduler schedul
     public Subscriber<? super T> call(Subscriber<? super R> op) {
 
         Func1<Subscriber<? super GroupedObservable<Long, T>>, Subscriber<? super T>> groupBy =
-                new OnSubscribeGroupBy<Long, T>(new Func1<T, Long>() {
+                new OperatorGroupBy<Long, T>(new Func1<T, Long>() {
 
                     long i = 0;
 
diff --git a/rxjava-core/src/test/java/rx/operators/OperatorGroupByTest.java b/rxjava-core/src/test/java/rx/operators/OperatorGroupByTest.java
@@ -50,7 +50,7 @@
 import rx.observables.GroupedObservable;
 import rx.schedulers.Schedulers;
 
-public class OnSubscribeGroupByTest {
+public class OperatorGroupByTest {
 
     final Func1<String, Integer> length = new Func1<String, Integer>() {
         @Override
@@ -62,7 +62,7 @@ public Integer call(String s) {
     @Test
     public void testGroupBy() {
         Observable<String> source = Observable.from("one", "two", "three", "four", "five", "six");
-        Observable<GroupedObservable<Integer, String>> grouped = source.lift(new OnSubscribeGroupBy<Integer, String>(length));
+        Observable<GroupedObservable<Integer, String>> grouped = source.lift(new OperatorGroupBy<Integer, String>(length));
 
         Map<Integer, Collection<String>> map = toMap(grouped);
 
@@ -75,7 +75,7 @@ public void testGroupBy() {
     @Test
     public void testEmpty() {
         Observable<String> source = Observable.empty();
-        Observable<GroupedObservable<Integer, String>> grouped = source.lift(new OnSubscribeGroupBy<Integer, String>(length));
+        Observable<GroupedObservable<Integer, String>> grouped = source.lift(new OperatorGroupBy<Integer, String>(length));
 
         Map<Integer, Collection<String>> map = toMap(grouped);
 
@@ -88,7 +88,7 @@ public void testError() {
         Observable<String> errorSource = Observable.error(new RuntimeException("forced failure"));
         Observable<String> source = Observable.concat(sourceStrings, errorSource);
 
-        Observable<GroupedObservable<Integer, String>> grouped = source.lift(new OnSubscribeGroupBy<Integer, String>(length));
+        Observable<GroupedObservable<Integer, String>> grouped = source.lift(new OperatorGroupBy<Integer, String>(length));
 
         final AtomicInteger groupCounter = new AtomicInteger();
         final AtomicInteger eventCounter = new AtomicInteger();
diff --git a/rxjava-core/src/test/java/rx/operators/OperatorGroupByUntilTest.java b/rxjava-core/src/test/java/rx/operators/OperatorGroupByUntilTest.java
@@ -45,7 +45,7 @@
 import rx.functions.Functions;
 import rx.observables.GroupedObservable;
 
-public class OnSubscribeGroupByUntilTest {
+public class OperatorGroupByUntilTest {
     @Mock
     Observer<Object> observer;
 
