added both new variants to observable

Author: Joachim Hofer

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

@@ -19,6 +19,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -2948,6 +2949,35 @@ public <U> Observable<T> distinctUntilChanged(Func1<? super T, ? extends U> keyS
         return create(OperationDistinctUntilChanged.distinctUntilChanged(this, keySelector));
     }
 
+    /**
+     * Returns an Observable that forwards all items emitted from the source Observable that are sequentially distinct according to
+     * a comparator.
+     * 
+     * @param equalityComparator
+     *            a comparator for deciding whether two emitted items are equal or not
+     * @return an Observable of sequentially distinct items
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229508%28v=vs.103%29.aspx">MSDN: Observable.distinctUntilChanged</a>
+     */
+    public <U> Observable<T> distinctUntilChanged(Comparator<T> equalityComparator) {
+        return create(OperationDistinctUntilChanged.distinctUntilChanged(this, equalityComparator));
+    }
+
+    /**
+     * Returns an Observable that forwards all items emitted from the source Observable that are sequentially distinct according to
+     * a key selector function and a comparator.
+     * 
+     * @param keySelector
+     *            a function that projects an emitted item to a key value which is used for deciding whether an item is sequentially
+     *            distinct from another one or not
+     * @param equalityComparator
+     *            a comparator for deciding whether two emitted item keys are equal or not
+     * @return an Observable of sequentially distinct items
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229508%28v=vs.103%29.aspx">MSDN: Observable.distinctUntilChanged</a>
+     */
+    public <U> Observable<T> distinctUntilChanged(Func1<? super T, ? extends U> keySelector, Comparator<U> equalityComparator) {
+        return create(OperationDistinctUntilChanged.distinctUntilChanged(this, keySelector, equalityComparator));
+    }
+
     /**
      * Returns an Observable that forwards all distinct items emitted from the source Observable.
      * 

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

@@ -43,6 +43,14 @@
  */
 public final class OperationDistinctUntilChanged {
 
+    /**
+     * Returns an Observable that emits all sequentially distinct items emitted by the source.
+     * @param source
+     *            The source Observable to emit the sequentially distinct items for.
+     * @param equalityComparator
+     *            The comparator to use for deciding whether to consider two items as equal or not.
+     * @return A subscription function for creating the target Observable.
+     */
     public static <T> OnSubscribeFunc<T> distinctUntilChanged(Observable<? extends T> source, Comparator<T> equalityComparator) {
         return new DistinctUntilChanged<T, T>(source, Functions.<T>identity(), equalityComparator);
     }
@@ -51,6 +59,22 @@ public static <T> OnSubscribeFunc<T> distinctUntilChanged(Observable<? extends T
      * Returns an Observable that emits all sequentially distinct items emitted by the source.
      * @param source
      *            The source Observable to emit the sequentially distinct items for.
+     * @param keySelector
+     *            The function to select the key to use for the equality checks.
+     * @param equalityComparator
+     *            The comparator to use for deciding whether to consider the two item keys as equal or not.
+     * @return A subscription function for creating the target Observable.
+     */
+    public static <T, U> OnSubscribeFunc<T> distinctUntilChanged(Observable<? extends T> source, Func1<? super T, ? extends U> keySelector, Comparator<U> equalityComparator) {
+        return new DistinctUntilChanged<T, U>(source, keySelector, equalityComparator);
+    }
+    
+    /**
+     * Returns an Observable that emits all sequentially distinct items emitted by the source.
+     * @param source
+     *            The source Observable to emit the sequentially distinct items for.
+     * @param keySelector
+     *            The function to select the key to use for the equality checks.
      * @return A subscription function for creating the target Observable.
      */
     public static <T, U> OnSubscribeFunc<T> distinctUntilChanged(Observable<? extends T> source, Func1<? super T, ? extends U> keySelector) {
@@ -142,6 +166,9 @@ public static class UnitTest {
         final Func1<String, String> TO_UPPER_WITH_EXCEPTION = new Func1<String, String>() {
             @Override
             public String call(String s) {
+                if (s.equals("x")) {
+                    return "xx";
+                }
                 return s.toUpperCase();
             }
         };
@@ -254,5 +281,19 @@ public void testDistinctUntilChangedWithComparator() {
             inOrder.verify(w, never()).onNext(anyString());
             verify(w, never()).onError(any(Throwable.class));
         }
+
+        @Test
+        public void testDistinctUntilChangedWithComparatorAndKeySelector() {
+            Observable<String> src = from("a", "b", "x", "aa", "bb", "c", "ddd");
+            create(distinctUntilChanged(src, TO_UPPER_WITH_EXCEPTION, COMPARE_LENGTH)).subscribe(w);
+            InOrder inOrder = inOrder(w); 
+            inOrder.verify(w, times(1)).onNext("a");
+            inOrder.verify(w, times(1)).onNext("x");
+            inOrder.verify(w, times(1)).onNext("c");
+            inOrder.verify(w, times(1)).onNext("ddd");
+            inOrder.verify(w, times(1)).onCompleted();
+            inOrder.verify(w, never()).onNext(anyString());
+            verify(w, never()).onError(any(Throwable.class));
+        }
     }
 }