1.x: distinctUntilChanged with direct value comparator - alternative (#4034)

* 1.x: distinctUntilChanged with direct value comparator - alternative

* Use primitive boolean

* Add experimental annotation

Author: akarnokd

src/main/java/rx/Observable.java

@@ -4804,6 +4804,29 @@ public final <U> Observable<T> distinctUntilChanged(Func1<? super T, ? extends U
         return lift(new OperatorDistinctUntilChanged<T, U>(keySelector));
     }
 
+    /**
+     * Returns an Observable that emits all items emitted by the source Observable that are distinct from their
+     * immediate predecessors when compared with each other via the provided comparator function.
+     * <p>
+     * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/distinctUntilChanged.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code distinctUntilChanged} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param comparator the function that receives the previous item and the current item and is
+     *                   expected to return true if the two are equal, thus skipping the current value.
+     * @return an Observable that emits those items from the source Observable that are distinct from their
+     *         immediate predecessors
+     * @see <a href="http://reactivex.io/documentation/operators/distinct.html">ReactiveX operators documentation: Distinct</a>
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical
+     *        with the release number)
+     */
+    @Experimental
+    public final Observable<T> distinctUntilChanged(Func2<? super T, ? super T, Boolean> comparator) {
+        return lift(new OperatorDistinctUntilChanged<T, T>(comparator));
+    }
+
     /**
      * Modifies the source Observable so that it invokes an action when it calls {@code onCompleted}.
      * <p>

src/main/java/rx/internal/operators/OperatorDistinctUntilChanged.java

@@ -18,17 +18,19 @@
 import rx.Observable.Operator;
 import rx.Subscriber;
 import rx.exceptions.Exceptions;
-import rx.functions.Func1;
+import rx.functions.*;
 import rx.internal.util.UtilityFunctions;
 
 /**
  * Returns an Observable that emits all sequentially distinct items emitted by the source.
  * @param <T> the value type
  * @param <U> the key type
  */
-public final class OperatorDistinctUntilChanged<T, U> implements Operator<T, T> {
+public final class OperatorDistinctUntilChanged<T, U> implements Operator<T, T>, Func2<U, U, Boolean> {
     final Func1<? super T, ? extends U> keySelector;
 
+    final Func2<? super U, ? super U, Boolean> comparator;
+    
     private static class Holder {
         static final OperatorDistinctUntilChanged<?,?> INSTANCE = new OperatorDistinctUntilChanged<Object,Object>(UtilityFunctions.identity());
     }
@@ -48,6 +50,19 @@ public static <T> OperatorDistinctUntilChanged<T, T> instance() {
 
     public OperatorDistinctUntilChanged(Func1<? super T, ? extends U> keySelector) {
         this.keySelector = keySelector;
+        this.comparator = this;
+        
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    public OperatorDistinctUntilChanged(Func2<? super U, ? super U, Boolean> comparator) {
+        this.keySelector = (Func1)UtilityFunctions.identity();
+        this.comparator = comparator;
+    }
+    
+    @Override
+    public Boolean call(U t1, U t2) {
+        return (t1 == t2 || (t1 != null && t1.equals(t2)));
     }
 
     @Override
@@ -68,7 +83,16 @@ public void onNext(T t) {
                 previousKey = key;
 
                 if (hasPrevious) {
-                    if (!(currentKey == key || (key != null && key.equals(currentKey)))) {
+                    boolean comparison;
+                    
+                    try {
+                        comparison = comparator.call(currentKey, key);
+                    } catch (Throwable e) {
+                        Exceptions.throwOrReport(e, child, key);
+                        return;
+                    }
+                    
+                    if (!comparison) {
                         child.onNext(t);
                     } else {
                         request(1);

src/test/java/rx/internal/operators/OperatorDistinctUntilChangedTest.java

@@ -16,25 +16,19 @@
 package rx.internal.operators;
 
 import static org.junit.Assert.assertFalse;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
-import static org.mockito.Mockito.inOrder;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
 import static org.mockito.MockitoAnnotations.initMocks;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.InOrder;
-import org.mockito.Mock;
+import org.junit.*;
+import org.mockito.*;
 
-import rx.Observable;
-import rx.Observer;
-import rx.functions.Action1;
-import rx.functions.Func1;
+import rx.*;
+import rx.exceptions.TestException;
+import rx.functions.*;
+import rx.observers.TestSubscriber;
 
 public class OperatorDistinctUntilChangedTest {
 
@@ -164,4 +158,43 @@ public void call(Throwable t) {
           .subscribe(w);
         assertFalse(errorOccurred.get());
     }
+    
+    @Test
+    public void customComparator() {
+        Observable<String> source = Observable.just("a", "b", "B", "A","a", "C");
+        
+        TestSubscriber<String> ts = TestSubscriber.create();
+        
+        source.distinctUntilChanged(new Func2<String, String, Boolean>() {
+            @Override
+            public Boolean call(String a, String b) {
+                return a.compareToIgnoreCase(b) == 0;
+            }
+        })
+        .subscribe(ts);
+        
+        ts.assertValues("a", "b", "A", "C");
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
+
+    @Test
+    public void customComparatorThrows() {
+        Observable<String> source = Observable.just("a", "b", "B", "A","a", "C");
+        
+        TestSubscriber<String> ts = TestSubscriber.create();
+        
+        source.distinctUntilChanged(new Func2<String, String, Boolean>() {
+            @Override
+            public Boolean call(String a, String b) {
+                throw new TestException();
+            }
+        })
+        .subscribe(ts);
+        
+        ts.assertValue("a");
+        ts.assertNotCompleted();
+        ts.assertError(TestException.class);
+    }
+
 }